feat(executors): detect and tear down dead SSH session transports (#3581)

Co-authored-by: nova28 <17953305+nova28@users.noreply.github.com> Co-authored-by: Carlos Florencio <carlosmflorencio@gmail.com>
2026-09-13 06:00:53 +08:00
parent f150dd32d1
commit aebb01e714
21 changed files with 3312 additions and 90 deletions
@@ -39,11 +39,11 @@ require (
 	go.opentelemetry.io/otel/trace v1.44.0
 	go.uber.org/goleak v1.3.0
 	go.uber.org/zap v1.26.0
-	golang.org/x/crypto v0.52.0
-	golang.org/x/net v0.55.0
-	golang.org/x/sync v0.20.0
-	golang.org/x/sys v0.45.0
-	golang.org/x/text v0.37.0
+	golang.org/x/crypto v0.57.0
+	golang.org/x/net v0.58.0
+	golang.org/x/sync v0.23.0
+	golang.org/x/sys v0.48.0
+	golang.org/x/text v0.42.0
 	google.golang.org/grpc v1.83.1
 	google.golang.org/protobuf v1.36.11
 	gopkg.in/natefinch/lumberjack.v2 v2.2.1
@@ -133,7 +133,7 @@ require (
 	go.yaml.in/yaml/v3 v3.0.4 // indirect
 	golang.org/x/arch v0.3.0 // indirect
 	golang.org/x/oauth2 v0.36.0 // indirect
-	golang.org/x/term v0.43.0 // indirect
+	golang.org/x/term v0.46.0 // indirect
 	golang.org/x/time v0.11.0 // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20260526163538-3dc84a4a5aaa // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa // indirect
@@ -427,8 +427,8 @@ golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8U
 golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/crypto v0.0.0-20210421170649-83a5a9bb288b/go.mod h1:T9bdIzuCu7OtxOm1hfPfRQxPLYneinmdGuTeoZ9dtd4=
 golang.org/x/crypto v0.0.0-20220722155217-630584e8d5aa/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
-golang.org/x/crypto v0.52.0 h1:RMs7fP2rXdep0CftQlK8Uf+kibLm7qkCcradZWYz988=
-golang.org/x/crypto v0.52.0/go.mod h1:1QgfPxDqh0T2M/elOJtp9RvuR95kVjir0e6/BvEmGbc=
+golang.org/x/crypto v0.57.0 h1:3ZVCjf8Ggz7zneR/EHRVx68Ctf+2pmIMP2UFhh9cC6M=
+golang.org/x/crypto v0.57.0/go.mod h1:Fdz0i5U6CoizGwLda9DttjSk6qlZo25zYNtR+ycvuZA=
 golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
 golang.org/x/exp v0.0.0-20190306152737-a1d7652674e8/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
 golang.org/x/exp v0.0.0-20190510132918-efd6b22b2522/go.mod h1:ZjyILWgesfNpC6sMxTJOJm9Kp84zZh5NQWvqDGG3Qr8=
@@ -494,8 +494,8 @@ golang.org/x/net v0.0.0-20201209123823-ac852fbbde11/go.mod h1:m0MpNAwzfU5UDzcl9v
 golang.org/x/net v0.0.0-20201224014010-6772e930b67b/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
 golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
 golang.org/x/net v0.0.0-20211112202133-69e39bad7dc2/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
-golang.org/x/net v0.55.0 h1:bcvxaJn3e1U6InsFWt1JUq1aSjnRxLzT2rtD2KfkDF8=
-golang.org/x/net v0.55.0/go.mod h1:L5U2KuzuOe1lY7Z+aWVIKK6qEeJXnXV9yzGA+WCHJww=
+golang.org/x/net v0.58.0 h1:ynWG7rqYi4ccpTEuPZ2QGWHktVEM9DMCj9yzDE0Q7To=
+golang.org/x/net v0.58.0/go.mod h1:YwCddHnFlT7eLQqVprV19OnhLGtc5xOKgE0RyqgfWAU=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
 golang.org/x/oauth2 v0.0.0-20190226205417-e64efc72b421/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
 golang.org/x/oauth2 v0.0.0-20190604053449-0f29369cfe45/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
@@ -517,8 +517,8 @@ golang.org/x/sync v0.0.0-20200317015054-43a5402ce75a/go.mod h1:RxMgew5VJxzue5/jJ
 golang.org/x/sync v0.0.0-20200625203802-6e8e738ad208/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20201207232520-09787c993a3a/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
-golang.org/x/sync v0.20.0 h1:e0PTpb7pjO8GAtTs2dQ6jYa5BWYlMuX047Dco/pItO4=
-golang.org/x/sync v0.20.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0=
+golang.org/x/sync v0.23.0 h1:KameEIfc1IkluZyXWLn39Wd4tURc6GbCiISGiZm2bQk=
+golang.org/x/sync v0.23.0/go.mod h1:sUUOizhqBxiL6pEWpqNLUiaJn1ShEbZ6BBqskPbjZm0=
 golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190312061237-fead79001313/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
@@ -562,11 +562,11 @@ golang.org/x/sys v0.0.0-20220704084225-05e143d24a9e/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.0.0-20220908164124-27713097b956/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.8.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.45.0 h1:dO4czNzziLiiXplLQgBCEpCvXQ3dnkn0SdaZSYdQ+FY=
-golang.org/x/sys v0.45.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
+golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo=
+golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
-golang.org/x/term v0.43.0 h1:S4RLU2sB31O/NCl+zFN9Aru9A/Cq2aqKpTZJ6B+DwT4=
-golang.org/x/term v0.43.0/go.mod h1:lrhlHNdQJHO+1qVYiHfFKVuVioJIheAc3fBSMFYEIsk=
+golang.org/x/term v0.46.0 h1:3+OXuTbaKDgwk8jTi3aSLHRlmWqHEUDUtxnbFigO4YE=
+golang.org/x/term v0.46.0/go.mod h1:+K02xbkittuwc0Am4abfA3Fc+XRGXkvBXNO88NCXPoc=
 golang.org/x/text v0.0.0-20170915032832-14c0d48ead0c/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.1-0.20180807135948-17ff2d5776d2/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
@@ -575,8 +575,8 @@ golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.4/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
-golang.org/x/text v0.37.0 h1:Cqjiwd9eSg8e0QAkyCaQTNHFIIzWtidPahFWR83rTrc=
-golang.org/x/text v0.37.0/go.mod h1:a5sjxXGs9hsn/AJVwuElvCAo9v8QYLzvavO5z2PiM38=
+golang.org/x/text v0.42.0 h1:JbOZXgfeCPU9gacVtYliJqOhD+zhrEqK4LfdpmlUZqI=
+golang.org/x/text v0.42.0/go.mod h1:ojzP1Z+2QtioaF8DTtO8K5q7JWVVYwZKenzujK0Zd0E=
 golang.org/x/time v0.0.0-20181108054448-85acf8d2951c/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.0.0-20190308202827-9d24e82272b4/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.0.0-20191024005414-555d28b269f0/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
@@ -629,8 +629,8 @@ golang.org/x/tools v0.0.0-20201208233053-a543418bbed2/go.mod h1:emZCQorbCU4vsT4f
 golang.org/x/tools v0.0.0-20210105154028-b0ab187a4818/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
 golang.org/x/tools v0.0.0-20210108195828-e2f9c7f1fc8e/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
 golang.org/x/tools v0.1.0/go.mod h1:xkSsbof2nBLbhDlRMhhhyNLN/zl3eTqcnHD5viDpcZ0=
-golang.org/x/tools v0.44.0 h1:UP4ajHPIcuMjT1GqzDWRlalUEoY+uzoZKnhOjbIPD2c=
-golang.org/x/tools v0.44.0/go.mod h1:KA0AfVErSdxRZIsOVipbv3rQhVXTnlU6UhKxHd1seDI=
+golang.org/x/tools v0.49.0 h1:3NI7VXzL9+1WZD52Dx2ttoPwD5DWrFGpl9mFZDlmisI=
+golang.org/x/tools v0.49.0/go.mod h1:SJNXV9DBKT0UbdttsQjbfJlAE/q+y36++zo3uL3N0Oo=
 golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
@@ -203,6 +203,7 @@ const (
 	MetadataKeySpriteState              = "sprite_state"
 	MetadataKeySpriteCreatedAt          = "sprite_created_at"
 	MetadataKeyLocalPort                = "local_port"
+	MetadataKeyReuseExistingProcess     = "reuse_existing_process"

 	// Kubernetes executor connection metadata. These names mirror the persisted
 	// executor config parsed by internal/agent/kubernetes.
@@ -373,9 +373,9 @@ func (r *DockerExecutor) reconnectToContainer(ctx context.Context, dockerClient
 		ContainerIP:   containerIP,
 		WorkspacePath: dockerWorkspacePath,
 		Metadata: map[string]interface{}{
-			MetadataKeyIsRemote:      true,
-			MetadataKeyContainerID:   info.ID,
-			"reuse_existing_process": conn.reusingProcess,
+			MetadataKeyIsRemote:             true,
+			MetadataKeyContainerID:          info.ID,
+			MetadataKeyReuseExistingProcess: conn.reusingProcess,
 		},
 		AuthToken: refreshedAuthToken,
 	}, nil
@@ -577,12 +577,12 @@ func (r *SpritesExecutor) buildInstanceResult(
 			agentctl.WithSessionID(req.SessionID)),
 		WorkspacePath: spritesWorkspacePath,
 		Metadata: map[string]interface{}{
-			MetadataKeySpriteName:      spriteName,
-			MetadataKeySpriteState:     strings.TrimSpace(sprite.Status),
-			MetadataKeySpriteCreatedAt: sprite.CreatedAt,
-			MetadataKeyLocalPort:       localPort,
-			"reuse_existing_process":   reusingExisting,
-			MetadataKeyIsRemote:        true,
+			MetadataKeySpriteName:           spriteName,
+			MetadataKeySpriteState:          strings.TrimSpace(sprite.Status),
+			MetadataKeySpriteCreatedAt:      sprite.CreatedAt,
+			MetadataKeyLocalPort:            localPort,
+			MetadataKeyReuseExistingProcess: reusingExisting,
+			MetadataKeyIsRemote:             true,
 		},
 	}
 }
@@ -24,16 +24,22 @@ import (
 )

 const (
-	sshAgentctlHealthTimeout  = 20 * time.Second
-	sshAgentctlCleanupTimeout = 20 * time.Second
-	sshAgentctlLoopbackHost   = "127.0.0.1"
+	sshAgentctlHealthTimeout = 20 * time.Second
+	sshAgentctlLoopbackHost  = "127.0.0.1"

 	sshStatusRunning      = "running"
 	sshStatusUnknown      = "unknown"
 	sshStatusDisconnected = "disconnected"
 	sshStatusAgentctlDown = "agentctl-down"
+
+	sshTransportLostMessage = "ssh session transport lost"
 )

+// sshAgentctlCleanupTimeout bounds the remote controller stop a disposal
+// path issues. A package-level variable, not a constant, so a backstop test
+// need not wait out real seconds (see executor_ssh_keepalive.go).
+var sshAgentctlCleanupTimeout = 20 * time.Second
+
 // sshSessionState tracks per-session resources we need to clean up later:
 // 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
@@ -53,6 +59,34 @@ type sshSessionState struct {
 	prepareEnv       map[string]string
 	platform         SSHRemotePlatform
 	runtimeAPITunnel *sshRuntimeAPITunnel
+
+	// port is the remote agentctl instance's port (the one the local forward
+	// targets) and workdirRoot is the resolved workdir root used to launch
+	// it. Both are set once at session creation and read back by a caller
+	// that loses the create/resume race (AC-EXECUTORS-SSH-TRANSPORT-LIVENESS-002.1):
+	// its own request never produced this session, so its instance metadata
+	// must come from here, not from re-deriving values the surviving session
+	// may not actually have.
+	port        int
+	workdirRoot string
+
+	// Transport liveness (docs/specs/executors/requirements/ssh-transport-liveness.md).
+	// watchdog, transportLost and lastProbeReply are guarded by the owning
+	// SSHExecutor's mutex, the same lock that guards the session map.
+	watchdog       *sshKeepaliveWatchdog
+	transportLost  bool
+	lastProbeReply time.Time
+
+	// Independent once-guards for the two handles teardown and every
+	// disposal path share: whichever reaches a handle first closes it
+	// exactly once, and the other observes the same result.
+	forwarderCloseOnce sync.Once
+	forwarderCloseErr  error
+	clientCloseOnce    sync.Once
+	clientCloseErr     error
+	// teardownOnce guards transport teardown as a whole, so a second
+	// declaration logs no further warning.
+	teardownOnce sync.Once
 }

 // SSHExecutor implements ExecutorBackend for SSH-reachable Linux and macOS hosts.
@@ -120,13 +154,11 @@ func (r *SSHExecutor) Close() error {
 	}
 	r.mu.Unlock()
 	for _, s := range states {
+		r.stopWatchdogAndWait(s)
+		_ = r.closeForwarderOnce(s)
+		_ = r.closeClientOnce(s)
 		_ = s.runtimeAPITunnel.Close()
-		if s.forwarder != nil {
-			_ = s.forwarder.Close()
-		}
-		if s.client != nil {
-			_ = r.closeSSHClient(s.client)
-		}
+		r.awaitProbeExit(s)
 	}
 	return nil
 }
@@ -203,6 +235,12 @@ func (r *SSHExecutor) CreateInstance(ctx context.Context, req *ExecutorCreateReq
 	}
 	resumed, ok := r.resumedStateForCreate(req)
 	if ok {
+		r.mu.Lock()
+		lost := resumed.transportLost
+		r.mu.Unlock()
+		if lost {
+			return nil, fmt.Errorf("ssh: reused session unavailable: %w", ErrSSHTransportLost)
+		}
 		return r.buildResumedInstance(req, resumed), nil
 	}
 	if _, err := validateRemoteContributions(req.RemoteContributions); err != nil {
@@ -224,8 +262,8 @@ func (r *SSHExecutor) CreateInstance(ctx context.Context, req *ExecutorCreateReq
 	released := false
 	defer func() {
 		if !released {
-			_ = runtimeAPITunnel.Close()
 			_ = client.Close()
+			_ = runtimeAPITunnel.Close()
 		}
 	}()
 	r.report(req.OnProgress, "Connecting to SSH host", PrepareStepCompleted, "")
@@ -286,7 +324,24 @@ func (r *SSHExecutor) CreateInstance(ctx context.Context, req *ExecutorCreateReq
 	}

 	r.mu.Lock()
-	r.sessions[req.InstanceID] = &sshSessionState{
+	if existing, ok := r.sessions[req.InstanceID]; ok {
+		// AC-EXECUTORS-SSH-TRANSPORT-LIVENESS-002.1: another caller already
+		// recorded a session for this instance while we were dialing and
+		// preparing. Leave it and its watchdog alone, close what we just
+		// built instead of orphaning it, and proceed with the existing
+		// session on the same terms as any other reuse.
+		lost := existing.transportLost
+		r.mu.Unlock()
+		_ = fwd.Close()
+		_ = client.Close()
+		_ = runtimeAPITunnel.Close()
+		released = true
+		if lost {
+			return nil, fmt.Errorf("ssh: reused session unavailable: %w", ErrSSHTransportLost)
+		}
+		return r.buildInstanceForLostRace(req, existing), nil
+	}
+	state := &sshSessionState{
 		target:           target,
 		client:           client,
 		forwarder:        fwd,
@@ -298,13 +353,61 @@ func (r *SSHExecutor) CreateInstance(ctx context.Context, req *ExecutorCreateReq
 		prepareEnv:       sshRemoteContributionEnv(req, agentctlBin),
 		platform:         platform,
 		runtimeAPITunnel: runtimeAPITunnel,
+		port:             port,
+		workdirRoot:      workdir,
 	}
+	r.sessions[req.InstanceID] = state
+	r.startWatchdogLocked(req.InstanceID, state)
 	r.mu.Unlock()
 	released = true // ownership transferred to session state; released on StopInstance

 	return r.buildInstance(req, target, fwd, taskDir, sessionDir, port, pid, workdir, authToken), nil
 }

+// buildInstanceForLostRace builds an ExecutorInstance for a caller that lost
+// the create/resume race in the executor's tracked-sessions map: another
+// concurrent caller already recorded (and started a watchdog for) the
+// session this caller was about to create. It reads only from the existing
+// session state, never from this caller's request, since the request's
+// remote-agentctl metadata describes a session this caller never created.
+func (r *SSHExecutor) buildInstanceForLostRace(req *ExecutorCreateRequest, existing *sshSessionState) *ExecutorInstance {
+	client := existing.agentctlClient
+	if client == nil {
+		client = agentctl.NewClient(sshAgentctlLoopbackHost, existing.forwarder.LocalPort(), r.logger,
+			agentctl.WithExecutionID(req.InstanceID),
+			agentctl.WithSessionID(req.SessionID), agentctl.WithAuthToken(existing.authToken))
+	}
+	metadata := map[string]interface{}{
+		MetadataKeySSHHost:               existing.target.Host,
+		MetadataKeySSHPort:               strconv.Itoa(existing.target.Port),
+		MetadataKeySSHUser:               existing.target.User,
+		MetadataKeySSHHostFingerprint:    existing.target.PinnedFingerprint,
+		MetadataKeySSHRemoteTaskDir:      existing.remoteTaskDir,
+		MetadataKeySSHRemoteSessionDir:   existing.remoteDir,
+		MetadataKeySSHRemoteAgentctlPort: strconv.Itoa(existing.port),
+		MetadataKeySSHRemoteAgentctlPID:  strconv.Itoa(existing.pid),
+		MetadataKeySSHLocalForwardPort:   strconv.Itoa(existing.forwarder.LocalPort()),
+		MetadataKeySSHWorkdirRoot:        existing.workdirRoot,
+		MetadataKeyIsRemote:              true,
+		MetadataKeyReuseExistingProcess:  existing.reusingProcess,
+	}
+	// existing.metadata is the winning session's own cloned request metadata
+	// (cloneSSHMetadata(req.Metadata), captured after openSSHRuntimeAPITunnelForRequest
+	// ran for the winner), so it carries the winner's still-open tunnel's keys —
+	// never this caller's own, since this caller's tunnel was just closed above.
+	copySSHRuntimeAPIMetadata(metadata, existing.metadata)
+	return &ExecutorInstance{
+		InstanceID:    req.InstanceID,
+		TaskID:        req.TaskID,
+		SessionID:     req.SessionID,
+		RuntimeName:   r.Name(),
+		Client:        client,
+		WorkspacePath: existing.remoteTaskDir,
+		AuthToken:     existing.authToken,
+		Metadata:      metadata,
+	}
+}
+
 func reuseRequiredRemoteTaskDir(req *ExecutorCreateRequest) (string, error) {
 	if req == nil {
 		return "", fmt.Errorf("%w: missing remote task directory", models.ErrWorkspaceReuseUnsafe)
@@ -602,7 +705,7 @@ func (r *SSHExecutor) buildResumedInstance(req *ExecutorCreateRequest, state *ss
 		MetadataKeySSHLocalForwardPort:   strconv.Itoa(state.forwarder.LocalPort()),
 		MetadataKeySSHWorkdirRoot:        workdir,
 		MetadataKeyIsRemote:              true,
-		"reuse_existing_process":         state.reusingProcess,
+		MetadataKeyReuseExistingProcess:  state.reusingProcess,
 	}
 	copySSHRuntimeAPIMetadata(metadata, req.Metadata)
 	return &ExecutorInstance{
@@ -641,18 +744,31 @@ func (r *SSHExecutor) StopInstance(ctx context.Context, instance *ExecutorInstan
 	}
 	r.mu.Lock()
 	state := r.sessions[instance.InstanceID]
-	delete(r.sessions, instance.InstanceID)
+	var classification sshTransportClassification
+	if state != nil {
+		// AC-EXECUTORS-SSH-TRANSPORT-LIVENESS-002.8: the single disposal
+		// reading, taken in the same critical section as the removal, before
+		// any remote command on this path.
+		classification = classifySSHTransportLocked(state)
+		delete(r.sessions, instance.InstanceID)
+	}
 	r.mu.Unlock()
 	if state == nil {
 		return r.stopPersistedRemoteAgentctl(ctx, instance, force)
 	}
-	_ = state.runtimeAPITunnel.Close()
-	if state.forwarder != nil {
-		_ = state.forwarder.Close()
-	}
-	if shouldRunExecutorCleanup(instance.StopReason) && state.client != nil && r.cleanupScript != nil {
+
+	// Step 2: signal the watchdog loop and wait for it — only it — to exit.
+	r.stopWatchdogAndWait(state)
+	// Step 3: close the local port forward through its once-guard.
+	_ = r.closeForwarderOnce(state)
+
+	// Step 4: this path's own remote commands, skipped entirely when the
+	// reading said the transport is lost or unresponsive.
+	if classification == sshTransportAnswering && shouldRunExecutorCleanup(instance.StopReason) && state.client != nil && r.cleanupScript != nil {
 		metadata := mergeSSHMetadata(state.metadata, instance.Metadata)
-		if err := r.cleanupScript(ctx, state.client, state.remoteTaskDir, metadata, state.prepareEnv, state.platform, instance.InstanceID, instance.StopReason); err != nil {
+		if err := r.runWithTransportBackstop(state, sshCleanupTimeout, func() error {
+			return r.cleanupScript(ctx, state.client, state.remoteTaskDir, metadata, state.prepareEnv, state.platform, instance.InstanceID, instance.StopReason)
+		}); err != nil {
 			r.logger.Warn("failed to run SSH cleanup script", zap.String("instance_id", instance.InstanceID), zap.Error(err))
 		}
 	}
@@ -660,25 +776,34 @@ func (r *SSHExecutor) StopInstance(ctx context.Context, instance *ExecutorInstan
 		r.logger.Info("preserving SSH agentctl after agent stop",
 			zap.String("instance_id", instance.InstanceID),
 			zap.String("stop_reason", instance.StopReason))
-		if state.client != nil {
-			_ = r.closeSSHClient(state.client)
-		}
+		// Step 5 and step 6: close the client and wait for the prober exit
+		// even on this early-return path, so no watchdog activity survives
+		// the call.
+		_ = r.closeClientOnce(state)
+		_ = state.runtimeAPITunnel.Close()
+		r.awaitProbeExit(state)
 		return nil
 	}

 	// 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 {
-		cleanupCtx, cancel := sshRemoteCleanupContext(ctx)
+	if classification == sshTransportAnswering && state.client != nil {
 		stopRemote := r.stopRemote
 		if stopRemote == nil {
 			stopRemote = stopRemoteAgentctl
 		}
-		_ = stopRemote(cleanupCtx, state.client, state.remoteDir, state.pid)
-		cancel()
-		_ = r.closeSSHClient(state.client)
+		_ = r.runWithTransportBackstop(state, sshAgentctlCleanupTimeout, func() error {
+			cleanupCtx, cancel := sshRemoteCleanupContext(ctx)
+			defer cancel()
+			return stopRemote(cleanupCtx, state.client, state.remoteDir, state.pid)
+		})
 	}
+	// Step 5: close the session SSH client, through its once-guard.
+	_ = r.closeClientOnce(state)
+	_ = state.runtimeAPITunnel.Close()
+	// Step 6: wait for the prober to exit.
+	r.awaitProbeExit(state)
 	return nil
 }

@@ -871,14 +996,14 @@ func (r *SSHExecutor) ResumeRemoteInstance(ctx context.Context, req *ExecutorCre
 	remotePort, _ := strconv.Atoi(portStr)
 	fwd, err := StartPortForward(client, remotePort, r.logger)
 	if err != nil {
-		_ = runtimeAPITunnel.Close()
 		_ = client.Close()
+		_ = runtimeAPITunnel.Close()
 		return fmt.Errorf("ssh resume: port forward: %w", err)
 	}
 	if err := waitAgentctlHealthy(ctx, fwd.LocalPort(), sshAgentctlHealthTimeout); err != nil {
-		_ = runtimeAPITunnel.Close()
 		_ = fwd.Close()
 		_ = client.Close()
+		_ = runtimeAPITunnel.Close()
 		return fmt.Errorf("ssh resume: agentctl health: %w", err)
 	}
 	instanceClient, reusingProcess := r.newResumedAgentctlClient(
@@ -886,7 +1011,18 @@ func (r *SSHExecutor) ResumeRemoteInstance(ctx context.Context, req *ExecutorCre
 	)

 	r.mu.Lock()
-	r.sessions[req.InstanceID] = &sshSessionState{
+	if _, ok := r.sessions[req.InstanceID]; ok {
+		// AC-EXECUTORS-SSH-TRANSPORT-LIVENESS-002.1: a concurrent resume or
+		// create already recorded a session for this instance while we were
+		// dialing. Leave it and its watchdog alone and close what we just
+		// built instead of orphaning it.
+		r.mu.Unlock()
+		_ = fwd.Close()
+		_ = client.Close()
+		_ = runtimeAPITunnel.Close()
+		return nil
+	}
+	state := &sshSessionState{
 		target:           target,
 		client:           client,
 		forwarder:        fwd,
@@ -899,7 +1035,11 @@ func (r *SSHExecutor) ResumeRemoteInstance(ctx context.Context, req *ExecutorCre
 		metadata:         cloneSSHMetadata(req.Metadata),
 		prepareEnv:       sshRemoteContributionEnv(req, agentctlBin),
 		runtimeAPITunnel: runtimeAPITunnel,
+		port:             remotePort,
+		workdirRoot:      r.workdirRoot(req.Metadata),
 	}
+	r.sessions[req.InstanceID] = state
+	r.startWatchdogLocked(req.InstanceID, state)
 	r.mu.Unlock()

 	// Refresh transient metadata so downstream consumers see the new local
@@ -988,6 +1128,13 @@ func (r *SSHExecutor) resetManagedBrokerResume(
 	return nil
 }

+// resetTrackedManagedBrokerResume disposes of a tracked session as part of
+// resetting a credential-broker-backed resume (the third of the three
+// disposal paths, AC-EXECUTORS-SSH-TRANSPORT-LIVENESS-002.4). Unlike the stop
+// and close paths, its own reading is taken in its own, earlier critical
+// section: the mutex is taken to read (and to re-check the session is still
+// tracked, closing the race F36 identifies before the preflight's remote
+// round trip), released for the preflight, then re-taken for the removal.
 func (r *SSHExecutor) resetTrackedManagedBrokerResume(
 	ctx context.Context,
 	req *ExecutorCreateRequest,
@@ -995,25 +1142,71 @@ func (r *SSHExecutor) resetTrackedManagedBrokerResume(
 	brokerPreflight func(context.Context, *ssh.Client, *ExecutorCreateRequest, SSHRemotePlatform) error,
 	stopRemote func(context.Context, *ssh.Client, string, int) error,
 ) error {
-	if err := brokerPreflight(ctx, state.client, req, SSHRemotePlatform{}); err != nil {
-		return err
-	}
 	r.mu.Lock()
-	if r.sessions[req.InstanceID] == state {
+	tracked := r.sessions[req.InstanceID] == state
+	classification := classifySSHTransportLocked(state)
+	r.mu.Unlock()
+	if !tracked {
+		// A concurrent disposal already removed this session: return success
+		// immediately, closing no handle and not waiting for the other
+		// call's watchdog (AC-EXECUTORS-SSH-TRANSPORT-LIVENESS-002.4).
+		return nil
+	}
+
+	if classification == sshTransportAnswering {
+		err := r.runWithTransportBackstop(state, sshBrokerPreflightTimeout, func() error {
+			preflightCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), sshBrokerPreflightTimeout)
+			defer cancel()
+			return brokerPreflight(preflightCtx, state.client, req, SSHRemotePlatform{})
+		})
+		if err != nil {
+			if r.isTransportLost(state) {
+				return ErrSSHTransportLost
+			}
+			return err
+		}
+	}
+
+	r.mu.Lock()
+	tracked = r.sessions[req.InstanceID] == state
+	if tracked {
 		delete(r.sessions, req.InstanceID)
 	}
 	r.mu.Unlock()
-	if state.forwarder != nil {
-		_ = state.forwarder.Close()
+	if !tracked {
+		return nil
 	}
-	if err := stopRemote(ctx, state.client, state.remoteDir, state.pid); err != nil {
-		if state.client != nil {
-			_ = r.closeSSHClient(state.client)
+
+	// Step 2: signal the watchdog loop and wait for it — only it — to exit.
+	r.stopWatchdogAndWait(state)
+	// Step 3: close the local port forward through its once-guard.
+	_ = r.closeForwarderOnce(state)
+
+	if classification != sshTransportAnswering {
+		// AC-EXECUTORS-SSH-TRANSPORT-LIVENESS-002.15: skip the remote
+		// controller stop, leave resume metadata unchanged, and report
+		// transport loss.
+		_ = r.closeClientOnce(state)
+		_ = state.runtimeAPITunnel.Close()
+		r.awaitProbeExit(state)
+		return ErrSSHTransportLost
+	}
+
+	stopErr := r.runWithTransportBackstop(state, sshAgentctlCleanupTimeout, func() error {
+		cleanupCtx, cancel := sshRemoteCleanupContext(ctx)
+		defer cancel()
+		return stopRemote(cleanupCtx, state.client, state.remoteDir, state.pid)
+	})
+	// Step 5: close the session SSH client, through its once-guard.
+	_ = r.closeClientOnce(state)
+	_ = state.runtimeAPITunnel.Close()
+	// Step 6: wait for the prober to exit.
+	r.awaitProbeExit(state)
+	if stopErr != nil {
+		if r.isTransportLost(state) {
+			return ErrSSHTransportLost
 		}
-		return fmt.Errorf("ssh resume: stop stale broker-backed agentctl: %w", err)
-	}
-	if state.client != nil {
-		_ = r.closeSSHClient(state.client)
+		return fmt.Errorf("ssh resume: stop stale broker-backed agentctl: %w", stopErr)
 	}
 	clearSSHResumeRuntimeMetadata(req.Metadata)
 	return nil
@@ -1039,8 +1232,19 @@ func (r *SSHExecutor) GetRemoteStatus(ctx context.Context, instance *ExecutorIns
 	if instance == nil {
 		return nil, errors.New("instance is nil")
 	}
+	// AC-EXECUTORS-SSH-TRANSPORT-LIVENESS-001.7: read the transport-lost
+	// marker and the client handle in one critical section and decide there,
+	// so a teardown can't close the client between the read and the probe.
+	// Only the decision needs the mutex; the healthy-session probe below
+	// runs after it is released.
 	r.mu.Lock()
 	state := r.sessions[instance.InstanceID]
+	var transportLost bool
+	var client *ssh.Client
+	if state != nil {
+		transportLost = state.transportLost
+		client = state.client
+	}
 	r.mu.Unlock()
 	now := time.Now()
 	status := &RemoteStatus{
@@ -1059,13 +1263,27 @@ func (r *SSHExecutor) GetRemoteStatus(ctx context.Context, instance *ExecutorIns
 	status.Details["user"] = state.target.User
 	status.Details["fingerprint"] = state.target.PinnedFingerprint

-	probeCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
-	defer cancel()
-	if state.client == nil {
+	if transportLost {
+		status.State = sshStatusDisconnected
+		status.ErrorMessage = sshTransportLostMessage
+		return status, nil
+	}
+	if client == nil {
 		status.State = sshStatusDisconnected
 		return status, nil
 	}
-	if !isRemoteAgentctlAlive(probeCtx, state.client, state.pid) {
+	probeCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
+	defer cancel()
+	if !isRemoteAgentctlAlive(probeCtx, client, state.pid) {
+		// A concurrent teardown may have closed client after the read above
+		// but before (or during) this probe, so a failed probe alone doesn't
+		// distinguish "agentctl not answering" from "transport lost mid-probe"
+		// (AC-EXECUTORS-SSH-TRANSPORT-LIVENESS-001.7). Recheck the marker.
+		if r.isTransportLost(state) {
+			status.State = sshStatusDisconnected
+			status.ErrorMessage = sshTransportLostMessage
+			return status, nil
+		}
 		status.State = sshStatusAgentctlDown
 		return status, nil
 	}
@@ -10,6 +10,7 @@ import (
 	"path/filepath"
 	"strconv"
 	"strings"
+	"sync"
 	"time"

 	"github.com/kevinburke/ssh_config"
@@ -502,6 +503,33 @@ func dialDirect(ctx context.Context, addr string, cfg *ssh.ClientConfig) (*ssh.C
 	return ssh.NewClient(sshConn, chans, reqs), nil
 }

+// sshProxyJumpConn owns both legs of a ProxyJump connection. Closing the
+// bastion transport first releases any read blocked in the inner SSH client.
+// The inner channel close then completes without relying on a remote reply.
+type sshProxyJumpConn struct {
+	net.Conn
+	bastion  interface{ Close() error }
+	once     sync.Once
+	closeErr error
+}
+
+func (c *sshProxyJumpConn) Close() error {
+	if c == nil {
+		return nil
+	}
+	c.once.Do(func() {
+		var bastionErr, tunnelErr error
+		if c.bastion != nil {
+			bastionErr = c.bastion.Close()
+		}
+		if c.Conn != nil {
+			tunnelErr = c.Conn.Close()
+		}
+		c.closeErr = errors.Join(bastionErr, tunnelErr)
+	})
+	return c.closeErr
+}
+
 // dialViaJump implements ProxyJump as a single bastion hop. The bastion is
 // resolved from its own ~/.ssh/config Host block, defaulting to the same
 // identity source as the target (passes the user's agent / key through).
@@ -539,17 +567,16 @@ func dialViaJump(ctx context.Context, target *SSHTarget, finalAddr string, final
 		_ = bastionClient.Close()
 		return nil, fmt.Errorf("ssh: bastion tunnel to %s: %w", finalAddr, err)
 	}
-	sshConn, chans, reqs, err := ssh.NewClientConn(tunnel, finalAddr, finalCfg)
+	jumpConn := &sshProxyJumpConn{Conn: tunnel, bastion: bastionClient}
+	sshConn, chans, reqs, err := ssh.NewClientConn(jumpConn, finalAddr, finalCfg)
 	if err != nil {
-		_ = tunnel.Close()
-		_ = bastionClient.Close()
+		_ = jumpConn.Close()
 		return nil, fmt.Errorf("ssh: handshake with %s via %s: %w", finalAddr, target.ProxyJump, err)
 	}
-	// Attach the bastion close to the final client lifetime.
 	final := ssh.NewClient(sshConn, chans, reqs)
 	go func() {
 		_ = final.Wait()
-		_ = bastionClient.Close()
+		_ = jumpConn.Close()
 	}()
 	return final, nil
 }
@@ -713,7 +740,11 @@ func parsePortString(s string) (int, bool) {
 }

 // SSH connections are owned per-session (no shared pool): a session's client
-// lives on sshSessionState and is Close()d by StopInstance. The earlier
-// pool/refcount/keepalive plumbing was dead code — the executor never used it
-// in production, and the orphaned keepalive goroutines surfaced under e2e
-// fault-injection. See PR #927 for the removal rationale.
+// lives on sshSessionState and is Close()d by StopInstance. An earlier
+// pool/refcount/keepalive design was removed as dead code in PR #927 — the
+// executor never used it in production, and its background goroutines
+// outlived their connections under e2e fault injection. The per-session
+// transport-liveness watchdog in executor_ssh_keepalive.go restores a
+// keepalive, but owned by the session it watches and stopped by the same
+// disposal paths that close the client, so it cannot repeat that leak: see
+// docs/specs/executors/requirements/ssh-transport-liveness.md.
@@ -0,0 +1,456 @@
+package lifecycle
+
+import (
+	"errors"
+	"fmt"
+	"sync"
+	"time"
+
+	"go.uber.org/zap"
+	"golang.org/x/crypto/ssh"
+)
+
+// docs/specs/executors/requirements/ssh-transport-liveness.md
+// AC-EXECUTORS-SSH-TRANSPORT-LIVENESS-001.10
+const (
+	defaultSSHKeepaliveInterval = 15 * time.Second
+	defaultSSHKeepaliveDeadline = 45 * time.Second
+)
+
+// Tuning values for SSH session transport liveness. Package-level variables,
+// not constants, so tests can shorten them (see goleak_test.go's TestMain,
+// which zeroes the keepalive pair for this package's test binary — a
+// non-positive value disables the watchdog per sshKeepaliveTuningValid).
+var (
+	sshKeepaliveInterval = defaultSSHKeepaliveInterval
+	sshKeepaliveDeadline = defaultSSHKeepaliveDeadline
+)
+
+// sshBrokerPreflightTimeout bounds the credential-broker reachability
+// preflight a broker-backed resume reset issues before it stops the stale
+// remote controller. The preflight runs on the raw caller context today and
+// is otherwise unbounded; 20s mirrors sshAgentctlCleanupTimeout, since both
+// are reachability-scale checks, not long-running work.
+var sshBrokerPreflightTimeout = 20 * time.Second
+
+// ErrSSHTransportLost identifies a disposal or reuse decision made because a
+// session's SSH transport was declared lost (deadline/probe-error teardown)
+// or has become unresponsive, distinguishing that cause from an ordinary
+// remote failure or a host that refused the call.
+var ErrSSHTransportLost = errors.New("ssh: session transport lost")
+
+const sshKeepaliveRequestType = "keepalive@openssh.com"
+
+const (
+	sshTransportLostReasonProbeError = "probe error"
+	sshTransportLostReasonDeadline   = "liveness deadline exceeded"
+)
+
+// sshTransportClassification is the outcome of the single disposal reading
+// AC-EXECUTORS-SSH-TRANSPORT-LIVENESS-002.8 requires: it governs every
+// remote command a disposal path issues.
+type sshTransportClassification int
+
+const (
+	sshTransportAnswering sshTransportClassification = iota
+	sshTransportUnresponsive
+	sshTransportLost
+)
+
+// sshKeepaliveTuningValid reports whether interval/deadline are usable
+// (AC-EXECUTORS-SSH-TRANSPORT-LIVENESS-002.9): both positive, and the
+// deadline strictly greater than twice the interval.
+func sshKeepaliveTuningValid(interval, deadline time.Duration) bool {
+	return interval > 0 && deadline > 0 && deadline > 2*interval
+}
+
+// sshKeepaliveWatchdog runs the two goroutines described in the system
+// design's Control flow section for exactly one session SSH client: a
+// prober that sends transport liveness probes one at a time, and a loop that
+// decides when the transport is lost. It exposes start (via
+// startSSHKeepaliveWatchdog), stop-deciding (stopAndAwaitLoop) and
+// await-probe-exit (awaitProbeExit) — the first and last steps of a
+// disposal, always in that order.
+type sshKeepaliveWatchdog struct {
+	interval time.Duration
+	deadline time.Duration
+	client   *ssh.Client
+
+	stopCh   chan struct{}
+	stopOnce sync.Once
+
+	loopDone   chan struct{}
+	proberDone chan struct{}
+
+	recordReply func(time.Time)
+	onLost      func(reason string, silence time.Duration)
+}
+
+// startSSHKeepaliveWatchdog starts a watchdog for client. Callers must have
+// already validated interval/deadline with sshKeepaliveTuningValid.
+// startedAt is the watchdog's start time, used as the silence interval's
+// baseline until the first probe reply completes.
+func startSSHKeepaliveWatchdog(
+	client *ssh.Client,
+	interval, deadline time.Duration,
+	startedAt time.Time,
+	recordReply func(time.Time),
+	onLost func(reason string, silence time.Duration),
+) *sshKeepaliveWatchdog {
+	w := &sshKeepaliveWatchdog{
+		interval:    interval,
+		deadline:    deadline,
+		client:      client,
+		stopCh:      make(chan struct{}),
+		loopDone:    make(chan struct{}),
+		proberDone:  make(chan struct{}),
+		recordReply: recordReply,
+		onLost:      onLost,
+	}
+	// replies is capacity-1 and lossy by design: a coalesced reply is
+	// harmless because another follows within one interval. errs is also
+	// capacity-1 but is written at most once ever (the prober returns
+	// immediately after), so the buffered send always succeeds instantly and
+	// is never dropped — unlike a reply, the loop must always observe it.
+	replies := make(chan time.Time, 1)
+	errs := make(chan error, 1)
+	go w.runProber(replies, errs)
+	go w.runLoop(replies, errs, startedAt)
+	return w
+}
+
+// runProber sends one transport liveness probe at a time, spaced by the
+// interval (AC-EXECUTORS-SSH-TRANSPORT-LIVENESS-001.2). It has exactly two
+// exits, both unconditional: a probe error (published on the guaranteed-
+// delivery errs channel before the goroutine returns, so the loop always
+// observes it) and the stop signal observed while idle between probes. It
+// always terminates once the client is closed, since SendRequest then errors
+// immediately.
+func (w *sshKeepaliveWatchdog) runProber(replies chan<- time.Time, errs chan<- error) {
+	defer close(w.proberDone)
+	for {
+		_, _, err := w.client.SendRequest(sshKeepaliveRequestType, true, nil)
+		now := time.Now()
+		if err != nil {
+			errs <- err
+			return
+		}
+		select {
+		case replies <- now:
+		default:
+		}
+		select {
+		case <-w.stopCh:
+			return
+		case <-time.After(w.interval):
+		}
+	}
+}
+
+// runLoop decides when the transport is lost. Two precedence rules apply,
+// because a select with several ready cases picks at random: a pending
+// completed reply or probe error beats an expired deadline timer (handled by
+// the non-blocking re-check inside the timer.C case below), and the stop
+// signal beats both. The top-of-loop non-blocking check only catches a stop
+// already signaled before that iteration's blocking select begins — it
+// cannot catch a stop delivered concurrently with a probe error or an
+// expired deadline, since the blocking select itself picks at random among
+// whichever cases are ready together. So every path that is about to
+// declare loss re-checks the stop signal immediately first and yields to it.
+func (w *sshKeepaliveWatchdog) runLoop(replies <-chan time.Time, errs <-chan error, startedAt time.Time) {
+	defer close(w.loopDone)
+	lastReply := startedAt
+	timer := time.NewTimer(w.deadline)
+	defer timer.Stop()
+	for {
+		select {
+		case <-w.stopCh:
+			return
+		default:
+		}
+		select {
+		case <-w.stopCh:
+			return
+		case at := <-replies:
+			w.handleReply(at, &lastReply, timer)
+		case <-errs:
+			if w.stopRequested() {
+				return
+			}
+			w.declareLost(sshTransportLostReasonProbeError, time.Since(lastReply))
+			return
+		case <-timer.C:
+			// A completed reply or a probe error may have arrived in the
+			// same instant the deadline fired. Either wins over the timer
+			// alone.
+			select {
+			case at := <-replies:
+				w.handleReply(at, &lastReply, timer)
+			case err := <-errs:
+				_ = err
+				if w.stopRequested() {
+					return
+				}
+				w.declareLost(sshTransportLostReasonProbeError, time.Since(lastReply))
+				return
+			default:
+				if w.stopRequested() {
+					return
+				}
+				if w.handleDeadline(lastReply, timer) {
+					return
+				}
+			}
+		}
+	}
+}
+
+// stopRequested reports whether the stop signal has been raised, without
+// blocking. Used immediately before every action that would declare
+// transport loss, so a stop delivered in the same instant as a probe error
+// or an expired deadline always wins.
+func (w *sshKeepaliveWatchdog) stopRequested() bool {
+	select {
+	case <-w.stopCh:
+		return true
+	default:
+		return false
+	}
+}
+
+func (w *sshKeepaliveWatchdog) handleReply(at time.Time, lastReply *time.Time, timer *time.Timer) {
+	*lastReply = at
+	if w.recordReply != nil {
+		w.recordReply(*lastReply)
+	}
+	if !timer.Stop() {
+		select {
+		case <-timer.C:
+		default:
+		}
+	}
+	timer.Reset(w.deadline)
+}
+
+func (w *sshKeepaliveWatchdog) handleDeadline(lastReply time.Time, timer *time.Timer) bool {
+	silence := time.Since(lastReply)
+	if silence > w.deadline {
+		w.declareLost(sshTransportLostReasonDeadline, silence)
+		return true
+	}
+	// Spurious wake (the timer fired just as the deadline's boundary was
+	// reached but the strictly-greater check above didn't trip): reset for
+	// the remaining time rather than looping busily.
+	remaining := w.deadline - silence
+	if remaining <= 0 {
+		remaining = time.Millisecond
+	}
+	timer.Reset(remaining)
+	return false
+}
+
+func (w *sshKeepaliveWatchdog) declareLost(reason string, silence time.Duration) {
+	if w.onLost != nil {
+		w.onLost(reason, silence)
+	}
+}
+
+// stopAndAwaitLoop signals the watchdog to stop and waits for the watchdog
+// loop — not the prober — to exit. Bounded: the loop is either parked in its
+// select, already returned, or performing a teardown (two local handle
+// closes and one log line), so this never touches the network. Idempotent
+// and safe on a nil-free zero watchdog that never started. Callers must not
+// hold the executor mutex while waiting here: a mid-teardown loop acquires
+// that mutex before closing loopDone, so holding it would deadlock.
+func (w *sshKeepaliveWatchdog) stopAndAwaitLoop() {
+	w.stopOnce.Do(func() { close(w.stopCh) })
+	<-w.loopDone
+}
+
+// awaitProbeExit waits for the prober goroutine to exit. Call after
+// stopAndAwaitLoop and after the session SSH client has been closed: the
+// close is what unblocks a prober mid-SendRequest, whether idle or wedged.
+func (w *sshKeepaliveWatchdog) awaitProbeExit() {
+	<-w.proberDone
+}
+
+// classifySSHTransportLocked reads state's transport-lost marker and, when a
+// watchdog is running, its silence interval, and classifies the transport
+// (AC-EXECUTORS-SSH-TRANSPORT-LIVENESS-002.8). Callers must hold the
+// executor mutex — the marker and the last-probe-reply timestamp it reads
+// are guarded by it.
+func classifySSHTransportLocked(state *sshSessionState) sshTransportClassification {
+	if state.transportLost {
+		return sshTransportLost
+	}
+	if state.watchdog == nil {
+		return sshTransportAnswering
+	}
+	silence := time.Since(state.lastProbeReply)
+	if silence > 2*state.watchdog.interval {
+		return sshTransportUnresponsive
+	}
+	return sshTransportAnswering
+}
+
+// closeForwarderOnce closes state's local port forward through its
+// once-guard (AC-EXECUTORS-SSH-TRANSPORT-LIVENESS-002.7): whichever of
+// transport teardown or a disposal path reaches it first performs the
+// close, and the other observes the same result.
+func (r *SSHExecutor) closeForwarderOnce(state *sshSessionState) error {
+	state.forwarderCloseOnce.Do(func() {
+		if state.forwarder != nil {
+			state.forwarderCloseErr = state.forwarder.Close()
+		}
+	})
+	return state.forwarderCloseErr
+}
+
+// closeClientOnce closes state's session SSH client through its own
+// once-guard, independent of the forwarder's.
+func (r *SSHExecutor) closeClientOnce(state *sshSessionState) error {
+	state.clientCloseOnce.Do(func() {
+		if state.client != nil {
+			state.clientCloseErr = r.closeSSHClient(state.client)
+		}
+	})
+	return state.clientCloseErr
+}
+
+// markTransportLost sets state's transport-lost marker without performing a
+// full transport teardown. Used by the disposal backstop
+// (AC-EXECUTORS-SSH-TRANSPORT-LIVENESS-002.8): it closes only the client
+// through its once-guard and must not log the teardown warning, so it is not
+// "transport teardown" under the spec's Terminology.
+func (r *SSHExecutor) markTransportLost(state *sshSessionState) {
+	r.mu.Lock()
+	state.transportLost = true
+	r.mu.Unlock()
+}
+
+// isTransportLost reports state's transport-lost marker under the executor
+// mutex.
+func (r *SSHExecutor) isTransportLost(state *sshSessionState) bool {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	return state.transportLost
+}
+
+// transportTeardown performs transport teardown for state
+// (AC-EXECUTORS-SSH-TRANSPORT-LIVENESS-001.5): set the marker, close the
+// forward then the client (each through its once-guard), close the optional
+// runtime API tunnel after the client, and log one warning
+// naming the executor instance, remote host, and silence interval. Runs at
+// most once per session (AC-EXECUTORS-SSH-TRANSPORT-LIVENESS-002.5); it does
+// not remove the session from the tracked-sessions map
+// (AC-EXECUTORS-SSH-TRANSPORT-LIVENESS-002.6).
+func (r *SSHExecutor) transportTeardown(instanceID string, state *sshSessionState, reason string, silence time.Duration) {
+	state.teardownOnce.Do(func() {
+		r.mu.Lock()
+		state.transportLost = true
+		r.mu.Unlock()
+		forwarderErr := r.closeForwarderOnce(state)
+		clientErr := r.closeClientOnce(state)
+		_ = state.runtimeAPITunnel.Close()
+		r.logTransportLoss(instanceID, state, reason, silence, forwarderErr, clientErr)
+	})
+}
+
+func (r *SSHExecutor) logTransportLoss(instanceID string, state *sshSessionState, reason string, silence time.Duration, forwarderErr, clientErr error) {
+	host := ""
+	if state.target != nil {
+		host = state.target.Host
+	}
+	fields := []zap.Field{
+		zap.String("instance_id", instanceID),
+		zap.String("host", host),
+		zap.Duration("silence_interval", silence),
+		zap.String("reason", reason),
+	}
+	if forwarderErr != nil {
+		fields = append(fields, zap.NamedError("forwarder_close_error", forwarderErr))
+	}
+	if clientErr != nil {
+		fields = append(fields, zap.NamedError("client_close_error", clientErr))
+	}
+	r.logger.Warn(sshTransportLostMessage, fields...)
+}
+
+// startWatchdogLocked starts a watchdog for state's session SSH client, in
+// the same critical section that recorded state in r.sessions
+// (AC-EXECUTORS-SSH-TRANSPORT-LIVENESS-001.1). Callers must hold r.mu. A
+// non-positive interval/deadline, or a deadline not greater than twice the
+// interval, leaves state.watchdog nil and every existing disposal path
+// unchanged (AC-EXECUTORS-SSH-TRANSPORT-LIVENESS-002.9).
+func (r *SSHExecutor) startWatchdogLocked(instanceID string, state *sshSessionState) {
+	interval, deadline := sshKeepaliveInterval, sshKeepaliveDeadline
+	if !sshKeepaliveTuningValid(interval, deadline) {
+		r.logger.Debug("ssh: not starting transport watchdog", zap.String("instance_id", instanceID))
+		return
+	}
+	startedAt := time.Now()
+	state.lastProbeReply = startedAt
+	state.watchdog = startSSHKeepaliveWatchdog(state.client, interval, deadline, startedAt,
+		func(at time.Time) {
+			r.mu.Lock()
+			state.lastProbeReply = at
+			r.mu.Unlock()
+		},
+		func(reason string, silence time.Duration) {
+			r.transportTeardown(instanceID, state, reason, silence)
+		},
+	)
+}
+
+// stopWatchdogAndWait signals state's watchdog to stop and waits for its
+// loop to exit. A no-op on a session that never started one.
+func (r *SSHExecutor) stopWatchdogAndWait(state *sshSessionState) {
+	if state.watchdog != nil {
+		state.watchdog.stopAndAwaitLoop()
+	}
+}
+
+// awaitProbeExit waits for state's watchdog's prober to exit. A no-op on a
+// session that never started one.
+func (r *SSHExecutor) awaitProbeExit(state *sshSessionState) {
+	if state.watchdog != nil {
+		state.watchdog.awaitProbeExit()
+	}
+}
+
+// runWithTransportBackstop runs fn — a remote command issued over state's
+// session SSH client — on its own goroutine and races it against timeout
+// (AC-EXECUTORS-SSH-TRANSPORT-LIVENESS-002.8). fn is expected to carry its
+// own bound (typically a context.WithTimeout of the same duration) for the
+// ordinary case where it is already executing; the backstop exists for the
+// harder case where fn is blocked opening its SSH channel, which no context
+// bounds. On timeout it closes the client through its once-guard — the only
+// thing that releases a call blocked in Client.NewSession — marks the client
+// transport-lost, and waits for fn's goroutine to actually exit before
+// returning, so the abandoned command never outlives this call.
+func (r *SSHExecutor) runWithTransportBackstop(state *sshSessionState, timeout time.Duration, fn func() error) error {
+	done := make(chan error, 1)
+	go func() { done <- fn() }()
+	timer := time.NewTimer(timeout)
+	defer timer.Stop()
+	select {
+	case err := <-done:
+		return err
+	case <-timer.C:
+		// fn may have completed in the same instant the timer fired — a
+		// select with both cases ready picks at random. Recheck done
+		// before treating this as a timeout, so a command that actually
+		// finished is never reported as abandoned.
+		select {
+		case err := <-done:
+			return err
+		default:
+		}
+		r.markTransportLost(state)
+		_ = r.closeClientOnce(state)
+		if err := <-done; err != nil {
+			return err
+		}
+		return fmt.Errorf("ssh: remote command exceeded %s and was abandoned", timeout)
+	}
+}
@@ -14,6 +14,7 @@ import (
 	"strings"
 	"sync"
 	"testing"
+	"time"

 	"golang.org/x/crypto/ssh"

@@ -322,6 +323,7 @@ func (h *sshLaunchHarness) createInstanceBodies() []string {
 }

 func TestSSHExecutorCreateInstanceProvisionsAndTracksTheSession(t *testing.T) {
+	withSSHKeepaliveTuning(t, 5*time.Second, 20*time.Second)
 	harness := newSSHLaunchHarness(t, "4242")
 	exec := NewSSHExecutor(nil, nil, NewAgentctlResolver(newTestLogger()), newTestLogger())
 	t.Cleanup(func() { _ = exec.Close() })
@@ -432,6 +434,9 @@ func TestSSHExecutorCreateInstanceProvisionsAndTracksTheSession(t *testing.T) {
 	if state.platform.GOOS != "linux" || state.platform.GOARCH != "amd64" {
 		t.Fatalf("session platform = %+v", state.platform)
 	}
+	if state.watchdog == nil {
+		t.Fatal("CreateInstance must start a transport-liveness watchdog for the session (AC-EXECUTORS-SSH-TRANSPORT-LIVENESS-001.1)")
+	}

 	// StopInstance releases the forwarder and SSH client and forgets the state.
 	if err := exec.StopInstance(context.Background(), instance, true); err != nil {
@@ -15,10 +15,12 @@ import (
 	"github.com/kandev/kandev/internal/scriptengine"
 )

-const (
-	sshCleanupTimeout        = 60 * time.Second
-	sshPrepareOutputMaxLines = 20
-)
+const sshPrepareOutputMaxLines = 20
+
+// sshCleanupTimeout bounds the remote cleanup script a session stop issues.
+// A package-level variable, not a constant, so a backstop test need not wait
+// out real seconds (see executor_ssh_keepalive.go).
+var sshCleanupTimeout = 60 * time.Second

 func cloneSSHMetadata(metadata map[string]interface{}) map[string]interface{} {
 	if len(metadata) == 0 {
@@ -6,6 +6,7 @@ import (
 	"strconv"
 	"strings"
 	"testing"
+	"time"

 	"golang.org/x/crypto/ssh"

@@ -90,6 +91,44 @@ func TestSSHExecutorGetRemoteStatus(t *testing.T) {
 			t.Fatalf("State = %q, want %q", status.State, sshStatusAgentctlDown)
 		}
 	})
+
+	t.Run("transport lost during the probe reports disconnected, not agentctl-down", func(t *testing.T) {
+		// AC-EXECUTORS-SSH-TRANSPORT-LIVENESS-001.7: GetRemoteStatus reads the
+		// transport-lost marker and the client handle in one critical section,
+		// then releases the mutex before probing. A teardown racing in after
+		// that read (marker still false, client still open at read time) but
+		// before the probe completes must still surface as `disconnected`,
+		// not `agentctl-down` — a failed probe alone can't tell the two apart.
+		exec := NewSSHExecutor(nil, nil, nil, newTestLogger())
+		var state *sshSessionState
+		server := newFakeSSHServer(t, func(string, string) sshExecResult {
+			// Simulate a concurrent watchdog-driven teardown landing exactly
+			// between GetRemoteStatus's marker read and this probe: flip the
+			// marker under the same mutex the real teardown uses, then fail
+			// the probe the way a torn-down client would.
+			exec.mu.Lock()
+			state.transportLost = true
+			exec.mu.Unlock()
+			return sshFail("no such process")
+		})
+		state = &sshSessionState{
+			target: &SSHTarget{Host: "build.example"},
+			client: server.dial(t),
+			pid:    4242,
+		}
+		exec.sessions["i"] = state
+
+		status, err := exec.GetRemoteStatus(context.Background(), &ExecutorInstance{InstanceID: "i"})
+		if err != nil {
+			t.Fatalf("GetRemoteStatus: %v", err)
+		}
+		if status.State != sshStatusDisconnected {
+			t.Fatalf("State = %q, want %q", status.State, sshStatusDisconnected)
+		}
+		if status.ErrorMessage != sshTransportLostMessage {
+			t.Fatalf("ErrorMessage = %q", status.ErrorMessage)
+		}
+	})
 }

 func TestSSHShouldStopRemoteAgentctl(t *testing.T) {
@@ -219,6 +258,7 @@ func TestSSHExecutorResumeRemoteInstance(t *testing.T) {
 	})

 	t.Run("re-attaches to a live remote agentctl", func(t *testing.T) {
+		withSSHKeepaliveTuning(t, 5*time.Second, 20*time.Second)
 		harness := newSSHLaunchHarness(t, "4242")
 		exec := NewSSHExecutor(nil, nil, nil, newTestLogger())
 		t.Cleanup(func() { _ = exec.Close() })
@@ -255,6 +295,9 @@ func TestSSHExecutorResumeRemoteInstance(t *testing.T) {
 			t.Fatalf("metadata forward port = %v, want the new local port %d",
 				forwardPort, state.forwarder.LocalPort())
 		}
+		if state.watchdog == nil {
+			t.Fatal("ResumeRemoteInstance must start a transport-liveness watchdog for the session (AC-EXECUTORS-SSH-TRANSPORT-LIVENESS-001.1)")
+		}
 	})

 	// @covers AC-EXECUTORS-SSH-EXECUTOR-001.9
@@ -0,0 +1,169 @@
+package lifecycle
+
+import (
+	"context"
+	"io"
+	"net"
+	"sync"
+	"testing"
+	"time"
+
+	"golang.org/x/crypto/ssh"
+)
+
+type clientReleaseListener struct {
+	clientClosed <-chan struct{}
+	closed       chan struct{}
+	once         sync.Once
+}
+
+func (l *clientReleaseListener) Accept() (net.Conn, error) {
+	<-l.closed
+	return nil, net.ErrClosed
+}
+
+func (l *clientReleaseListener) Close() error {
+	<-l.clientClosed
+	l.once.Do(func() { close(l.closed) })
+	return nil
+}
+
+func (l *clientReleaseListener) Addr() net.Addr {
+	return &net.TCPAddr{IP: net.IPv4(127, 0, 0, 1)}
+}
+
+// @covers AC-EXECUTORS-SSH-TRANSPORT-LIVENESS-002.4
+func TestSSHExecutorStopInstanceClosesClientBeforeRuntimeAPITunnel(t *testing.T) {
+	clientClosed := make(chan struct{})
+	listenerClosed := make(chan struct{})
+	listener := &clientReleaseListener{clientClosed: clientClosed, closed: listenerClosed}
+
+	exec := NewSSHExecutor(nil, nil, nil, newTestLogger())
+	exec.stopRemote = func(context.Context, *ssh.Client, string, int) error { return nil }
+	exec.closeClient = func(*ssh.Client) error {
+		close(clientClosed)
+		return nil
+	}
+	exec.sessions["instance-1"] = &sshSessionState{
+		client:           &ssh.Client{},
+		runtimeAPITunnel: &sshRuntimeAPITunnel{listener: listener},
+		remoteDir:        "/remote/session",
+		pid:              4242,
+	}
+
+	done := make(chan error, 1)
+	go func() {
+		done <- exec.StopInstance(context.Background(), &ExecutorInstance{
+			InstanceID: "instance-1",
+			StopReason: StopReasonTaskDeleted,
+		}, false)
+	}()
+	select {
+	case err := <-done:
+		if err != nil {
+			t.Fatalf("StopInstance: %v", err)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("StopInstance waited for the runtime API tunnel before closing the SSH client")
+	}
+	select {
+	case <-listenerClosed:
+	case <-time.After(time.Second):
+		t.Fatal("runtime API tunnel did not close")
+	}
+}
+
+// @covers AC-EXECUTORS-SSH-TRANSPORT-LIVENESS-001.5
+func TestSSHKeepaliveTeardownClosesRuntimeAPITunnelAfterClient(t *testing.T) {
+	exec, _ := newObservedSSHExecutor(t)
+	server := newFakeSSHServer(t, nil)
+	client := server.dial(t)
+
+	clientClosed := make(chan struct{})
+	listenerClosed := make(chan struct{})
+	state := &sshSessionState{
+		target: &SSHTarget{Host: "build.example", User: "deploy", PinnedFingerprint: "SHA256:x"},
+		client: client,
+		runtimeAPITunnel: &sshRuntimeAPITunnel{listener: &clientReleaseListener{
+			clientClosed: clientClosed,
+			closed:       listenerClosed,
+		}},
+	}
+	exec.closeClient = func(c *ssh.Client) error {
+		close(clientClosed)
+		return c.Close()
+	}
+
+	exec.transportTeardown("instance-1", state, sshTransportLostReasonDeadline, time.Second)
+	select {
+	case <-listenerClosed:
+	case <-time.After(time.Second):
+		t.Fatal("transport teardown did not close the runtime API tunnel")
+	}
+}
+
+type orderedCloseConn struct {
+	bastionClosed <-chan struct{}
+	closed        chan struct{}
+	once          sync.Once
+}
+
+func (c *orderedCloseConn) Read([]byte) (int, error) { return 0, io.EOF }
+
+func (c *orderedCloseConn) Write(p []byte) (int, error) { return len(p), nil }
+
+func (c *orderedCloseConn) Close() error {
+	<-c.bastionClosed
+	c.once.Do(func() { close(c.closed) })
+	return nil
+}
+
+func (c *orderedCloseConn) LocalAddr() net.Addr {
+	return &net.TCPAddr{IP: net.IPv4(127, 0, 0, 1)}
+}
+
+func (c *orderedCloseConn) RemoteAddr() net.Addr {
+	return &net.TCPAddr{IP: net.IPv4(127, 0, 0, 1)}
+}
+
+func (c *orderedCloseConn) SetDeadline(time.Time) error      { return nil }
+func (c *orderedCloseConn) SetReadDeadline(time.Time) error  { return nil }
+func (c *orderedCloseConn) SetWriteDeadline(time.Time) error { return nil }
+
+type closeSignal struct {
+	closed chan struct{}
+	once   sync.Once
+}
+
+func (s *closeSignal) Close() error {
+	s.once.Do(func() { close(s.closed) })
+	return nil
+}
+
+func TestSSHProxyJumpConnClosesBastionBeforeTunnel(t *testing.T) {
+	bastionClosed := make(chan struct{})
+	tunnelClosed := make(chan struct{})
+	conn := &sshProxyJumpConn{
+		Conn:    &orderedCloseConn{bastionClosed: bastionClosed, closed: tunnelClosed},
+		bastion: &closeSignal{closed: bastionClosed},
+	}
+
+	done := make(chan error, 1)
+	go func() { done <- conn.Close() }()
+	select {
+	case err := <-done:
+		if err != nil {
+			t.Fatalf("sshProxyJumpConn.Close: %v", err)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("sshProxyJumpConn.Close waited for the tunnel before closing the bastion")
+	}
+	select {
+	case <-tunnelClosed:
+	case <-time.After(time.Second):
+		t.Fatal("sshProxyJumpConn did not close the tunnel")
+	}
+	if err := conn.Close(); err != nil {
+		t.Fatalf("second sshProxyJumpConn.Close: %v", err)
+	}
+}
@@ -16,6 +16,15 @@ import (
 // only when `OTEL_EXPORTER_OTLP_ENDPOINT` is set — in tests it isn't, so
 // the noop provider is used and no batchSpanProcessor / OTLP retry
 // goroutines are spawned. No IgnoreTopFunction is needed here.
+//
+// sshKeepaliveInterval/sshKeepaliveDeadline are zeroed for this test binary
+// so the fifteen CreateInstance/ResumeRemoteInstance call sites across the
+// package don't each acquire a real 15s SSH transport watchdog under this
+// leak check. A non-positive value disables the watchdog
+// (sshKeepaliveTuningValid); tests that exercise it set millisecond values
+// of their own and must not run with t.Parallel().
 func TestMain(m *testing.M) {
+	sshKeepaliveInterval = 0
+	sshKeepaliveDeadline = 0
 	goleak.VerifyTestMain(m)
 }
@@ -154,7 +154,7 @@ func (m *Manager) startAgentProcess(ctx context.Context, executionID string) (re
 	// Check if we're reconnecting to an existing running agent process.
 	// When the existing process is still alive inside a remote executor (e.g., Sprites),
 	// we skip subprocess launch and go directly to ACP session initialization.
-	reuseExisting := execution.metadataBool("reuse_existing_process")
+	reuseExisting := execution.metadataBool(MetadataKeyReuseExistingProcess)

 	if !reuseExisting && execution.AgentCommand == "" {
 		return fmt.Errorf("execution %q has no agent command configured", executionID)
@@ -9,6 +9,7 @@ import (
 	"strconv"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"testing"

 	"github.com/pkg/sftp"
@@ -66,8 +67,22 @@ type fakeSSHServer struct {
 	// tcpTarget maps a requested direct-tcpip port onto a real local address.
 	// Returning "" rejects the channel.
 	tcpTarget func(port uint32) string
+
+	// silent, when set, makes the server accept a global request or a new
+	// channel open and answer neither — so SendRequest and Client.NewSession
+	// both block, as on a wedged link. Toggled live so a test can run a
+	// session healthy, then flip the server silent mid-test.
+	silent atomic.Bool
 }

+// setSilent switches the server's silent mode on or off. While silent, every
+// global request (including keepalive@openssh.com) and every new channel
+// open (including a session for a remote command) goes unanswered, so the
+// requester's SendRequest or Client.NewSession blocks exactly as it would on
+// a transport that has stopped carrying traffic. Existing, already-open
+// channels are unaffected.
+func (s *fakeSSHServer) setSilent(v bool) { s.silent.Store(v) }
+
 // newFakeSSHServer starts a listener on 127.0.0.1 and serves SSH until the
 // test finishes. handler may be nil, in which case every command succeeds with
 // empty output.
@@ -200,6 +215,13 @@ func (s *fakeSSHServer) serveConn(conn net.Conn) {
 	go s.serveGlobalRequests(serverConn, reqs)

 	for newChan := range chans {
+		if s.silent.Load() {
+			// Stall the channel open: neither Accept nor Reject, so the
+			// client's OpenChannel blocks until the connection itself tears
+			// down. No goroutine needed — the pending request resolves on
+			// its own when the mux beneath it closes.
+			continue
+		}
 		switch newChan.ChannelType() {
 		case "session":
 			s.wg.Add(1)
@@ -213,9 +235,16 @@ func (s *fakeSSHServer) serveConn(conn net.Conn) {
 	}
 }

+// serveGlobalRequests answers every global request (replying false to one
+// that wants a reply, mirroring ssh.DiscardRequests) unless the server is
+// silent, in which case it drains the request without replying — so a
+// wantReply sender's SendRequest blocks until the connection tears down.
 func (s *fakeSSHServer) serveGlobalRequests(conn *ssh.ServerConn, requests <-chan *ssh.Request) {
 	defer s.wg.Done()
 	for req := range requests {
+		if s.silent.Load() {
+			continue
+		}
 		switch req.Type {
 		case "tcpip-forward":
 			s.handleTCPIPForward(conn, req)