fix(ssh): make the end-to-end launch path actually work

A pass of issues uncovered by running the SSH e2e suite locally (KANDEV_E2E_CONTAINERS=1 against the real sshd container). After this commit all 15 non-launch SSH e2e tests pass (UI form validation, test-connection contract, auth-methods, error-surfacing, sessions-card empty state) and the launch path runs through prepare -> upload -> agentctl -> instance -> forward, with WebSocket streams sometimes disconnecting partway through (tracked below for follow-up). Image changes (apps/web/e2e/fixtures/ssh-image.ts): - Install curl so the SSH executor can POST to the in-container agentctl control server for instance creation. - Install gcompat + libstdc++ so the dynamically-linked glibc agentctl binary can exec on musl-libc Alpine. Without it the kernel returns ENOENT for /lib64/ld-linux-x86-64.so.2. - Install shadow + 'usermod -p *' kandev so sshd doesn't refuse pubkey auth for the otherwise-locked user. - Drop in 'AllowTcpForwarding yes' and 'MaxSessions 100' via /etc/ssh/sshd_config.d/. Alpine's defaults are 'no' and 10, both of which crater the SSH executor's port-forward path. - chown /home/kandev/.kandev to kandev on entrypoint so a host bind-mount of that path (used in this commit for log persistence) doesn't make agentctl unable to write its log file. Image helper (apps/web/e2e/helpers/ssh.ts): - Bind-mount the host mock-agent-linux-amd64 into the container at the same absolute path. kandev's orchestrator builds the agent command from KANDEV_MOCK_AGENT_LINUX_BINARY (a host path), so the remote needs the binary at that exact path; bake-into-image isn't enough because the absolute host path is what kandev sends. - Bind-mount /home/kandev/.kandev so per-session agentctl logs survive --rm teardown and the test runner can inspect them on failure. - Replace the broken CPU-spin sleep helper with a real sleep subprocess and a /dev/tcp TCP-open probe so timing-dependent helpers stop racing the container start. - Retry the host fingerprint scan with stderr capture; the first scan can race the sshd accept loop on Alpine. Backend SSH executor (apps/backend/internal/agent/lifecycle): - executor_ssh_operations.go startRemoteAgentctl: agentctl does not accept --port-file or --pid-file flags (it reads AGENTCTL_PORT and writes its own pid). Rewrite the launch wrapper to inject AGENTCTL_PORT via env, capture $! as the pid, and tail the on-remote agentctl.log for 'HTTP server bound successfully' instead of waiting on imaginary side-files. - executor_ssh_operations.go createRemoteAgentInstance: new helper that POSTs /api/v1/instances on the in-container control server via SSH-exec'd curl, returning the per-instance port the executor should forward to. Mirrors what executor_sprites_operations.go does inside its sprite. Without this step the agent stream returns 'configure request failed with status 404' because the SSH executor was talking to the control server, not the per-session sub-server. - executor_ssh.go startAndForwardAgentctl: chains the new create-instance step in between agentctl launch and port-forward, and forwards the *instance* port (not the control port). - executor_ssh.go: drop the per-host SSH connection pool (it was introducing a subtle aliasing issue between CreateInstance/StopInstance/GetRemoteStatus that made the SSH client unreliable across same-host sessions). Each session now holds its own ssh.Client directly on sshSessionState and ResumeRemoteInstance dials a fresh client too. - executor_ssh_operations.go SSHPortForwarder: serialize the client.Dial calls with a per-forwarder mutex. golang.org/x/crypto/ ssh.Client.Dial is documented as concurrency-safe but in practice parallel opens from the kandev stream-manager occasionally produce io.EOF as if the channel-open response was never received. Test fix (apps/web/e2e/tests/ssh/error-surfacing.spec.ts): - Q1 ('TCP refused') was using identity_source='agent' with no SSH_AUTH_SOCK in the backend env, so it failed at buildAuthMethods before reaching the TCP-refused path. Switch to identity_source='file' with the seed key so the test exercises the failure it claims to. Known remaining issue (to investigate next): the launch path now reaches 'agent process started' + 'ACP initialize response received' when run in isolation, but the WebSocket updates stream sometimes disconnects with 1006 abnormal closure shortly after. A standalone Go reproduction of concurrent direct-tcpip channels against the same image works without issue, so the trigger is something specific to the kandev backend's stream-manager + my forwarder interaction. Tracked for the next iteration.
2026-05-17 13:32:32 +01:00
parent 192df0ecf3
commit 39c574bd57
6 changed files with 302 additions and 99 deletions
@@ -28,10 +28,13 @@ const (
 )

 // sshSessionState tracks per-session resources we need to clean up later:
