fix: lazy Docker client initialization to avoid startup errors (#300)

This commit is contained in:
Carlos Florêncio
2026-02-27 15:24:12 +00:00
committed by GitHub
parent 2e828d343c
commit 31dc3d148b
13 changed files with 570 additions and 181 deletions
+4 -11
View File
@@ -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,
+1 -20
View File
@@ -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)
+5 -17
View File
@@ -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,
+43 -14
View File
@@ -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"})
@@ -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) {
@@ -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.
@@ -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()
})
}
@@ -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.
@@ -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()
}
}
@@ -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
}
@@ -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))
@@ -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) {
@@ -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 {