feat(executor): reuse SSH connections instead of dialing per command

cmd/agentssh/main.go

@@ -49,10 +49,16 @@ var newExecutor = func(cfg *config.Config) executor.Executor {
 		if cfg != nil {
 			options.HostKeyPolicy = cfg.Inventory.HostKeyPolicy
 			options.PasswordSource = passwordSourceForRun(cfg.Paths)
+			options.DisableConnectionReuse = !sshMultiplexingEnabled(cfg)
+			options.KeepAliveInterval = sshKeepAliveInterval(cfg)
 		}
 		return executor.NewNativeExecutor(options)
 	default:
-		return executor.NewSSHExecutor(nil)
+		return executor.NewSSHExecutorWithOptions(nil, executor.SSHOptions{
+			DisableMultiplexing: !sshMultiplexingEnabled(cfg),
+			ControlPersist:      sshControlPersist(cfg),
+			KeepAliveInterval:   sshKeepAliveInterval(cfg),
+		})
 	}
 }
 
@@ -976,6 +982,46 @@ func selectedTransport(cfg *config.Config) string {
 	}
 }
 
+func sshMultiplexingEnabled(cfg *config.Config) bool {
+	if cfg == nil {
+		return true
+	}
+	switch strings.ToLower(strings.TrimSpace(cfg.Inventory.SSH.Multiplexing)) {
+	case "", "on", "true", "yes", "1", "auto":
+		return true
+	case "off", "false", "no", "0", "disabled":
+		return false
+	default:
+		return true
+	}
+}
+
+func sshControlPersist(cfg *config.Config) time.Duration {
+	if cfg == nil {
+		return 0
+	}
+	return sshDuration(cfg.Inventory.SSH.ControlPersist)
+}
+
+func sshKeepAliveInterval(cfg *config.Config) time.Duration {
+	if cfg == nil {
+		return 0
+	}
+	return sshDuration(cfg.Inventory.SSH.KeepAliveInterval)
+}
+
+func sshDuration(value string) time.Duration {
+	value = strings.TrimSpace(value)
+	if value == "" {
+		return 0
+	}
+	duration, err := time.ParseDuration(value)
+	if err != nil {
+		return 0
+	}
+	return duration
+}
+
 type inventoryAddOptions struct {
 	Name         string
 	Addr         string
@@ -1549,6 +1595,7 @@ func runInventoryDiscover(cmd *cobra.Command, opts inventoryDiscoverOptions) err
 			// will run (env-only master; no prompting across many candidates).
 			PasswordSource: passwordSourceForRun(cfg.Paths),
 		})