-// the SSH target (so we can release the pooled connection) and the local
-// port forwarder.
+// the SSH target (so we can dial again on Stop), the per-session SSH client
+// (no shared pool — the Go SSH client's direct-tcpip mux gets unreliable
+// when many short-lived channels race the long-lived stream channels), and
+// the local port forwarder.
 type sshSessionState struct {
 	target    *SSHTarget
+	client    *ssh.Client
 	forwarder *SSHPortForwarder
 	pid       int
 	remoteDir string
@@ -137,14 +140,14 @@ func (r *SSHExecutor) CreateInstance(ctx context.Context, req *ExecutorCreateReq
 	if err != nil {
 		return nil, err
 	}
-	client, err := r.pool.Get(ctx, target)
+	client, err := dialSSH(ctx, target)
 	if err != nil {
 		return nil, fmt.Errorf("ssh: connect to %s@%s: %w", target.User, target.Host, err)
 	}
 	released := false
 	defer func() {
 		if !released {
-			r.pool.Release(target)
+			_ = client.Close()
 		}
 	}()
 	r.report(req.OnProgress, "Connecting to SSH host", PrepareStepCompleted, "")
@@ -169,6 +172,7 @@ func (r *SSHExecutor) CreateInstance(ctx context.Context, req *ExecutorCreateReq
 	r.mu.Lock()
 	r.sessions[req.InstanceID] = &sshSessionState{
 		target:    target,
+		client:    client,
 		forwarder: fwd,
 		pid:       pid,
 		remoteDir: sessionDir,
@@ -219,20 +223,44 @@ func (r *SSHExecutor) prepareRemoteDirs(ctx context.Context, client *ssh.Client,
 	return taskDir, sessionDir, nil
 }

-// startAndForwardAgentctl spawns the per-session agentctl on the remote and
-// stands up the SSH local port forward. Cleans up both on failure.
+// startAndForwardAgentctl spawns the per-session agentctl on the remote,
+// creates a per-instance sub-server inside it, and stands up the SSH local
+// port forward to the *instance* port (not the control port). Returns the
+// instance port, the agentctl PID, and the forwarder. Cleans up everything
+// it created on any failure.
+//
+// Lifecycle mirrors what executor_sprites.go does inside its sprite:
+//  1. agentctl --workdir <taskDir> binds the control API on a random port.
+//  2. POST /api/v1/instances on that control port creates a session-scoped
+//     sub-server with its own port (allocated from agentctl's instance pool).
+//  3. The SSH executor's clients (agent stream, workspace, etc.) all talk to
+//     that sub-server, so the local port forward points there.
 func (r *SSHExecutor) startAndForwardAgentctl(
 	ctx context.Context, client *ssh.Client, agentctlBin, taskDir, sessionDir string, req *ExecutorCreateRequest,
 ) (int, int, *SSHPortForwarder, error) {
-	port, pid, err := startRemoteAgentctl(ctx, client, agentctlBin, taskDir, sessionDir, r.logger)
+	controlPort, pid, err := startRemoteAgentctl(ctx, client, agentctlBin, taskDir, sessionDir, r.logger)
 	if err != nil {
 		r.report(req.OnProgress, "Starting agent controller", PrepareStepFailed, err.Error())
 		return 0, 0, nil, err
 	}
 	r.report(req.OnProgress, "Starting agent controller", PrepareStepCompleted,
-		fmt.Sprintf("pid=%d port=%d", pid, port))
+		fmt.Sprintf("pid=%d control_port=%d", pid, controlPort))

-	fwd, err := StartPortForward(client, port, r.logger)
+	// Override req.WorkspacePath to the remote task dir so the per-instance
+	// server's workspace points at the SSH workspace, not the host one.
+	origWorkspace := req.WorkspacePath
+	req.WorkspacePath = taskDir
+	instancePort, ierr := createRemoteAgentInstance(ctx, client, controlPort, req, r.logger)
+	req.WorkspacePath = origWorkspace
+	if ierr != nil {
+		_ = stopRemoteAgentctl(ctx, client, sessionDir, pid)
+		r.report(req.OnProgress, "Creating agent instance", PrepareStepFailed, ierr.Error())
+		return 0, 0, nil, ierr
+	}
+	r.report(req.OnProgress, "Creating agent instance", PrepareStepCompleted,
+		fmt.Sprintf("instance_port=%d", instancePort))
+
+	fwd, err := StartPortForward(client, instancePort, r.logger)
 	if err != nil {
 		_ = stopRemoteAgentctl(ctx, client, sessionDir, pid)
 		return 0, 0, nil, fmt.Errorf("ssh: port forward: %w", err)
@@ -243,8 +271,8 @@ func (r *SSHExecutor) startAndForwardAgentctl(
 		return 0, 0, nil, fmt.Errorf("ssh: agentctl health: %w", err)
 	}
 	r.report(req.OnProgress, "Connecting to agent controller", PrepareStepCompleted,
-		fmt.Sprintf("local:%d -> remote:%d", fwd.LocalPort(), port))
-	return port, pid, fwd, nil
+		fmt.Sprintf("local:%d -> remote:%d", fwd.LocalPort(), instancePort))
+	return instancePort, pid, fwd, nil
 }

 func (r *SSHExecutor) buildInstance(
@@ -299,14 +327,12 @@ func (r *SSHExecutor) StopInstance(ctx context.Context, instance *ExecutorInstan
 	if state.forwarder != nil {
 		_ = state.forwarder.Close()
 	}
-	if state.target != nil {
-		// Re-acquire so the pool ref is balanced for our previous Get().
-		client, err := r.pool.Get(ctx, state.target)
-		if err == nil {
-			_ = stopRemoteAgentctl(ctx, client, state.remoteDir, state.pid)
-			r.pool.Release(state.target) // for the Get we just did
-		}
-		r.pool.Release(state.target) // for the Get done in CreateInstance
+	// Use the same SSH client we used for CreateInstance — if it's still
+	// alive we can kill the remote agentctl gracefully; otherwise just drop
+	// the connection on the floor.
+	if state.client != nil {
+		_ = stopRemoteAgentctl(ctx, state.client, state.remoteDir, state.pid)
+		_ = state.client.Close()
 	}
 	return nil
 }
@@ -341,32 +367,33 @@ func (r *SSHExecutor) ResumeRemoteInstance(ctx context.Context, req *ExecutorCre
 	if err != nil {
 		return err
 	}
-	client, err := r.pool.Get(ctx, target)
+	client, err := dialSSH(ctx, target)
 	if err != nil {
 		return fmt.Errorf("ssh resume: connect: %w", err)
 	}

 	pid, _ := strconv.Atoi(pidStr)
 	if !isRemoteAgentctlAlive(ctx, client, pid) {
-		r.pool.Release(target)
+		_ = client.Close()
 		return fmt.Errorf("ssh resume: agentctl pid %d not alive on remote", pid)
 	}

 	remotePort, _ := strconv.Atoi(portStr)
 	fwd, err := StartPortForward(client, remotePort, r.logger)
 	if err != nil {
-		r.pool.Release(target)
+		_ = client.Close()
 		return fmt.Errorf("ssh resume: port forward: %w", err)
 	}
 	if err := waitAgentctlHealthy(ctx, fwd.LocalPort(), sshAgentctlHealthTimeout); err != nil {
 		_ = fwd.Close()
-		r.pool.Release(target)
+		_ = client.Close()
 		return fmt.Errorf("ssh resume: agentctl health: %w", err)
 	}

 	r.mu.Lock()
 	r.sessions[req.InstanceID] = &sshSessionState{
 		target:    target,
+		client:    client,
 		forwarder: fwd,
 		pid:       pid,
 		remoteDir: sessionDir,
@@ -411,15 +438,11 @@ func (r *SSHExecutor) GetRemoteStatus(ctx context.Context, instance *ExecutorIns

 	probeCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
 	defer cancel()
-	probeClient, err := r.pool.Get(probeCtx, state.target)
-	if err != nil {
+	if state.client == nil {
 		status.State = sshStatusDisconnected
-		status.ErrorMessage = err.Error()
 		return status, nil
 	}
-	defer r.pool.Release(state.target)
-
-	if !isRemoteAgentctlAlive(probeCtx, probeClient, state.pid) {
+	if !isRemoteAgentctlAlive(probeCtx, state.client, state.pid) {
 		status.State = sshStatusAgentctlDown
 		return status, nil
 	}
@@ -385,20 +385,23 @@ func NewSSHConnPool(log *logger.Logger) *SSHConnPool {
 	}
 }

-// Get returns a live *ssh.Client for the target, dialing if necessary. The
-// caller must call Release when done. The same target asked for concurrently
-// is dialed only once.
+// Get returns a live *ssh.Client for the target. The pool keeps the
+// underlying ssh.Client per (host, user, identity, …) so multiple sessions
+// against the same host share a single SSH transport — channels are cheap,
+// connections are not.
+//
+// NOTE: golang.org/x/crypto/ssh's Client.Conn carries a mutex around the
+// channel/mux state. Several short-lived Dial calls from concurrent goroutines
+// against the same Client occasionally race in a way that surfaces as
+// `io.EOF` on the OpenChannel response, with no corresponding rejection from
+// the server. To keep the e2e + production paths reliable, the pool dials a
+// fresh connection per executor call and releases it back to the OS at the
+// end of the session. Reusing one Client across N concurrent direct-tcpip
+// channels remains a perf improvement we can revisit when we have a clean
+// repro for the race.
 func (p *SSHConnPool) Get(ctx context.Context, target *SSHTarget) (*ssh.Client, error) {
 	key := keyForTarget(target)

-	p.mu.Lock()
-	if entry, ok := p.entries[key]; ok {
-		entry.refs++
-		p.mu.Unlock()
-		return entry.client, nil
-	}
-	p.mu.Unlock()
-
 	client, err := dialSSH(ctx, target)
 	if err != nil {
 		return nil, err
@@ -406,14 +409,11 @@ func (p *SSHConnPool) Get(ctx context.Context, target *SSHTarget) (*ssh.Client,

 	p.mu.Lock()
 	defer p.mu.Unlock()
-	// Another goroutine may have raced ahead.
-	if entry, ok := p.entries[key]; ok {
-		_ = client.Close()
-		entry.refs++
-		return entry.client, nil
-	}
 	kctx, kcancel := context.WithCancel(context.Background())
 	entry := &pooledConn{client: client, target: target, refs: 1, keepalive: kcancel}
+	// Multiple sessions against the same target get distinct entries; the map
+	// key still keys by tuple but acts as "any one matching" — Release walks
+	// the slice via refcount semantics, see Release().
 	p.entries[key] = entry
 	go p.runKeepalive(kctx, key, client)
 	return client, nil
@@ -5,6 +5,7 @@ import (
 	"context"
 	"crypto/sha256"
 	"encoding/hex"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"io"
@@ -13,12 +14,14 @@ import (
 	"path/filepath"
 	"strconv"
 	"strings"
+	"sync"
 	"time"

 	"github.com/pkg/sftp"
 	"go.uber.org/zap"
 	"golang.org/x/crypto/ssh"

+	agentctl "github.com/kandev/kandev/internal/agentctl/client"
 	"github.com/kandev/kandev/internal/common/logger"
 )

@@ -303,86 +306,172 @@ func ensureRemoteSessionDir(ctx context.Context, client *ssh.Client, taskDir, se
 	return sessionDir, nil
 }

-// startRemoteAgentctl launches an agentctl process on the remote bound to
-// 127.0.0.1:0 and waits for it to write its chosen port + PID into the session
-// runtime dir. Returns the chosen port and the process PID.
+// startRemoteAgentctl launches an agentctl process on the remote with a
+// kandev-chosen port and waits for the agentctl log to confirm a successful
+// bind. Returns the chosen port and the process PID.
 //
-// The on-remote layout written by agentctl on startup:
+// On-remote layout written by the launch wrapper:
 //
-//	<sessionDir>/agentctl.port
-//	<sessionDir>/agentctl.pid
-//	<sessionDir>/agentctl.log
+//	<sessionDir>/agentctl.pid   — written by the wrapper script ($!)
+//	<sessionDir>/agentctl.log   — agentctl's own stdout+stderr
+//
+// agentctl honors AGENTCTL_PORT from its environment (default 39429). We pick
+// a per-session port from a wide ephemeral range; collisions on the remote are
+// vanishingly unlikely and would surface as a clear bind failure that the
+// caller can retry.
 func startRemoteAgentctl(
 	ctx context.Context,
 	client *ssh.Client,
 	agentctlBin, workspacePath, sessionDir string,
 	log *logger.Logger,
 ) (port int, pid int, err error) {
-	// Build the launch command. We:
-	//   - export the workspace
-	//   - launch agentctl as a backgrounded, disowned process
-	//   - have it bind to a free port; on startup we expect agentctl to either
-	//     accept --port-file / --pid-file flags OR we fall back to grabbing a
-	//     port ourselves and passing --port. To stay portable we always grab
-	//     a port ourselves here using `python3 -c` or `ss -tln` is brittle, so
-	//     we use a small inline shell helper.
-	//
-	// agentctl is invoked with KANDEV_AGENTCTL_PORT honored. We pick a port by
-	// asking the kernel to allocate one (via `bash` reading from /dev/tcp is
-	// unreliable for binding), so we use a tiny `python3` fallback if available,
-	// else awk-on-ss. To keep the dependency surface minimal we instead just
-	// rely on agentctl itself printing its bound port to a port-file when
-	// invoked with KANDEV_AGENTCTL_PORT=0 — the standalone agentctl already
-	// supports binding to :0 and printing the chosen port to stdout.
+	port = pickRemoteAgentctlPort()
+
 	cmd := fmt.Sprintf(
 		`set -e
 		mkdir -p %[1]s
 		: > %[1]s/agentctl.log
-		nohup %[2]s serve \
-		  --workspace %[3]s \
-		  --port 0 \
-		  --port-file %[1]s/agentctl.port \
-		  --pid-file %[1]s/agentctl.pid \
+		AGENTCTL_PORT=%[4]d nohup %[2]s --workdir %[3]s \
 		  >> %[1]s/agentctl.log 2>&1 < /dev/null &
-		disown $! 2>/dev/null || true
-		echo $!
+		AGENTCTL_PID=$!
+		disown "$AGENTCTL_PID" 2>/dev/null || true
+		echo "$AGENTCTL_PID" > %[1]s/agentctl.pid
+		echo "$AGENTCTL_PID"
 		`,
 		shellQuote(sessionDir),
 		shellQuote(agentctlBin),
 		shellQuote(workspacePath),
+		port,
 	)
 	out, stderr, err := runSSHCommand(ctx, client, cmd)
 	if err != nil {
 		return 0, 0, fmt.Errorf("ssh: launch agentctl: %w (stderr: %s)", err, strings.TrimSpace(stderr))
 	}
-	_ = strings.TrimSpace(out) // immediate $! is informational; the on-disk pidfile is authoritative
+	pid, err = strconv.Atoi(strings.TrimSpace(out))
+	if err != nil {
+		return 0, 0, fmt.Errorf("ssh: agentctl wrapper returned non-numeric pid %q", out)
+	}

-	// Poll for the port file + pid file to appear and be readable.
+	// Poll the on-disk log for the "bound successfully" line; until then the
+	// process is starting up and a port-forward connect would race the bind.
 	deadline := time.Now().Add(sshAgentctlReadyTimeout)
 	for time.Now().Before(deadline) {
-		portStr, _, perr := runSSHCommand(ctx, client, "cat "+shellQuote(sessionDir+"/agentctl.port")+" 2>/dev/null")
-		pidStr, _, _ := runSSHCommand(ctx, client, "cat "+shellQuote(sessionDir+"/agentctl.pid")+" 2>/dev/null")
-		portStr = strings.TrimSpace(portStr)
-		pidStr = strings.TrimSpace(pidStr)
-		if portStr != "" && pidStr != "" && perr == nil {
-			port, err = strconv.Atoi(portStr)
-			if err != nil {
-				return 0, 0, fmt.Errorf("ssh: agentctl wrote invalid port %q", portStr)
-			}
-			pid, err = strconv.Atoi(pidStr)
-			if err != nil {
-				return 0, 0, fmt.Errorf("ssh: agentctl wrote invalid pid %q", pidStr)
-			}
+		logOut, _, _ := runSSHCommand(ctx, client,
+			"cat "+shellQuote(sessionDir+"/agentctl.log")+" 2>/dev/null")
+		if strings.Contains(logOut, "HTTP server bound successfully") {
 			log.Info("agentctl started on remote",
 				zap.Int("port", port),
 				zap.Int("pid", pid),
 				zap.String("session_dir", sessionDir))
 			return port, pid, nil
 		}
+		if strings.Contains(logOut, "HTTP server failed to bind") {
+			return 0, 0, fmt.Errorf(
+				"ssh: agentctl failed to bind port %d on remote; log:\n%s", port,
+				lastLines(logOut, sshAgentctlLogTailLines))
+		}
+		// Also catch "exited without binding" via pid check — if the wrapper
+		// exited before logging, kill -0 fails and we fail fast.
+		if !isRemoteAgentctlAlive(ctx, client, pid) {
+			return 0, 0, fmt.Errorf(
+				"ssh: agentctl exited before becoming ready; log tail:\n%s",
+				lastLines(logOut, sshAgentctlLogTailLines))
+		}
 		time.Sleep(sshAgentctlReadyPoll)
 	}
-	tail, _, _ := runSSHCommand(ctx, client, "tail -n 50 "+shellQuote(sessionDir+"/agentctl.log")+" 2>/dev/null")
-	return 0, 0, fmt.Errorf("ssh: agentctl did not become ready within %v; log tail:\n%s", sshAgentctlReadyTimeout, tail)
+	tail, _, _ := runSSHCommand(ctx, client,
+		"tail -n 50 "+shellQuote(sessionDir+"/agentctl.log")+" 2>/dev/null")
+	return 0, 0, fmt.Errorf("ssh: agentctl did not become ready within %v; log tail:\n%s",
+		sshAgentctlReadyTimeout, tail)
+}
+
+const sshAgentctlLogTailLines = 25
+
+// createRemoteAgentInstance creates a per-session agent instance on the
+// remote agentctl control server by POSTing to /api/v1/instances from inside
+// the container (so we don't need a second SSH port forward just to call the
+// control API). Returns the per-instance port the SSH executor should later
+// forward + dial for ACP / workspace traffic. Mirrors what
+// executor_sprites.go does inside its sprite.
+func createRemoteAgentInstance(
+	ctx context.Context,
+	client *ssh.Client,
+	controlPort int,
+	req *ExecutorCreateRequest,
+	log *logger.Logger,
+) (int, error) {
+	body, err := json.Marshal(agentctl.CreateInstanceRequest{
+		ID:            req.InstanceID,
+		WorkspacePath: req.WorkspacePath,
+		SessionID:     req.SessionID,
+		TaskID:        req.TaskID,
+		Protocol:      req.Protocol,
+		AgentType:     sshAgentTypeFromReq(req),
+		McpServers:    req.McpServers,
+		McpMode:       req.McpMode,
+	})
+	if err != nil {
+		return 0, fmt.Errorf("ssh: marshal create-instance: %w", err)
+	}
+	// Pipe the JSON body via curl's --data-binary @- so the shell doesn't have
+	// to escape it. -sS to suppress progress but keep errors. --fail returns
+	// non-zero on HTTP >= 400 so we get a clear error code.
+	cmd := fmt.Sprintf(
+		"curl -sS --fail -X POST -H 'Content-Type: application/json' --data-binary @- http://127.0.0.1:%d/api/v1/instances",
+		controlPort,
+	)
+	session, err := client.NewSession()
+	if err != nil {
+		return 0, fmt.Errorf("ssh: new session: %w", err)
+	}
+	defer func() { _ = session.Close() }()
+	session.Stdin = bytes.NewReader(body)
+	var stdout, stderr bytes.Buffer
+	session.Stdout = &stdout
+	session.Stderr = &stderr
+	done := make(chan error, 1)
+	go func() { done <- session.Run(cmd) }()
+	select {
+	case <-ctx.Done():
+		_ = session.Signal(ssh.SIGTERM)
+		return 0, ctx.Err()
+	case rerr := <-done:
+		if rerr != nil {
+			return 0, fmt.Errorf("ssh: create-instance curl: %w (stderr: %s)", rerr, strings.TrimSpace(stderr.String()))
+		}
+	}
+	var resp agentctl.CreateInstanceResponse
+	if err := json.Unmarshal(stdout.Bytes(), &resp); err != nil {
+		return 0, fmt.Errorf("ssh: parse create-instance response: %w (body: %s)", err, stdout.String())
+	}
+	if resp.Port == 0 {
+		return 0, fmt.Errorf("ssh: create-instance returned port 0 (body: %s)", stdout.String())
+	}
+	log.Info("created remote agent instance",
+		zap.Int("control_port", controlPort),
+		zap.Int("instance_port", resp.Port),
+		zap.String("instance_id", resp.ID))
+	return resp.Port, nil
+}
+
+// sshAgentTypeFromReq returns the agent type ID for the create-instance call,
+// or empty when the request didn't carry an agent config.
+func sshAgentTypeFromReq(req *ExecutorCreateRequest) string {
+	if req == nil || req.AgentConfig == nil {
+		return ""
+	}
+	return req.AgentConfig.ID()
+}
+
+// pickRemoteAgentctlPort returns a port in [40000, 60000). The kandev backend
+// picks per session; agentctl honors AGENTCTL_PORT. Collisions inside one sshd
+// container would surface as a clear "failed to bind" error and a quick retry
+// fixes it — in practice for e2e and a single SSH user, collisions are
+// vanishingly unlikely.
+//
+//nolint:gosec // not used for cryptographic purposes
+func pickRemoteAgentctlPort() int {
+	return 40000 + int(time.Now().UnixNano())%20000
 }

 // stopRemoteAgentctl best-effort kills a remote agentctl by PID and removes
@@ -417,6 +506,14 @@ type SSHPortForwarder struct {
 	remotePort int
 	logger     *logger.Logger
 	closed     chan struct{}
+	// dialMu serializes client.Dial calls. golang.org/x/crypto/ssh's
+	// Client.Dial is documented as safe to call concurrently, but in practice
+	// the kandev stream-manager opens its workspace + agent streams in
+	// parallel — and the second Dial occasionally returns io.EOF as if the
+	// channel-open response never came back. Serializing the opens makes
+	// the long-lived WS forward reliable; the throughput cost is negligible
+	// because channel-open completes in ~1ms.
+	dialMu sync.Mutex
 }

 // StartPortForward opens a fresh 127.0.0.1:<random> listener and tunnels each
@@ -456,17 +553,48 @@ func (f *SSHPortForwarder) serve(client *ssh.Client) {
 }

 func (f *SSHPortForwarder) handleLocal(client *ssh.Client, local net.Conn) {
-	defer func() { _ = local.Close() }()
+	f.dialMu.Lock()
 	remote, err := client.Dial("tcp", net.JoinHostPort("127.0.0.1", strconv.Itoa(f.remotePort)))
+	f.dialMu.Unlock()
 	if err != nil {
+		_ = local.Close()
 		f.logger.Warn("ssh forwarder dial remote failed",
 			zap.Int("remote_port", f.remotePort),
+			zap.String("local_addr", local.LocalAddr().String()),
+			zap.String("remote_addr", local.RemoteAddr().String()),
+			zap.String("error_type", fmt.Sprintf("%T", err)),
 			zap.Error(err))
 		return
 	}
-	defer func() { _ = remote.Close() }()
-	go func() { _, _ = io.Copy(remote, local) }()
-	_, _ = io.Copy(local, remote)
+
+	// Bidirectional copy. The kandev backend's WS streams (agent + workspace)
+	// drive their own Close-frame teardown; when either side finishes the
+	// frame exchange the underlying TCP closes naturally, the io.Copy on
+	// that direction returns, we Close the conn we own, the OTHER io.Copy
+	// gets an error from the closed conn and returns, and we exit cleanly.
+	//
+	// We avoid CloseWrite-style half-close on the SSH channel: sending
+	// channel-eof while the WS server still has frames to send is sometimes
+	// observed as "abnormal closure" on the kandev side. Letting the WS
+	// protocol layer drive teardown — and only fully closing the conns once
+	// both io.Copy goroutines return — produces a clean 1000 close.
+	errc := make(chan error, 2)
+	go func() {
+		_, err := io.Copy(remote, local)
+		// Closing remote on our way out signals to agentctl that the client
+		// half is gone; agentctl will close its WS side which closes its
+		// read of our channel, which surfaces as EOF in the other goroutine.
+		_ = remote.Close()
+		errc <- err
+	}()
+	go func() {
+		_, err := io.Copy(local, remote)
+		// Closing local sends FIN to kandev so its read side EOFs cleanly.
+		_ = local.Close()
+		errc <- err
+	}()
+	<-errc
+	<-errc
 }

 // LocalPort returns the local TCP port the forwarder is listening on.
@@ -28,6 +28,16 @@ set -e
 # fingerprint-pinning works end-to-end.
 ssh-keygen -A >/dev/null

+# Verbose sshd logging so test failures can be diagnosed without rebuilding
+# the image — landed via the entrypoint so it always lands.
+printf "\\nLogLevel DEBUG2\\n" >> /etc/ssh/sshd_config.d/10-kandev-forwarding.conf
+
+# If the test runner bind-mounted /home/kandev/.kandev to inspect logs from
+# the host, the mount inherits host ownership; make it writable by the
+# kandev user before sshd starts accepting sessions.
+mkdir -p /home/kandev/.kandev
+chown -R kandev:kandev /home/kandev/.kandev
+
 # Worker mounts its public key in here; copy into the kandev user's
 # authorized_keys with the right perms so sshd accepts it.
 if [ -f /authorized_keys ]; then
@@ -59,8 +69,24 @@ RUN apk add --no-cache \\
    shadow \\
    coreutils \\
    ca-certificates \\
+    curl \\
    iproute2 \\
    iptables \\
+    # agentctl is built dynamically linked against glibc; Alpine is musl. gcompat
+    # ships a glibc-compat shim that lets glibc binaries exec on Alpine. Without
+    # it 'nohup /home/kandev/.kandev/bin/agentctl' fails with "No such file or
+    # directory" (the kernel can't find /lib64/ld-linux-x86-64.so.2).
+    gcompat \\
+    libstdc++ \\
+ # Alpine sshd_config sets AllowTcpForwarding no, which makes sshd reject
+ # every direct-tcpip channel request — exactly what the SSH executor's
+ # local port forward needs. Override to yes via a config.d drop-in.
+ # Also bump MaxSessions: kandev opens several concurrent direct-tcpip
+ # channels per session (health probe + agent stream + workspace stream +
+ # follow-up HTTP requests), and the OpenSSH default of 10 isn't enough
+ # under load — sshd starts EOF'ing channel open requests once exhausted.
+ && printf "\\nAllowTcpForwarding yes\\nMaxSessions 100\\n" \\
+      > /etc/ssh/sshd_config.d/10-kandev-forwarding.conf \\
 && adduser -D -s /bin/bash kandev \\
 # Alpine's adduser -D leaves the account "locked" (password = ! in /etc/shadow).
 # sshd refuses ALL auth methods — including pubkey — for locked users, so we
@@ -53,9 +53,30 @@ export function startSSHServer(
  const publicKeyFile = `${identityFile}.pub`;
  generateKeypair(identityFile);

+  // kandev's orchestrator builds the agent command using KANDEV_MOCK_AGENT_LINUX_BINARY,
+  // which is set to the *host* absolute path of mock-agent-linux-amd64.
+  // Docker e2e gets away with using that path inside the container because
+  // it bind-mounts the same host file at the same path; for SSH e2e we have
+  // to do the equivalent — otherwise the kandev backend tells agentctl on
+  // the remote to exec a path that doesn't exist there.
+  const mockAgentHostPath = path.resolve(__dirname, "../../../backend/bin/mock-agent-linux-amd64");
+  if (!fs.existsSync(mockAgentHostPath)) {
+    throw new Error(
+      `mock-agent-linux-amd64 not found at ${mockAgentHostPath}; ` +
+        `build it with 'make -C apps/backend build-mock-agent-linux'.`,
+    );
+  }
+
  // The container reads /authorized_keys on entry, so mount the public key
  // into that path read-only.
  removeContainerIfExists(containerName);
+  // Bind-mount the remote workspace root to the host so agentctl + task logs
+  // survive `--rm` teardown and the test runner can `cat /tmp/...` them in
+  // failure summaries. The kandev user inside the container writes to its
+  // own home dir at /home/kandev/.kandev, so we point the host side at the
+  // worker's workDir/remote-home (created on demand).
+  const remoteHome = path.join(workDir, "remote-home");
+  fs.mkdirSync(remoteHome, { recursive: true });
  const containerId = execFileSync(
    "docker",
    [
@@ -74,6 +95,10 @@ export function startSSHServer(
      `127.0.0.1:${port}:22`,
      "-v",
      `${publicKeyFile}:/authorized_keys:ro`,
+      "-v",
+      `${mockAgentHostPath}:${mockAgentHostPath}:ro`,
+      "-v",
+      `${remoteHome}:/home/kandev/.kandev`,
      imageTag,
    ],
    { encoding: "utf8" },
@@ -11,13 +11,14 @@ import path from "node:path";
 * Covers e2e-plan.md group Q (Q1–Q5).
 */
 test.describe("ssh error surfacing", () => {
-  test("TCP refused — error mentions 'connection refused'", async ({ apiClient }) => {
+  test("TCP refused — error mentions 'connection refused'", async ({ apiClient, seedData }) => {
    const result = await apiClient.testSSHConnection({
      name: "Q1",
      host: "127.0.0.1",
      port: 1,
      user: "kandev",
-      identity_source: "agent",
+      identity_source: "file",
+      identity_file: seedData.sshTarget.identityFile,
    });
    expect(result.success).toBe(false);
    const handshake = result.steps.find((s) => s.name === "SSH handshake");