From 31dc3d148b5680bf8bfb82023a487f17fb26a9c9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Carlos=20Flor=C3=AAncio?= Date: Fri, 27 Feb 2026 15:24:12 +0000 Subject: [PATCH] fix: lazy Docker client initialization to avoid startup errors (#300) --- apps/backend/cmd/kandev/agents.go | 15 +- apps/backend/cmd/kandev/helpers.go | 21 +- apps/backend/cmd/kandev/main.go | 22 +- .../backend/internal/agent/docker/handlers.go | 57 ++- .../agent/handlers/shell_handlers_test.go | 4 +- .../agent/lifecycle/executor_docker.go | 194 +++++----- .../agent/lifecycle/executor_docker_test.go | 345 ++++++++++++++++++ .../agent/lifecycle/executor_registry.go | 22 ++ .../internal/agent/lifecycle/manager.go | 37 +- .../agent/lifecycle/manager_events_test.go | 2 +- .../agent/lifecycle/manager_lifecycle.go | 26 +- .../internal/agent/lifecycle/manager_test.go | 4 +- .../task/handlers/process_handlers_test.go | 2 +- 13 files changed, 570 insertions(+), 181 deletions(-) create mode 100644 apps/backend/internal/agent/lifecycle/executor_docker_test.go diff --git a/apps/backend/cmd/kandev/agents.go b/apps/backend/cmd/kandev/agents.go index b80461fedd..ed705e015e 100644 --- a/apps/backend/cmd/kandev/agents.go +++ b/apps/backend/cmd/kandev/agents.go @@ -7,7 +7,6 @@ import ( "go.uber.org/zap" "github.com/kandev/kandev/internal/agent/credentials" - "github.com/kandev/kandev/internal/agent/docker" agentexecutor "github.com/kandev/kandev/internal/agent/executor" "github.com/kandev/kandev/internal/agent/lifecycle" "github.com/kandev/kandev/internal/agent/mcpconfig" @@ -26,7 +25,6 @@ func provideLifecycleManager( cfg *config.Config, log *logger.Logger, eventBus bus.EventBus, - dockerClient *docker.Client, agentSettingsRepo settingsstore.Repository, agentRegistry *registry.Registry, secretStore secrets.SecretStore, @@ -58,15 +56,11 @@ func provideLifecycleManager( 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) - dockerExec := lifecycle.NewDockerExecutor(dockerClient, log) + // Register Docker runtime if enabled (client is created lazily on first use) + if cfg.Docker.Enabled { + dockerExec := lifecycle.NewDockerExecutor(cfg.Docker, log) executorRegistry.Register(dockerExec) - log.Info("Docker runtime registered") - } else if cfg.Docker.Enabled && dockerClient == nil { - log.Debug("Docker runtime enabled but Docker client not available") + log.Info("Docker runtime registered (lazy initialization)") } // Register Remote Docker runtime (always available, instances are created lazily per host) @@ -97,7 +91,6 @@ func provideLifecycleManager( agentRegistry, eventBus, executorRegistry, - containerMgr, credsMgr, profileResolver, mcpService, diff --git a/apps/backend/cmd/kandev/helpers.go b/apps/backend/cmd/kandev/helpers.go index 8880f5cd11..5d4fa6d4ce 100644 --- a/apps/backend/cmd/kandev/helpers.go +++ b/apps/backend/cmd/kandev/helpers.go @@ -17,7 +17,6 @@ import ( analyticshandlers "github.com/kandev/kandev/internal/analytics/handlers" analyticsrepository "github.com/kandev/kandev/internal/analytics/repository" "github.com/kandev/kandev/internal/clarification" - "github.com/kandev/kandev/internal/common/config" "github.com/kandev/kandev/internal/common/logger" debughandlers "github.com/kandev/kandev/internal/debug" editorcontroller "github.com/kandev/kandev/internal/editors/controller" @@ -44,23 +43,6 @@ import ( ws "github.com/kandev/kandev/pkg/websocket" ) -// initDockerClient initializes the Docker client and verifies connectivity. -// Returns nil if Docker is unavailable or the connection check fails. -func initDockerClient(ctx context.Context, cfg *config.Config, log *logger.Logger) *docker.Client { - dockerClient, err := docker.NewClient(cfg.Docker, log) - if err != nil { - log.Debug("Failed to initialize Docker client - agent features will be disabled", zap.Error(err)) - return nil - } - if err := dockerClient.Ping(ctx); err != nil { - log.Debug("Docker daemon not available - agent features will be disabled", zap.Error(err)) - _ = dockerClient.Close() - return nil - } - log.Info("Connected to Docker daemon") - return dockerClient -} - // buildSessionDataProvider constructs the session data provider function used by the WebSocket hub // to send initial data (git status, context window, available commands) when a client subscribes. func buildSessionDataProvider(taskRepo *sqliterepo.Repository, lifecycleMgr *lifecycle.Manager, log *logger.Logger) func(context.Context, string) ([]*ws.Message, error) { @@ -265,7 +247,6 @@ func newGitHubTaskPRUpdatedHandler(gateway *gateways.Gateway, log *logger.Logger type routeParams struct { router *gin.Engine gateway *gateways.Gateway - dockerClient *docker.Client taskSvc *taskservice.Service taskRepo *sqliterepo.Repository analyticsRepo analyticsrepository.Repository @@ -363,7 +344,7 @@ func registerSecondaryRoutes( p.log.Debug("Registered GitHub handlers (HTTP + WebSocket)") } - docker.RegisterDockerRoutes(p.router, p.dockerClient, p.log) + docker.RegisterDockerRoutes(p.router, p.lifecycleMgr.DockerClientProvider(), p.log) p.log.Debug("Registered Docker management handlers (HTTP)") registerMCPAndDebugRoutes(p, workflowCtrl, clarificationStore, planService) diff --git a/apps/backend/cmd/kandev/main.go b/apps/backend/cmd/kandev/main.go index 5fe867ba19..fd57880196 100644 --- a/apps/backend/cmd/kandev/main.go +++ b/apps/backend/cmd/kandev/main.go @@ -28,7 +28,6 @@ import ( githubpkg "github.com/kandev/kandev/internal/github" // Agent infrastructure - "github.com/kandev/kandev/internal/agent/docker" "github.com/kandev/kandev/internal/agent/lifecycle" "github.com/kandev/kandev/internal/agent/registry" agentsettingscontroller "github.com/kandev/kandev/internal/agent/settings/controller" @@ -170,13 +169,7 @@ func run(cfg *config.Config, log *logger.Logger, cleanups *[]func() error, runCl addCleanup(cleanup) eventBus := eventBusProvider.Bus - // 5. Initialize Docker client - dockerClient := initDockerClient(ctx, cfg, log) - if dockerClient != nil { - defer func() { _ = dockerClient.Close() }() - } - - return startServices(ctx, cfg, log, addCleanup, eventBus, dockerClient, runCleanups) + return startServices(ctx, cfg, log, addCleanup, eventBus, runCleanups) } // startServices initializes task-level services and all downstream infrastructure. @@ -186,7 +179,6 @@ func startServices( //nolint:cyclop log *logger.Logger, addCleanup func(func() error), eventBus bus.EventBus, - dockerClient *docker.Client, runCleanups func(), ) bool { // ============================================ @@ -242,7 +234,7 @@ func startServices( //nolint:cyclop }() } - return startAgentInfrastructure(ctx, cfg, log, addCleanup, eventBus, dockerClient, dbPool, repos, services, agentSettingsController, agentRegistry, runCleanups) + return startAgentInfrastructure(ctx, cfg, log, addCleanup, eventBus, dbPool, repos, services, agentSettingsController, agentRegistry, runCleanups) } // startAgentInfrastructure initializes the agent lifecycle manager, worktree, orchestrator, @@ -253,7 +245,6 @@ func startAgentInfrastructure( log *logger.Logger, addCleanup func(func() error), eventBus bus.EventBus, - dockerClient *docker.Client, dbPool *db.Pool, repos *Repositories, services *Services, @@ -264,7 +255,7 @@ func startAgentInfrastructure( // ============================================ // AGENT MANAGER // ============================================ - lifecycleMgr, err := provideLifecycleManager(ctx, cfg, log, eventBus, dockerClient, repos.AgentSettings, agentRegistry, repos.Secrets) + lifecycleMgr, err := provideLifecycleManager(ctx, cfg, log, eventBus, repos.AgentSettings, agentRegistry, repos.Secrets) if err != nil { log.Error("Failed to initialize agent manager", zap.Error(err)) return false @@ -321,7 +312,7 @@ func startAgentInfrastructure( log.Info("GitHub poller started") } - return startGatewayAndServe(ctx, cfg, log, eventBus, dockerClient, repos, services, + return startGatewayAndServe(ctx, cfg, log, eventBus, repos, services, agentSettingsController, lifecycleMgr, agentRegistry, orchestratorSvc, msgCreator, runCleanups) } @@ -332,7 +323,6 @@ func startGatewayAndServe( cfg *config.Config, log *logger.Logger, eventBus bus.EventBus, - dockerClient *docker.Client, repos *Repositories, services *Services, agentSettingsController *agentsettingscontroller.Controller, @@ -374,7 +364,7 @@ func startGatewayAndServe( // ============================================ // HTTP SERVER // ============================================ - server := buildHTTPServer(cfg, log, gateway, dockerClient, repos, services, agentSettingsController, + server := buildHTTPServer(cfg, log, gateway, repos, services, agentSettingsController, lifecycleMgr, eventBus, orchestratorSvc, notificationCtrl, msgCreator, agentRegistry) port := cfg.Server.Port @@ -403,7 +393,6 @@ func buildHTTPServer( cfg *config.Config, log *logger.Logger, gateway *gateways.Gateway, - dockerClient *docker.Client, repos *Repositories, services *Services, agentSettingsController *agentsettingscontroller.Controller, @@ -424,7 +413,6 @@ func buildHTTPServer( registerRoutes(routeParams{ router: router, gateway: gateway, - dockerClient: dockerClient, taskSvc: services.Task, taskRepo: repos.Task, analyticsRepo: repos.Analytics, diff --git a/apps/backend/internal/agent/docker/handlers.go b/apps/backend/internal/agent/docker/handlers.go index 08d697ae3d..c13e4e638c 100644 --- a/apps/backend/internal/agent/docker/handlers.go +++ b/apps/backend/internal/agent/docker/handlers.go @@ -33,24 +33,38 @@ type stopContainerRequest struct { TimeoutSeconds int `json:"timeout_seconds,omitempty"` } -// RegisterDockerRoutes registers Docker management HTTP routes on the given router. -func RegisterDockerRoutes(router *gin.Engine, dockerClient *Client, log *logger.Logger) { - if dockerClient == nil { - log.Debug("Docker client is nil, skipping Docker route registration") - return - } +// ClientProvider lazily resolves the Docker client. Returns nil if Docker is unavailable. +type ClientProvider func() *Client +// RegisterDockerRoutes registers Docker management HTTP routes on the given router. +// The clientProvider lazily resolves the Docker client on each request. +func RegisterDockerRoutes(router *gin.Engine, clientProvider ClientProvider, log *logger.Logger) { api := router.Group("/api/v1/docker") - api.POST("/build", handleBuildImage(dockerClient, log)) - api.GET("/containers", handleListContainers(dockerClient, log)) - api.POST("/containers/:id/stop", handleStopContainer(dockerClient, log)) - api.DELETE("/containers/:id", handleRemoveContainer(dockerClient, log)) + api.POST("/build", handleBuildImage(clientProvider, log)) + api.GET("/containers", handleListContainers(clientProvider, log)) + api.POST("/containers/:id/stop", handleStopContainer(clientProvider, log)) + api.DELETE("/containers/:id", handleRemoveContainer(clientProvider, log)) +} + +// requireDocker resolves the Docker client and returns 503 if unavailable. +func requireDocker(c *gin.Context, clientProvider ClientProvider) *Client { + client := clientProvider() + if client == nil { + c.JSON(http.StatusServiceUnavailable, gin.H{"error": "Docker is not available"}) + return nil + } + return client } // handleBuildImage handles POST /api/v1/docker/build. // It streams the Docker build output as JSON lines to the client. -func handleBuildImage(dockerClient *Client, log *logger.Logger) gin.HandlerFunc { +func handleBuildImage(clientProvider ClientProvider, log *logger.Logger) gin.HandlerFunc { return func(c *gin.Context) { + dockerClient := requireDocker(c, clientProvider) + if dockerClient == nil { + return + } + var req buildImageRequest if err := c.ShouldBindJSON(&req); err != nil { c.JSON(http.StatusBadRequest, gin.H{"error": "invalid request: " + err.Error()}) @@ -99,8 +113,13 @@ func streamBuildOutput(c *gin.Context, reader interface{ Read([]byte) (int, erro // handleListContainers handles GET /api/v1/docker/containers. // Supports optional query params: image, labels (comma-separated key=value pairs). -func handleListContainers(dockerClient *Client, log *logger.Logger) gin.HandlerFunc { +func handleListContainers(clientProvider ClientProvider, log *logger.Logger) gin.HandlerFunc { return func(c *gin.Context) { + dockerClient := requireDocker(c, clientProvider) + if dockerClient == nil { + return + } + labels := parseLabelsQuery(c) addImageFilter(c, labels) @@ -174,8 +193,13 @@ func addImageFilter(c *gin.Context, labels map[string]string) { } // handleStopContainer handles POST /api/v1/docker/containers/:id/stop. -func handleStopContainer(dockerClient *Client, log *logger.Logger) gin.HandlerFunc { +func handleStopContainer(clientProvider ClientProvider, log *logger.Logger) gin.HandlerFunc { return func(c *gin.Context) { + dockerClient := requireDocker(c, clientProvider) + if dockerClient == nil { + return + } + containerID := c.Param("id") if containerID == "" { c.JSON(http.StatusBadRequest, gin.H{"error": "container id is required"}) @@ -202,8 +226,13 @@ func handleStopContainer(dockerClient *Client, log *logger.Logger) gin.HandlerFu } // handleRemoveContainer handles DELETE /api/v1/docker/containers/:id. -func handleRemoveContainer(dockerClient *Client, log *logger.Logger) gin.HandlerFunc { +func handleRemoveContainer(clientProvider ClientProvider, log *logger.Logger) gin.HandlerFunc { return func(c *gin.Context) { + dockerClient := requireDocker(c, clientProvider) + if dockerClient == nil { + return + } + containerID := c.Param("id") if containerID == "" { c.JSON(http.StatusBadRequest, gin.H{"error": "container id is required"}) diff --git a/apps/backend/internal/agent/handlers/shell_handlers_test.go b/apps/backend/internal/agent/handlers/shell_handlers_test.go index c4cc560936..77f6ea3362 100644 --- a/apps/backend/internal/agent/handlers/shell_handlers_test.go +++ b/apps/backend/internal/agent/handlers/shell_handlers_test.go @@ -183,8 +183,8 @@ func newTestManager() *lifecycle.Manager { eventBus := &MockEventBus{} 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, lifecycle.ExecutorFallbackWarn, log) + // Pass nil for runtime - tests don't need them + return lifecycle.NewManager(reg, eventBus, nil, credsMgr, profileResolver, nil, lifecycle.ExecutorFallbackWarn, log) } func TestWsShellStatus_NoInstanceFound(t *testing.T) { diff --git a/apps/backend/internal/agent/lifecycle/executor_docker.go b/apps/backend/internal/agent/lifecycle/executor_docker.go index 82e54c8275..d07626ab97 100644 --- a/apps/backend/internal/agent/lifecycle/executor_docker.go +++ b/apps/backend/internal/agent/lifecycle/executor_docker.go @@ -4,14 +4,15 @@ import ( "context" "fmt" "strings" + "sync" "time" "go.uber.org/zap" "github.com/kandev/kandev/internal/agent/docker" "github.com/kandev/kandev/internal/agent/executor" - agentctl "github.com/kandev/kandev/internal/agentctl/client" "github.com/kandev/kandev/internal/agentctl/server/process" + "github.com/kandev/kandev/internal/common/config" "github.com/kandev/kandev/internal/common/logger" ) @@ -41,35 +42,85 @@ func getMetadataBool(metadata map[string]interface{}, key string) bool { } // DockerExecutor implements Runtime for Docker-based agent execution. +// The Docker client is created lazily on first use (not at startup). type DockerExecutor struct { - containerMgr *ContainerManager + cfg config.DockerConfig + logger *logger.Logger + + // newClientFunc creates the Docker client. Defaults to docker.NewClient. + // Override in tests to simulate failures. + newClientFunc func(config.DockerConfig, *logger.Logger) (*docker.Client, error) + + // Lazy-initialized on first use via ensureClient(). + // Uses mu + initialized instead of sync.Once so that transient Docker + // daemon failures can be retried on subsequent calls. + mu sync.Mutex + initialized bool docker *docker.Client - logger *logger.Logger + containerMgr *ContainerManager } // NewDockerExecutor creates a new Docker runtime. -func NewDockerExecutor(dockerClient *docker.Client, log *logger.Logger) *DockerExecutor { +// The Docker client is NOT created here — it is initialized lazily +// when CreateInstance is called. +func NewDockerExecutor(cfg config.DockerConfig, log *logger.Logger) *DockerExecutor { return &DockerExecutor{ - containerMgr: NewContainerManager(dockerClient, "", log), - docker: dockerClient, - logger: log.WithFields(zap.String("runtime", "docker")), + cfg: cfg, + logger: log.WithFields(zap.String("runtime", "docker")), + newClientFunc: docker.NewClient, } } +// ensureClient lazily creates the Docker client and ContainerManager. +// Unlike sync.Once, this retries on failure so transient Docker daemon +// unavailability doesn't permanently disable the executor. +func (r *DockerExecutor) ensureClient() (*docker.Client, *ContainerManager, error) { + r.mu.Lock() + defer r.mu.Unlock() + + if r.initialized { + return r.docker, r.containerMgr, nil + } + + cli, err := r.newClientFunc(r.cfg, r.logger) + if err != nil { + return nil, nil, fmt.Errorf("failed to create docker client: %w", err) + } + + r.docker = cli + r.containerMgr = NewContainerManager(cli, "", r.logger) + r.initialized = true + + return r.docker, r.containerMgr, nil +} + +// Client returns the lazily-initialized Docker client, or nil if Docker is unavailable. +func (r *DockerExecutor) Client() *docker.Client { + cli, _, _ := r.ensureClient() + return cli +} + +// ContainerMgr returns the lazily-initialized ContainerManager, or nil if Docker is unavailable. +func (r *DockerExecutor) ContainerMgr() *ContainerManager { + _, cm, _ := r.ensureClient() + return cm +} + func (r *DockerExecutor) Name() executor.Name { return executor.NameDocker } -func (r *DockerExecutor) HealthCheck(ctx context.Context) error { - // Check Docker is reachable by listing containers - _, err := r.docker.ListContainers(ctx, map[string]string{}) - if err != nil { - return fmt.Errorf("docker not reachable: %w", err) - } +func (r *DockerExecutor) HealthCheck(_ context.Context) error { + // No-op: Docker availability is checked lazily when CreateInstance is called. return nil } func (r *DockerExecutor) CreateInstance(ctx context.Context, req *ExecutorCreateRequest) (*ExecutorInstance, error) { + dockerClient, containerMgr, err := r.ensureClient() + if err != nil { + return nil, fmt.Errorf("docker unavailable: %w", err) + } + // Extract runtime-specific values from metadata mainRepoGitDir := getMetadataString(req.Metadata, MetadataKeyMainRepoGitDir) worktreeID := getMetadataString(req.Metadata, MetadataKeyWorktreeID) @@ -88,13 +139,13 @@ func (r *DockerExecutor) CreateInstance(ctx context.Context, req *ExecutorCreate } // Use ContainerManager to launch container - containerID, client, err := r.containerMgr.LaunchContainer(ctx, containerCfg) + containerID, client, err := containerMgr.LaunchContainer(ctx, containerCfg) if err != nil { return nil, fmt.Errorf("failed to launch container: %w", err) } // Get container IP for logging - containerIP, _ := r.docker.GetContainerIP(ctx, containerID) + containerIP, _ := dockerClient.GetContainerIP(ctx, containerID) // Build metadata metadata := make(map[string]interface{}) @@ -127,11 +178,15 @@ func (r *DockerExecutor) StopInstance(ctx context.Context, instance *ExecutorIns return nil // No container to stop } - var err error + dockerClient, _, err := r.ensureClient() + if err != nil { + return fmt.Errorf("docker unavailable: %w", err) + } + if force { - err = r.docker.KillContainer(ctx, instance.ContainerID, "SIGKILL") + err = dockerClient.KillContainer(ctx, instance.ContainerID, "SIGKILL") } else { - err = r.docker.StopContainer(ctx, instance.ContainerID, 30*time.Second) + err = dockerClient.StopContainer(ctx, instance.ContainerID, 30*time.Second) } if err != nil { @@ -141,89 +196,28 @@ func (r *DockerExecutor) StopInstance(ctx context.Context, instance *ExecutorIns return nil } -func (r *DockerExecutor) RecoverInstances(ctx context.Context) ([]*ExecutorInstance, error) { - // Find containers with kandev.managed label - containers, err := r.docker.ListContainers(ctx, map[string]string{ - "kandev.managed": "true", - }) - if err != nil { - return nil, fmt.Errorf("failed to list containers: %w", err) +func (r *DockerExecutor) RecoverInstances(_ context.Context) ([]*ExecutorInstance, error) { + // No-op: Docker client is initialized lazily on first use. + // If no session has used Docker yet, there's nothing to recover. + // Running containers from a previous backend process will be detected + // when the user navigates to that session (via EnsureWorkspaceExecutionForSession). + return nil, nil +} + +// Close closes the Docker client if it was initialized. +// Safe to call even if the client was never created. +func (r *DockerExecutor) Close() error { + r.mu.Lock() + defer r.mu.Unlock() + + if r.docker != nil { + err := r.docker.Close() + r.docker = nil + r.containerMgr = nil + r.initialized = false + return err } - - var recovered []*ExecutorInstance - for _, ctr := range containers { - // Only recover running containers - if ctr.State != "running" { - r.logger.Debug("skipping non-running container", - zap.String("container_id", ctr.ID), - zap.String("state", ctr.State)) - continue - } - - // Get container labels to extract instance info - labels, err := r.docker.GetContainerLabels(ctx, ctr.ID) - if err != nil { - r.logger.Warn("failed to get container labels", - zap.String("container_id", ctr.ID), - zap.Error(err)) - continue - } - - instanceID := labels["kandev.instance_id"] - taskID := labels["kandev.task_id"] - sessionID := labels["kandev.session_id"] - agentProfileID := labels["kandev.agent_profile_id"] - - if instanceID == "" || taskID == "" { - r.logger.Warn("container missing required labels", - zap.String("container_id", ctr.ID)) - continue - } - - // Get container IP - containerIP, err := r.docker.GetContainerIP(ctx, ctr.ID) - if err != nil { - r.logger.Warn("failed to get container IP", - zap.String("container_id", ctr.ID), - zap.Error(err)) - containerIP = "127.0.0.1" - } - - // Query the control server to get the instance port - ctl := agentctl.NewControlClient(containerIP, AgentCtlPort, r.logger) - instanceInfo, err := ctl.GetInstance(ctx, instanceID) - if err != nil { - r.logger.Warn("failed to get instance info from container", - zap.String("container_id", ctr.ID), - zap.String("instance_id", instanceID), - zap.Error(err)) - continue - } - - client := agentctl.NewClient(containerIP, instanceInfo.Port, r.logger, - agentctl.WithExecutionID(instanceID), - agentctl.WithSessionID(sessionID)) - - recovered = append(recovered, &ExecutorInstance{ - InstanceID: instanceID, - TaskID: taskID, - SessionID: sessionID, - RuntimeName: string(r.Name()), - Client: client, - ContainerID: ctr.ID, - ContainerIP: containerIP, - WorkspacePath: "/workspace", - Metadata: map[string]interface{}{"agent_profile_id": agentProfileID}, - }) - - r.logger.Info("recovered docker instance", - zap.String("instance_id", instanceID), - zap.String("task_id", taskID), - zap.String("container_id", ctr.ID), - zap.Int("instance_port", instanceInfo.Port)) - } - - return recovered, nil + return nil } // GetInteractiveRunner returns nil for Docker runtime. diff --git a/apps/backend/internal/agent/lifecycle/executor_docker_test.go b/apps/backend/internal/agent/lifecycle/executor_docker_test.go new file mode 100644 index 0000000000..fc5ff9b1a7 --- /dev/null +++ b/apps/backend/internal/agent/lifecycle/executor_docker_test.go @@ -0,0 +1,345 @@ +package lifecycle + +import ( + "context" + "fmt" + "sync/atomic" + "testing" + + "github.com/kandev/kandev/internal/agent/docker" + "github.com/kandev/kandev/internal/agent/executor" + "github.com/kandev/kandev/internal/common/config" + "github.com/kandev/kandev/internal/common/logger" +) + +func newTestDockerLogger() *logger.Logger { + log, _ := logger.NewLogger(logger.LoggingConfig{Level: "error", Format: "json"}) + return log +} + +// failingClientFactory returns a factory that always fails. +func failingClientFactory(errMsg string) func(config.DockerConfig, *logger.Logger) (*docker.Client, error) { + return func(_ config.DockerConfig, _ *logger.Logger) (*docker.Client, error) { + return nil, fmt.Errorf("%s", errMsg) + } +} + +func TestNewDockerExecutor(t *testing.T) { + log := newTestDockerLogger() + exec := NewDockerExecutor(config.DockerConfig{}, log) + + if exec == nil { + t.Fatal("expected non-nil executor") + } + if exec.initialized { + t.Error("expected initialized to be false") + } + if exec.docker != nil { + t.Error("expected docker client to be nil before first use") + } + if exec.containerMgr != nil { + t.Error("expected container manager to be nil before first use") + } + if exec.newClientFunc == nil { + t.Error("expected newClientFunc to be set") + } +} + +func TestDockerExecutor_Name(t *testing.T) { + log := newTestDockerLogger() + exec := NewDockerExecutor(config.DockerConfig{}, log) + + if exec.Name() != executor.NameDocker { + t.Errorf("expected name %q, got %q", executor.NameDocker, exec.Name()) + } +} + +func TestDockerExecutor_HealthCheck(t *testing.T) { + log := newTestDockerLogger() + exec := NewDockerExecutor(config.DockerConfig{}, log) + + if err := exec.HealthCheck(context.Background()); err != nil { + t.Errorf("expected nil error, got: %v", err) + } +} + +func TestDockerExecutor_RecoverInstances(t *testing.T) { + log := newTestDockerLogger() + exec := NewDockerExecutor(config.DockerConfig{}, log) + + instances, err := exec.RecoverInstances(context.Background()) + if err != nil { + t.Errorf("expected nil error, got: %v", err) + } + if instances != nil { + t.Errorf("expected nil instances, got: %v", instances) + } +} + +func TestDockerExecutor_GetInteractiveRunner(t *testing.T) { + log := newTestDockerLogger() + exec := NewDockerExecutor(config.DockerConfig{}, log) + + if runner := exec.GetInteractiveRunner(); runner != nil { + t.Error("expected nil interactive runner for docker executor") + } +} + +func TestDockerExecutor_EnsureClient_Success(t *testing.T) { + log := newTestDockerLogger() + exec := NewDockerExecutor(config.DockerConfig{}, log) + // Default factory uses docker.NewClient which succeeds even without Docker running + + cli, cm, err := exec.ensureClient() + if err != nil { + t.Fatalf("expected nil error, got: %v", err) + } + if cli == nil { + t.Error("expected non-nil client") + } + if cm == nil { + t.Error("expected non-nil container manager") + } + if !exec.initialized { + t.Error("expected initialized to be true after success") + } + + // Second call should return cached values + cli2, cm2, err2 := exec.ensureClient() + if err2 != nil { + t.Fatalf("expected nil error on second call, got: %v", err2) + } + if cli2 != cli { + t.Error("expected same client on second call") + } + if cm2 != cm { + t.Error("expected same container manager on second call") + } + + // Clean up + _ = exec.Close() +} + +func TestDockerExecutor_EnsureClient_RetriesOnFailure(t *testing.T) { + log := newTestDockerLogger() + exec := NewDockerExecutor(config.DockerConfig{}, log) + + var callCount atomic.Int32 + exec.newClientFunc = func(_ config.DockerConfig, _ *logger.Logger) (*docker.Client, error) { + callCount.Add(1) + return nil, fmt.Errorf("docker daemon not running") + } + + // First call should fail + cli, cm, err := exec.ensureClient() + if err == nil { + t.Fatal("expected error on first call") + } + if cli != nil || cm != nil { + t.Error("expected nil client and container manager on failure") + } + if exec.initialized { + t.Error("expected initialized to be false after failure") + } + + // Second call should retry (not return a cached error like sync.Once would) + _, _, err2 := exec.ensureClient() + if err2 == nil { + t.Fatal("expected error on second call") + } + if callCount.Load() != 2 { + t.Errorf("expected factory to be called twice (retry), got %d calls", callCount.Load()) + } +} + +func TestDockerExecutor_EnsureClient_RecoversAfterTransientFailure(t *testing.T) { + log := newTestDockerLogger() + exec := NewDockerExecutor(config.DockerConfig{}, log) + + var callCount atomic.Int32 + exec.newClientFunc = func(cfg config.DockerConfig, l *logger.Logger) (*docker.Client, error) { + n := callCount.Add(1) + if n == 1 { + return nil, fmt.Errorf("transient error") + } + // Succeed on second call using real factory + return docker.NewClient(cfg, l) + } + + // First call fails + _, _, err := exec.ensureClient() + if err == nil { + t.Fatal("expected error on first call") + } + + // Second call succeeds + cli, cm, err := exec.ensureClient() + if err != nil { + t.Fatalf("expected nil error after recovery, got: %v", err) + } + if cli == nil || cm == nil { + t.Error("expected non-nil client and container manager after recovery") + } + if !exec.initialized { + t.Error("expected initialized to be true after recovery") + } + + _ = exec.Close() +} + +func TestDockerExecutor_Client_ReturnsNilOnFailure(t *testing.T) { + log := newTestDockerLogger() + exec := NewDockerExecutor(config.DockerConfig{}, log) + exec.newClientFunc = failingClientFactory("docker unavailable") + + if cli := exec.Client(); cli != nil { + t.Error("expected nil client when Docker is unavailable") + } +} + +func TestDockerExecutor_ContainerMgr_ReturnsNilOnFailure(t *testing.T) { + log := newTestDockerLogger() + exec := NewDockerExecutor(config.DockerConfig{}, log) + exec.newClientFunc = failingClientFactory("docker unavailable") + + if cm := exec.ContainerMgr(); cm != nil { + t.Error("expected nil container manager when Docker is unavailable") + } +} + +func TestDockerExecutor_Client_ReturnsClientOnSuccess(t *testing.T) { + log := newTestDockerLogger() + exec := NewDockerExecutor(config.DockerConfig{}, log) + + cli := exec.Client() + if cli == nil { + t.Error("expected non-nil client with default config") + } + + _ = exec.Close() +} + +func TestDockerExecutor_Close_BeforeInit(t *testing.T) { + log := newTestDockerLogger() + exec := NewDockerExecutor(config.DockerConfig{}, log) + + if err := exec.Close(); err != nil { + t.Errorf("expected nil error, got: %v", err) + } +} + +func TestDockerExecutor_Close_AfterInit(t *testing.T) { + log := newTestDockerLogger() + exec := NewDockerExecutor(config.DockerConfig{}, log) + + // Initialize the client + _, _, _ = exec.ensureClient() + if !exec.initialized { + t.Fatal("expected initialized to be true") + } + + // Close should succeed and reset state + if err := exec.Close(); err != nil { + t.Errorf("expected nil error, got: %v", err) + } + if exec.initialized { + t.Error("expected initialized to be false after close") + } + if exec.docker != nil { + t.Error("expected docker to be nil after close") + } + if exec.containerMgr != nil { + t.Error("expected containerMgr to be nil after close") + } +} + +func TestDockerExecutor_Close_AfterFailedInit(t *testing.T) { + log := newTestDockerLogger() + exec := NewDockerExecutor(config.DockerConfig{}, log) + exec.newClientFunc = failingClientFactory("docker unavailable") + + // Trigger failed init + _, _, _ = exec.ensureClient() + + // Close after failed init should be a no-op + if err := exec.Close(); err != nil { + t.Errorf("expected nil error, got: %v", err) + } +} + +func TestDockerClientProvider_NilRegistry(t *testing.T) { + log := newTestDockerLogger() + mgr := NewManager(nil, nil, nil, nil, nil, nil, ExecutorFallbackWarn, log) + + provider := mgr.DockerClientProvider() + if provider == nil { + t.Fatal("expected non-nil provider function") + } + if client := provider(); client != nil { + t.Error("expected nil client from provider with nil registry") + } +} + +func TestDockerClientProvider_NoDockerExecutor(t *testing.T) { + log := newTestDockerLogger() + registry := NewExecutorRegistry(log) + registry.Register(&MockExecutor{name: executor.NameStandalone}) + mgr := NewManager(nil, nil, registry, nil, nil, nil, ExecutorFallbackWarn, log) + + provider := mgr.DockerClientProvider() + if client := provider(); client != nil { + t.Error("expected nil client when no Docker executor is registered") + } +} + +func TestDockerClientProvider_WithDockerExecutor(t *testing.T) { + log := newTestDockerLogger() + registry := NewExecutorRegistry(log) + dockerExec := NewDockerExecutor(config.DockerConfig{}, log) + dockerExec.newClientFunc = failingClientFactory("docker unavailable") + registry.Register(dockerExec) + mgr := NewManager(nil, nil, registry, nil, nil, nil, ExecutorFallbackWarn, log) + + provider := mgr.DockerClientProvider() + if client := provider(); client != nil { + t.Error("expected nil client from provider with unavailable Docker") + } +} + +func TestDockerClientProvider_WithWorkingDocker(t *testing.T) { + log := newTestDockerLogger() + registry := NewExecutorRegistry(log) + dockerExec := NewDockerExecutor(config.DockerConfig{}, log) + registry.Register(dockerExec) + mgr := NewManager(nil, nil, registry, nil, nil, nil, ExecutorFallbackWarn, log) + + provider := mgr.DockerClientProvider() + client := provider() + if client == nil { + t.Error("expected non-nil client from provider with working Docker executor") + } + + _ = dockerExec.Close() +} + +func TestExecutorRegistry_CloseAll(t *testing.T) { + t.Run("closes closeable backends", func(t *testing.T) { + log := newTestDockerLogger() + registry := NewExecutorRegistry(log) + + dockerExec := NewDockerExecutor(config.DockerConfig{}, log) + registry.Register(dockerExec) + registry.Register(&MockExecutor{name: executor.NameStandalone}) + + // Should not panic + registry.CloseAll() + }) + + t.Run("empty registry", func(t *testing.T) { + log := newTestDockerLogger() + registry := NewExecutorRegistry(log) + + // Should not panic + registry.CloseAll() + }) +} diff --git a/apps/backend/internal/agent/lifecycle/executor_registry.go b/apps/backend/internal/agent/lifecycle/executor_registry.go index f7b4ade300..78cc2bb758 100644 --- a/apps/backend/internal/agent/lifecycle/executor_registry.go +++ b/apps/backend/internal/agent/lifecycle/executor_registry.go @@ -94,6 +94,28 @@ func (r *ExecutorRegistry) HealthCheckAll(ctx context.Context) map[executor.Name return results } +// Closeable is an optional interface for executors that hold resources +// requiring cleanup on shutdown (e.g., Docker SDK client connections). +type Closeable interface { + Close() error +} + +// CloseAll closes all registered backends that implement Closeable. +func (r *ExecutorRegistry) CloseAll() { + r.mu.RLock() + defer r.mu.RUnlock() + + for name, rt := range r.backends { + if closer, ok := rt.(Closeable); ok { + if err := closer.Close(); err != nil { + r.logger.Warn("failed to close runtime", + zap.String("runtime", string(name)), + zap.Error(err)) + } + } + } +} + // 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. diff --git a/apps/backend/internal/agent/lifecycle/manager.go b/apps/backend/internal/agent/lifecycle/manager.go index 68cbb0b51b..8c42a45667 100644 --- a/apps/backend/internal/agent/lifecycle/manager.go +++ b/apps/backend/internal/agent/lifecycle/manager.go @@ -8,6 +8,8 @@ import ( "go.uber.org/zap" + "github.com/kandev/kandev/internal/agent/docker" + "github.com/kandev/kandev/internal/agent/executor" "github.com/kandev/kandev/internal/agent/registry" agentctl "github.com/kandev/kandev/internal/agentctl/client" "github.com/kandev/kandev/internal/common/logger" @@ -45,13 +47,12 @@ type Manager struct { executorFallbackPolicy ExecutorFallbackPolicy // 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 + historyManager *SessionHistoryManager // Stores session history for context injection (fork_session pattern) // Workspace info provider for on-demand instance creation workspaceInfoProvider WorkspaceInfoProvider @@ -76,13 +77,11 @@ type Manager struct { // NewManager creates a new lifecycle manager. // The executorRegistry 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, executorRegistry *ExecutorRegistry, - containerManager *ContainerManager, credsMgr CredentialsManager, profileResolver ProfileResolver, mcpProvider McpConfigProvider, @@ -125,7 +124,6 @@ func NewManager( commandBuilder: commandBuilder, sessionManager: sessionManager, eventPublisher: eventPublisher, - containerManager: containerManager, historyManager: historyManager, remoteStatusPollInterval: 60 * time.Second, remoteStatusBySession: make(map[string]*RemoteStatus), @@ -209,3 +207,22 @@ func (m *Manager) SetBootMessageService(svc BootMessageService) { func (m *Manager) SetPreparerRegistry(registry *PreparerRegistry) { m.preparerRegistry = registry } + +// DockerClientProvider returns a function that lazily resolves the Docker client +// from the Docker executor in the registry. Returns nil if Docker is unavailable. +func (m *Manager) DockerClientProvider() func() *docker.Client { + return func() *docker.Client { + if m.executorRegistry == nil { + return nil + } + backend, err := m.executorRegistry.GetBackend(executor.NameDocker) + if err != nil { + return nil + } + dockerExec, ok := backend.(*DockerExecutor) + if !ok { + return nil + } + return dockerExec.Client() + } +} diff --git a/apps/backend/internal/agent/lifecycle/manager_events_test.go b/apps/backend/internal/agent/lifecycle/manager_events_test.go index 2266f132a7..86f7bab242 100644 --- a/apps/backend/internal/agent/lifecycle/manager_events_test.go +++ b/apps/backend/internal/agent/lifecycle/manager_events_test.go @@ -66,7 +66,7 @@ func createTestManagerWithTracking() (*Manager, *MockEventBusWithTracking) { eventBus := &MockEventBusWithTracking{} credsMgr := &MockCredentialsManager{} profileResolver := &MockProfileResolver{} - mgr := NewManager(reg, eventBus, nil, nil, credsMgr, profileResolver, nil, ExecutorFallbackWarn, log) + mgr := NewManager(reg, eventBus, nil, credsMgr, profileResolver, nil, ExecutorFallbackWarn, log) return mgr, eventBus } diff --git a/apps/backend/internal/agent/lifecycle/manager_lifecycle.go b/apps/backend/internal/agent/lifecycle/manager_lifecycle.go index 31a246e956..b24cd57b66 100644 --- a/apps/backend/internal/agent/lifecycle/manager_lifecycle.go +++ b/apps/backend/internal/agent/lifecycle/manager_lifecycle.go @@ -133,13 +133,18 @@ func (m *Manager) GetRecoveredExecutions() []RecoveredExecution { return result } -// Stop stops the lifecycle manager +// Stop stops the lifecycle manager and releases resources held by executors. func (m *Manager) Stop() error { m.logger.Info("stopping lifecycle manager") close(m.stopCh) m.wg.Wait() + // Close executor backends that hold resources (e.g., Docker SDK client). + if m.executorRegistry != nil { + m.executorRegistry.CloseAll() + } + return nil } @@ -183,8 +188,23 @@ func (m *Manager) cleanupExitedContainer(ctx context.Context, containerID string return } + // Get the Docker executor's ContainerManager from the registry. + var containerMgr *ContainerManager + if m.executorRegistry != nil { + if backend, berr := m.executorRegistry.GetBackend(executor.NameDocker); berr == nil { + if dockerExec, ok := backend.(*DockerExecutor); ok { + containerMgr = dockerExec.ContainerMgr() + } + } + } + if containerMgr == nil { + m.logger.Warn("docker container manager unavailable, cannot clean up container", + zap.String("container_id", containerID)) + return + } + // Get container info to get exit code - info, err := m.containerManager.GetContainerInfo(ctx, containerID) + info, err := containerMgr.GetContainerInfo(ctx, containerID) if err != nil { m.logger.Warn("failed to get container info during cleanup", zap.String("container_id", containerID), @@ -200,7 +220,7 @@ func (m *Manager) cleanupExitedContainer(ctx context.Context, containerID string _ = m.MarkCompleted(execution.ID, info.ExitCode, errorMsg) // Remove the container - if err := m.containerManager.RemoveContainer(ctx, containerID, false); err != nil { + if err := containerMgr.RemoveContainer(ctx, containerID, false); err != nil { m.logger.Warn("failed to remove container during cleanup", zap.String("container_id", containerID), zap.Error(err)) diff --git a/apps/backend/internal/agent/lifecycle/manager_test.go b/apps/backend/internal/agent/lifecycle/manager_test.go index 212b02ec53..05e0ed7e5b 100644 --- a/apps/backend/internal/agent/lifecycle/manager_test.go +++ b/apps/backend/internal/agent/lifecycle/manager_test.go @@ -206,8 +206,8 @@ func newTestManager() *Manager { eventBus := &MockEventBus{} credsMgr := &MockCredentialsManager{} profileResolver := &MockProfileResolver{} - // Pass nil for runtime and containerManager - tests don't need them - return NewManager(reg, eventBus, nil, nil, credsMgr, profileResolver, nil, ExecutorFallbackWarn, log) + // Pass nil for runtime - tests don't need them + return NewManager(reg, eventBus, nil, credsMgr, profileResolver, nil, ExecutorFallbackWarn, log) } func TestNewManager(t *testing.T) { diff --git a/apps/backend/internal/task/handlers/process_handlers_test.go b/apps/backend/internal/task/handlers/process_handlers_test.go index be27566ea2..a45b1df780 100644 --- a/apps/backend/internal/task/handlers/process_handlers_test.go +++ b/apps/backend/internal/task/handlers/process_handlers_test.go @@ -462,7 +462,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, lifecycle.ExecutorFallbackWarn, log) + return lifecycle.NewManager(reg, nil, nil, nil, nil, nil, lifecycle.ExecutorFallbackWarn, log) } func newAgentctlClient(t *testing.T, serverURL string, log *logger.Logger) *agentctlclient.Client {