+		defer func() { _ = exec.Close() }()
 		result.Candidates = discovery.Probe(context.Background(), result.Candidates, discovery.ProbeOptions{
 			Executor:    exec,
 			Timeout:     executor.ProbeTimeout,
@@ -1667,6 +1714,7 @@ func runInventoryTest(cmd *cobra.Command, name string) error {
 		// (env-only master).
 		PasswordSource: passwordSourceForRun(cfg.Paths),
 	})
+	defer func() { _ = exec.Close() }()
 	result := exec.Probe(context.Background(), resolved.Targets[0])
 	if result.Err == nil && result.ExitCode == 0 {
 		refreshInventoryHostOS(cfg.Paths, name, result.OS)
@@ -1756,6 +1804,7 @@ func runDirect(cmd *cobra.Command, targetName string, remoteCommand string, flag
 	sessionStore := approval.SessionStore{Dir: cfg.Paths.SessionsDir}
 	pendingStore := approval.PendingStore{PendingDir: cfg.Paths.PendingDir, ResponsesDir: cfg.Paths.ResponsesDir}
 	ssh := newExecutor(cfg)
+	defer func() { _ = ssh.Close() }()
 	plans, err := buildRunPlans(cfg, resolved, remoteCommand, flags, runtime, sessionStore, runtime.Enabled)
 	if err != nil {
 		return err

cmd/agentssh/main_test.go

@@ -111,9 +111,13 @@ func TestTransportSelection(t *testing.T) {
 	}
 	cfg.Inventory.Transport = executor.TransportShell
 	shellExec := newExecutor(cfg)
-	if _, ok := shellExec.(executor.SSHExecutor); !ok {
+	shellSSH, ok := shellExec.(executor.SSHExecutor)
+	if !ok {
 		t.Fatalf("inventory shell transport = %T, want shell-out SSHExecutor", shellExec)
 	}
+	if shellSSH.Options.DisableMultiplexing {
+		t.Fatal("default shell executor disabled multiplexing")
+	}
 	t.Setenv("AGENTSSH_TRANSPORT", executor.TransportShell)
 	cfg.Inventory.Transport = executor.TransportNative
 	envExec := newExecutor(cfg)
@@ -123,6 +127,38 @@ func TestTransportSelection(t *testing.T) {
 	t.Logf("transport default=%T inventory_shell=%T env_shell_overrides=%T", defaultExec, shellExec, envExec)
 }
 
+func TestNewExecutorSSHReuseConfig(t *testing.T) {
+	cfg := &config.Config{}
+	cfg.Inventory.SSH.Multiplexing = "off"
+	cfg.Inventory.SSH.ControlPersist = "2m"
+	cfg.Inventory.SSH.KeepAliveInterval = "15s"
+
+	nativeExec := newExecutor(cfg)
+	nativeSSH, ok := nativeExec.(executor.NativeExecutor)
+	if !ok {
+		t.Fatalf("native executor = %T", nativeExec)
+	}
+	if !nativeSSH.Options.DisableConnectionReuse {
+		t.Fatal("native DisableConnectionReuse = false, want true")
+	}
+	if nativeSSH.Options.KeepAliveInterval != 15*time.Second {
+		t.Fatalf("native keepalive = %s, want 15s", nativeSSH.Options.KeepAliveInterval)
+	}
+
+	cfg.Inventory.Transport = executor.TransportShell
+	shellExec := newExecutor(cfg)
+	shellSSH, ok := shellExec.(executor.SSHExecutor)
+	if !ok {
+		t.Fatalf("shell executor = %T", shellExec)
+	}
+	if !shellSSH.Options.DisableMultiplexing {
+		t.Fatal("shell DisableMultiplexing = false, want true")
+	}
+	if shellSSH.Options.ControlPersist != 2*time.Minute || shellSSH.Options.KeepAliveInterval != 15*time.Second {
+		t.Fatalf("shell options = %#v", shellSSH.Options)
+	}
+}
+
 func TestPrintSSHExit255MappingDemo(t *testing.T) {
 	result := executor.Result{
 		ExitCode: 255,
@@ -2135,6 +2171,8 @@ func (e fakeExecutor) Run(_ context.Context, request executor.Request) executor.
 	}
 }
 
+func (e fakeExecutor) Close() error { return nil }
+
 var _ executor.Executor = fakeExecutor{}
 
 func (e fakeExecutor) RunStreaming(_ context.Context, request executor.Request, stdout io.Writer, stderr io.Writer) executor.Result {
@@ -2170,6 +2208,8 @@ func (e osDetectingExecutor) Run(_ context.Context, request executor.Request) ex
 	}
 }
 
+func (e osDetectingExecutor) Close() error { return nil }
+
 var _ executor.Executor = osDetectingExecutor{}
 
 func writeInChunks(w io.Writer, data []byte, size int) {
@@ -2208,6 +2248,8 @@ func (e bufferedOnlyExecutor) Run(_ context.Context, request executor.Request) e
 	}
 }
 
+func (e bufferedOnlyExecutor) Close() error { return nil }
+
 var _ executor.Executor = bufferedOnlyExecutor{}
 
 type cliTestSSHServer struct {

internal/config/config.go

@@ -28,6 +28,10 @@ const seedInventoryYAML = `# AgentSSH inventory — manage via 'agentssh tui' (H
 version: 1
 transport: native           # built-in Go SSH client; set "ssh" to shell out to system ssh
 host_key_policy: strict     # or "accept-new" for trust-on-first-use
+ssh:
+  multiplexing: on          # set "off" to disable connection reuse
+  control_persist: 60s      # shell transport ControlPersist lifetime
+  keepalive_interval: 30s   # pooled/native heartbeat and shell ServerAliveInterval
 hosts: {}
 `
 

internal/config/config_test.go

@@ -91,6 +91,9 @@ func TestEnsureHomeCreatesAndSeeds(t *testing.T) {
 	if cfg.Inventory.Transport != "native" {
 		t.Fatalf("seeded transport = %q, want native", cfg.Inventory.Transport)
 	}
+	if cfg.Inventory.SSH.Multiplexing != "on" || cfg.Inventory.SSH.ControlPersist != "60s" || cfg.Inventory.SSH.KeepAliveInterval != "30s" {
+		t.Fatalf("seeded ssh config = %#v", cfg.Inventory.SSH)
+	}
 	if len(cfg.Policy.Rules) != 0 || len(cfg.Policy.HostOverrides) != 0 {
 		t.Fatalf("seeded policy should have zero active rules: %+v", cfg.Policy)
 	}