From 32c6cd5bc9f1ae2dcc28fa05ac1702e2ca291dfc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jos=C3=A9=20Almeida?= Date: Thu, 7 May 2026 12:16:13 +0100 Subject: [PATCH] refactor(docker): split reconnectToContainer + cover reconnect helpers with unit tests MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address recurring review feedback on the docker reconnect path: - reconnectToContainer was 98 lines (over the 80-line repo limit). Split into ensureContainerRunning (inspect + start + re-inspect) and bringupAgentctl (health-check, find-or-create instance, transparent re-handshake). The orchestration body now reads top-down at 49 lines. - resolveDockerEndpoint and waitForAgentctlHealth grew real branches in this PR (host-port lookup with container-IP fallback; 240×500ms retry loop) but had no direct tests. Introduce narrow hostPortLookup / healthChecker interfaces, satisfied by *docker.Client and *agentctl.ControlClient, so both functions are now driven by stubs that exercise happy-path, fallback, retry-until-healthy, exhaustion, and context-cancellation branches. - Add direct unit tests for the docker package's pure helpers (normalizeDockerHostIP, buildDockerPortBindings) and for dockerAgentctlPortBindings — a regression in the published port range or host IP would silently break container reconnect. --- .../internal/agent/docker/client_test.go | 76 +++++++ .../agent/lifecycle/container_config_test.go | 37 ++++ .../agent/lifecycle/executor_docker.go | 195 +++++++++++------- .../executor_docker_reconnect_test.go | 133 ++++++++++++ 4 files changed, 368 insertions(+), 73 deletions(-) create mode 100644 apps/backend/internal/agent/docker/client_test.go create mode 100644 apps/backend/internal/agent/lifecycle/executor_docker_reconnect_test.go diff --git a/apps/backend/internal/agent/docker/client_test.go b/apps/backend/internal/agent/docker/client_test.go new file mode 100644 index 0000000000..d2d86476b9 --- /dev/null +++ b/apps/backend/internal/agent/docker/client_test.go @@ -0,0 +1,76 @@ +package docker + +import ( + "fmt" + "testing" + + "github.com/docker/go-connections/nat" +) + +func TestNormalizeDockerHostIP(t *testing.T) { + cases := []struct { + in string + want string + }{ + {in: "", want: "127.0.0.1"}, + {in: "0.0.0.0", want: "127.0.0.1"}, + {in: "::", want: "127.0.0.1"}, + {in: "127.0.0.1", want: "127.0.0.1"}, + {in: "10.0.0.5", want: "10.0.0.5"}, + {in: "::1", want: "::1"}, + } + for _, tc := range cases { + t.Run(tc.in, func(t *testing.T) { + if got := normalizeDockerHostIP(tc.in); got != tc.want { + t.Errorf("normalizeDockerHostIP(%q) = %q, want %q", tc.in, got, tc.want) + } + }) + } +} + +func TestBuildDockerPortBindings_EmptyReturnsNil(t *testing.T) { + exposed, bindings := buildDockerPortBindings(nil) + if exposed != nil { + t.Errorf("expected nil exposed ports, got %v", exposed) + } + if bindings != nil { + t.Errorf("expected nil bindings, got %v", bindings) + } +} + +func TestBuildDockerPortBindings_AssignsContainerAndHost(t *testing.T) { + in := []PortBindingConfig{ + {ContainerPort: 8080, HostIP: "127.0.0.1", HostPort: "0"}, + {ContainerPort: 9000, HostIP: "0.0.0.0", HostPort: "9001"}, + } + exposed, bindings := buildDockerPortBindings(in) + + if got := len(exposed); got != 2 { + t.Fatalf("exposed ports = %d, want 2", got) + } + for _, b := range in { + key := nat.Port(fmt.Sprintf("%d/tcp", b.ContainerPort)) + if _, ok := exposed[key]; !ok { + t.Errorf("exposed missing %s", key) + } + got := bindings[key] + if len(got) != 1 { + t.Fatalf("bindings[%s] = %d entries, want 1", key, len(got)) + } + if got[0].HostIP != b.HostIP || got[0].HostPort != b.HostPort { + t.Errorf("bindings[%s] = %+v, want host_ip=%q host_port=%q", key, got[0], b.HostIP, b.HostPort) + } + } +} + +func TestBuildDockerPortBindings_DeduplicatesContainerPort(t *testing.T) { + in := []PortBindingConfig{ + {ContainerPort: 7000, HostIP: "127.0.0.1", HostPort: "0"}, + {ContainerPort: 7000, HostIP: "10.0.0.5", HostPort: "7000"}, + } + _, bindings := buildDockerPortBindings(in) + got := bindings[nat.Port("7000/tcp")] + if len(got) != 2 { + t.Fatalf("want both bindings on port 7000, got %d", len(got)) + } +} diff --git a/apps/backend/internal/agent/lifecycle/container_config_test.go b/apps/backend/internal/agent/lifecycle/container_config_test.go index aab1fba98a..7a58e6af33 100644 --- a/apps/backend/internal/agent/lifecycle/container_config_test.go +++ b/apps/backend/internal/agent/lifecycle/container_config_test.go @@ -180,6 +180,43 @@ func TestBuildContainerConfig_PublishesAgentctlPorts(t *testing.T) { assertEnvContains(t, got.Env, "AGENTCTL_INSTANCE_PORT_MAX=41100") } +// TestDockerAgentctlPortBindings is a direct test for the helper that +// generates the published-port set for every kandev-managed Docker agent +// container. A regression here (wrong port range, missing agentctl port, +// non-loopback host IP) would silently break container reconnect, since +// `resolveDockerEndpoint` falls back to the container IP when the published +// port lookup fails. +func TestDockerAgentctlPortBindings(t *testing.T) { + bindings := dockerAgentctlPortBindings() + + wantTotal := 1 + (dockerAgentctlInstancePortMax - dockerAgentctlInstancePortBase + 1) + if len(bindings) != wantTotal { + t.Fatalf("got %d bindings, want %d (control + instance range)", len(bindings), wantTotal) + } + + // Control port must be present. + assertHasPortBinding(t, bindings, AgentCtlPort) + + // Every port in the instance range must be present and bound to loopback + // with a kernel-assigned host port. + have := make(map[int]docker.PortBindingConfig, len(bindings)) + for _, b := range bindings { + have[b.ContainerPort] = b + } + for port := dockerAgentctlInstancePortBase; port <= dockerAgentctlInstancePortMax; port++ { + b, ok := have[port] + if !ok { + t.Fatalf("missing instance port %d in published bindings", port) + } + if b.HostIP != "127.0.0.1" { + t.Errorf("port %d host_ip = %q, want 127.0.0.1", port, b.HostIP) + } + if b.HostPort != "0" { + t.Errorf("port %d host_port = %q, want kernel-assigned (\"0\")", port, b.HostPort) + } + } +} + // TestBuildContainerConfig_SessionDirIsKandevManagedForEveryAgent locks in // the agent-agnostic guarantee that bind sources for SessionDirTemplate // resolve to /agent-sessions// and never to diff --git a/apps/backend/internal/agent/lifecycle/executor_docker.go b/apps/backend/internal/agent/lifecycle/executor_docker.go index 3354c57a07..e162e0c912 100644 --- a/apps/backend/internal/agent/lifecycle/executor_docker.go +++ b/apps/backend/internal/agent/lifecycle/executor_docker.go @@ -227,87 +227,33 @@ func (r *DockerExecutor) reconnectToContainer(ctx context.Context, dockerClient if err != nil { return nil, err } - prevID := req.PreviousExecutionID - - info, err := dockerClient.GetContainerInfo(ctx, containerRef) + info, containerIP, err := r.ensureContainerRunning(ctx, dockerClient, containerRef) if err != nil { - return nil, fmt.Errorf("failed to inspect container %s: %w", containerRef, err) - } - if shouldStartExistingDockerContainer(info.State) { - r.logger.Info("starting stopped docker container for reconnect", - zap.String("container_id", info.ID), - zap.String("state", info.State)) - if err := dockerClient.StartContainer(ctx, info.ID); err != nil { - return nil, fmt.Errorf("failed to start container %s: %w", info.ID, err) - } - info, err = dockerClient.GetContainerInfo(ctx, info.ID) - if err != nil { - return nil, fmt.Errorf("failed to inspect started container %s: %w", containerRef, err) - } - } - if info.State != containerStateRunning { - return nil, fmt.Errorf("container %s is %s, not %s", containerRef, info.State, containerStateRunning) + return nil, err } - // Get the container IP - containerIP, err := dockerClient.GetContainerIP(ctx, info.ID) + conn, err := r.bringupAgentctl(ctx, dockerClient, info.ID, containerIP, req) if err != nil { - return nil, fmt.Errorf("failed to get IP for container %s: %w", info.ID, err) + return nil, err } - controlHost, controlPort := resolveDockerEndpoint(ctx, dockerClient, info.ID, AgentCtlPort, containerIP, r.logger) - - // Check if agentctl is healthy - ctl := agentctl.NewControlClient(controlHost, controlPort, r.logger, - agentctl.WithControlAuthToken(req.AuthToken)) - if err := r.waitForAgentctlHealth(ctx, ctl); err != nil { - return nil, fmt.Errorf("agentctl not healthy in container %s: %w", info.ID, err) - } - - // Check if the previous instance is still alive - authToken := req.AuthToken - instanceID := reconnectInstanceID(req, prevID) - instancePort, reusingProcess, err := r.findExistingInstance(ctx, dockerClient, ctl, req, info.ID, containerIP, instanceID, authToken) - if err != nil && req.BootstrapNonce != "" && isAgentctlAuthError(err) { - var handshakeErr error - authToken, handshakeErr = ctl.Handshake(ctx, req.BootstrapNonce) - if handshakeErr != nil { - return nil, fmt.Errorf("agentctl auth failed and re-handshake failed in container %s: %w", info.ID, handshakeErr) - } - instancePort, reusingProcess, err = r.findExistingInstance(ctx, dockerClient, ctl, req, info.ID, containerIP, instanceID, authToken) - } - if err != nil { - return nil, fmt.Errorf("failed to find instance in container %s: %w", info.ID, err) - } - instanceHost, resolvedInstancePort := resolveDockerEndpoint(ctx, dockerClient, info.ID, instancePort, containerIP, r.logger) - - // Create client pointing to the instance port - client := agentctl.NewClient(instanceHost, resolvedInstancePort, r.logger, + client := agentctl.NewClient(conn.instanceHost, conn.instancePort, r.logger, agentctl.WithExecutionID(req.InstanceID), agentctl.WithSessionID(req.SessionID), - agentctl.WithAuthToken(authToken)) - - metadata := map[string]interface{}{ - MetadataKeyIsRemote: true, - MetadataKeyContainerID: info.ID, - "reuse_existing_process": reusingProcess, - } - - refreshedAuthToken := "" - if authToken != "" && authToken != req.AuthToken { - refreshedAuthToken = authToken - } + agentctl.WithAuthToken(conn.authToken)) r.logger.Info("reconnected to existing docker container", zap.String("container_ref", containerRef), zap.String("container_id", info.ID), zap.String("container_ip", containerIP), - zap.String("control_host", controlHost), - zap.Int("control_port", controlPort), - zap.String("instance_host", instanceHost), - zap.Int("instance_port", resolvedInstancePort), - zap.Bool("reusing_process", reusingProcess)) + zap.String("instance_host", conn.instanceHost), + zap.Int("instance_port", conn.instancePort), + zap.Bool("reusing_process", conn.reusingProcess)) + refreshedAuthToken := "" + if conn.authToken != "" && conn.authToken != req.AuthToken { + refreshedAuthToken = conn.authToken + } return &ExecutorInstance{ InstanceID: req.InstanceID, TaskID: req.TaskID, @@ -317,8 +263,88 @@ func (r *DockerExecutor) reconnectToContainer(ctx context.Context, dockerClient ContainerID: info.ID, ContainerIP: containerIP, WorkspacePath: dockerWorkspacePath, - Metadata: metadata, - AuthToken: refreshedAuthToken, + Metadata: map[string]interface{}{ + MetadataKeyIsRemote: true, + MetadataKeyContainerID: info.ID, + "reuse_existing_process": conn.reusingProcess, + }, + AuthToken: refreshedAuthToken, + }, nil +} + +// ensureContainerRunning inspects containerRef, starts it if it's stopped, +// re-inspects, and returns the live container info and IP. Errors out if the +// container can't be brought to a running state — the caller can't reconnect +// to a container that isn't alive. +func (r *DockerExecutor) ensureContainerRunning(ctx context.Context, dockerClient *docker.Client, containerRef string) (*docker.ContainerInfo, string, error) { + info, err := dockerClient.GetContainerInfo(ctx, containerRef) + if err != nil { + return nil, "", fmt.Errorf("failed to inspect container %s: %w", containerRef, err) + } + if shouldStartExistingDockerContainer(info.State) { + r.logger.Info("starting stopped docker container for reconnect", + zap.String("container_id", info.ID), + zap.String("state", info.State)) + if err := dockerClient.StartContainer(ctx, info.ID); err != nil { + return nil, "", fmt.Errorf("failed to start container %s: %w", info.ID, err) + } + info, err = dockerClient.GetContainerInfo(ctx, info.ID) + if err != nil { + return nil, "", fmt.Errorf("failed to inspect started container %s: %w", containerRef, err) + } + } + if info.State != containerStateRunning { + return nil, "", fmt.Errorf("container %s is %s, not %s", containerRef, info.State, containerStateRunning) + } + containerIP, err := dockerClient.GetContainerIP(ctx, info.ID) + if err != nil { + return nil, "", fmt.Errorf("failed to get IP for container %s: %w", info.ID, err) + } + return info, containerIP, nil +} + +// reconnectAgentctlConn captures the resolved endpoint state needed by the +// caller after agentctl has been brought up: the instance-level host:port for +// the user-facing client, the auth token in effect (which may have been +// refreshed via re-handshake), and whether the agent subprocess is reusable. +type reconnectAgentctlConn struct { + instanceHost string + instancePort int + authToken string + reusingProcess bool +} + +// bringupAgentctl health-checks agentctl on the container, finds (or creates) +// the agent instance, transparently re-handshakes on a 401, and returns the +// resolved instance endpoint for the user-facing client. +func (r *DockerExecutor) bringupAgentctl(ctx context.Context, dockerClient *docker.Client, containerID, containerIP string, req *ExecutorCreateRequest) (reconnectAgentctlConn, error) { + controlHost, controlPort := resolveDockerEndpoint(ctx, dockerClient, containerID, AgentCtlPort, containerIP, r.logger) + ctl := agentctl.NewControlClient(controlHost, controlPort, r.logger, + agentctl.WithControlAuthToken(req.AuthToken)) + if err := r.waitForAgentctlHealth(ctx, ctl); err != nil { + return reconnectAgentctlConn{}, fmt.Errorf("agentctl not healthy in container %s: %w", containerID, err) + } + + authToken := req.AuthToken + instanceID := reconnectInstanceID(req, req.PreviousExecutionID) + instancePort, reusingProcess, err := r.findExistingInstance(ctx, dockerClient, ctl, req, containerID, containerIP, instanceID, authToken) + if err != nil && req.BootstrapNonce != "" && isAgentctlAuthError(err) { + var handshakeErr error + authToken, handshakeErr = ctl.Handshake(ctx, req.BootstrapNonce) + if handshakeErr != nil { + return reconnectAgentctlConn{}, fmt.Errorf("agentctl auth failed and re-handshake failed in container %s: %w", containerID, handshakeErr) + } + instancePort, reusingProcess, err = r.findExistingInstance(ctx, dockerClient, ctl, req, containerID, containerIP, instanceID, authToken) + } + if err != nil { + return reconnectAgentctlConn{}, fmt.Errorf("failed to find instance in container %s: %w", containerID, err) + } + instanceHost, resolvedInstancePort := resolveDockerEndpoint(ctx, dockerClient, containerID, instancePort, containerIP, r.logger) + return reconnectAgentctlConn{ + instanceHost: instanceHost, + instancePort: resolvedInstancePort, + authToken: authToken, + reusingProcess: reusingProcess, }, nil } @@ -424,10 +450,26 @@ func buildReconnectCreateInstanceRequest(req *ExecutorCreateRequest, instanceID } } -func (r *DockerExecutor) waitForAgentctlHealth(ctx context.Context, ctl *agentctl.ControlClient) error { - const maxRetries = 240 - const retryDelay = 500 * time.Millisecond +// healthChecker is the narrow interface waitForAgentctlHealth needs from the +// agentctl control client. *agentctl.ControlClient satisfies it; tests can +// stub it without spinning up an HTTP server. +type healthChecker interface { + Health(ctx context.Context) error +} +const ( + agentctlHealthMaxRetries = 240 + agentctlHealthRetryDelay = 500 * time.Millisecond +) + +func (r *DockerExecutor) waitForAgentctlHealth(ctx context.Context, ctl healthChecker) error { + return waitForAgentctlHealthWith(ctx, ctl, agentctlHealthMaxRetries, agentctlHealthRetryDelay) +} + +// waitForAgentctlHealthWith is the parameterised body — exposed for tests so +// they can drive the retry loop without a 2-minute wait. The production path +// always calls it with the package constants. +func waitForAgentctlHealthWith(ctx context.Context, ctl healthChecker, maxRetries int, retryDelay time.Duration) error { var lastErr error for i := 0; i < maxRetries; i++ { if err := ctl.Health(ctx); err == nil { @@ -449,9 +491,16 @@ func (r *DockerExecutor) waitForAgentctlHealth(ctx context.Context, ctl *agentct time.Duration(maxRetries)*retryDelay) } +// hostPortLookup is the narrow interface resolveDockerEndpoint needs from the +// docker client. *docker.Client satisfies it; tests can stub it without +// having to spin up a docker daemon. +type hostPortLookup interface { + GetContainerHostPort(ctx context.Context, containerID string, containerPort int) (string, int, error) +} + func resolveDockerEndpoint( ctx context.Context, - dockerClient *docker.Client, + dockerClient hostPortLookup, containerID string, containerPort int, fallbackHost string, diff --git a/apps/backend/internal/agent/lifecycle/executor_docker_reconnect_test.go b/apps/backend/internal/agent/lifecycle/executor_docker_reconnect_test.go new file mode 100644 index 0000000000..d144780ceb --- /dev/null +++ b/apps/backend/internal/agent/lifecycle/executor_docker_reconnect_test.go @@ -0,0 +1,133 @@ +package lifecycle + +import ( + "context" + "errors" + "sync/atomic" + "testing" + "time" +) + +// stubHostPortLookup satisfies hostPortLookup for resolveDockerEndpoint +// tests without needing a real docker daemon. +type stubHostPortLookup struct { + host string + port int + err error +} + +func (s stubHostPortLookup) GetContainerHostPort(_ context.Context, _ string, _ int) (string, int, error) { + return s.host, s.port, s.err +} + +func TestResolveDockerEndpoint_PrefersPublishedHostPort(t *testing.T) { + host, port := resolveDockerEndpoint( + context.Background(), + stubHostPortLookup{host: "127.0.0.1", port: 54321}, + "container-x", 4000, "172.17.0.5", + newTestDockerLogger(), + ) + if host != "127.0.0.1" || port != 54321 { + t.Fatalf("got %s:%d, want 127.0.0.1:54321 (published port wins)", host, port) + } +} + +func TestResolveDockerEndpoint_FallsBackToContainerIPOnLookupError(t *testing.T) { + // Non-Linux dev hosts can't reach the container IP directly, so + // resolveDockerEndpoint falls back to the published host port. When the + // lookup itself errors (e.g. docker daemon hiccup), the fallback path is + // the container IP at the requested container port — anything else + // silently routes traffic to the wrong endpoint. + host, port := resolveDockerEndpoint( + context.Background(), + stubHostPortLookup{err: errors.New("port not published")}, + "container-x", 4000, "172.17.0.5", + newTestDockerLogger(), + ) + if host != "172.17.0.5" || port != 4000 { + t.Fatalf("got %s:%d, want 172.17.0.5:4000 (container IP fallback)", host, port) + } +} + +// stubHealthChecker satisfies healthChecker for waitForAgentctlHealth tests. +// Health() succeeds after `becomesHealthyAfter` invocations and returns +// `failErr` until then. Tracks call count so tests can assert the retry +// budget was respected. +type stubHealthChecker struct { + becomesHealthyAfter int32 + failErr error + calls int32 +} + +func (s *stubHealthChecker) Health(_ context.Context) error { + n := atomic.AddInt32(&s.calls, 1) + if n >= s.becomesHealthyAfter { + return nil + } + return s.failErr +} + +func TestWaitForAgentctlHealth_HappyPathReturnsImmediately(t *testing.T) { + stub := &stubHealthChecker{becomesHealthyAfter: 1} + r := &DockerExecutor{logger: newTestDockerLogger()} + + if err := waitForAgentctlHealthWith(context.Background(), stub, 5, time.Millisecond); err != nil { + t.Fatalf("happy-path Health() should not error: %v", err) + } + _ = r // keep the parameter shape parallel with production callers + if got := atomic.LoadInt32(&stub.calls); got != 1 { + t.Fatalf("expected 1 Health() call, got %d", got) + } +} + +func TestWaitForAgentctlHealth_RetriesUntilHealthy(t *testing.T) { + stub := &stubHealthChecker{ + becomesHealthyAfter: 3, + failErr: errors.New("connection refused"), + } + + if err := waitForAgentctlHealthWith(context.Background(), stub, 10, time.Millisecond); err != nil { + t.Fatalf("expected eventual success, got %v", err) + } + if got := atomic.LoadInt32(&stub.calls); got != 3 { + t.Fatalf("expected 3 Health() calls, got %d", got) + } +} + +func TestWaitForAgentctlHealth_GivesUpAfterMaxRetries(t *testing.T) { + stub := &stubHealthChecker{ + becomesHealthyAfter: 999, // never succeeds within the test budget + failErr: errors.New("connection refused"), + } + + err := waitForAgentctlHealthWith(context.Background(), stub, 4, time.Millisecond) + if err == nil { + t.Fatal("expected exhaustion error, got nil") + } + if got := atomic.LoadInt32(&stub.calls); got != 4 { + t.Fatalf("expected 4 Health() calls (one per retry), got %d", got) + } + if !errors.Is(err, stub.failErr) { + t.Fatalf("exhaustion error should wrap the last underlying error: %v", err) + } +} + +func TestWaitForAgentctlHealth_HonoursContextCancellation(t *testing.T) { + stub := &stubHealthChecker{ + becomesHealthyAfter: 999, + failErr: errors.New("connection refused"), + } + + ctx, cancel := context.WithCancel(context.Background()) + cancel() // pre-cancelled + + err := waitForAgentctlHealthWith(ctx, stub, 100, time.Millisecond) + if !errors.Is(err, context.Canceled) { + t.Fatalf("expected context cancellation, got %v", err) + } + // One Health() call before the loop notices cancellation; the subsequent + // time.Sleep is short so cancellation gets observed promptly. + if got := atomic.LoadInt32(&stub.calls); got > 2 { + t.Fatalf("expected ≤2 Health() calls before honouring cancellation, got %d", got) + } +}