Feat/openssh multiplexing

cmd/cli/main.go

@@ -33,6 +33,14 @@ func main() {
 	}
 
 	if err := cmd.Execute(); err != nil {
+		if code, ok := cmd.ExitCodeFromError(err); ok {
+			if !cmd.ErrorAlreadyReported(err) {
+				// Control-command diagnostics must stay byte-for-byte compatible
+				// with OpenSSH: plain stderr, no color or log prefix.
+				fmt.Fprintln(os.Stderr, err.Error())
+			}
+			os.Exit(code)
+		}
 		reportError(err)
 		os.Exit(1)
 	}

cmd/command_output.go

@@ -34,7 +34,11 @@ func writeCommandOutput(ctx context.Context, target io.Writer, render func(io.Wr
 // and rechecks capabilities before starting any remote command. Never retry a
 // command on the legacy path after strict execution has begun.
 func commandOutputSupported(ctx context.Context) (bool, error) {
-	for _, target := range []*os.File{os.Stdout, os.Stderr} {
+	return streamOutputSupported(ctx, os.Stdout, os.Stderr)
+}
+
+func streamOutputSupported(ctx context.Context, stdout, stderr io.Writer) (bool, error) {
+	for _, target := range []io.Writer{stdout, stderr} {
 		_, release, supported, err := ssh.BindOutput(ctx, target)
 		err = errors.Join(err, release())
 		if err != nil {

cmd/exec.go

@@ -11,6 +11,7 @@ import (
 	"sync"
 
 	"github.com/spf13/cobra"
+	"github.com/spf13/pflag"
 	"github.com/wentf9/xops-cli/cmd/utils"
 	"github.com/wentf9/xops-cli/core/ssh"
 	"github.com/wentf9/xops-cli/pkg/adapter"
@@ -77,6 +78,7 @@ func newCmdExecWithOptions(o *ExecOptions) *cobra.Command {
 	cmd.Flags().StringVarP(&o.User, "login", "l", "", i18n.T("flag_login"))
 	cmd.Flags().StringVarP(&o.IdentityFile, "identity", "i", "", i18n.T("flag_identity"))
 	cmd.Flags().StringVarP(&o.JumpHost, "jump", "J", "", i18n.T("flag_jump"))
+	o.control.RegisterCommonFlags(cmd)
 
 	// xops-enhanced flags (long-form only, no short flags to avoid OpenSSH conflicts)
 	cmd.Flags().StringVar(&o.Host, "host", "", i18n.T("flag_hosts"))
@@ -245,6 +247,17 @@ func (o *ExecOptions) completeNonDashArgs(cmd *cobra.Command, args []string) err
 
 func (o *ExecOptions) Complete(cmd *cobra.Command, args []string) error {
 	o.args = args
+	o.appliedGeneralOpts = false
+	var flags *pflag.FlagSet
+	if cmd != nil {
+		flags = cmd.Flags()
+	}
+	if err := ApplyGeneralOptions(&o.SshOptions, o.control.Options, flags); err != nil {
+		return err
+	}
+	if _, err := o.control.ToSessionControlOptions(); err != nil {
+		return err
+	}
 	o.execution.capturePresence(cmd)
 	if cmd != nil && cmd.Flags().Changed("cmd") && o.Command == "" {
 		return fmt.Errorf("explicit command must not be empty")
@@ -303,6 +316,12 @@ func (o *ExecOptions) readStdinIfRequired(ctx context.Context) error {
 }
 
 func (o *ExecOptions) Validate() error {
+	if err := o.ensureGeneralOptionsApplied(); err != nil {
+		return err
+	}
+	if err := o.control.Validate(); err != nil {
+		return err
+	}
 	if o.OTP != "" && o.OTPStdin {
 		return errors.New(i18n.T("err_otp_mutually_exclusive"))
 	}
@@ -367,6 +386,9 @@ func (o *ExecOptions) Run() error {
 //
 //nolint:gocyclo
 func (o *ExecOptions) RunContext(ctx context.Context) (retErr error) {
+	if err := o.ensureGeneralOptionsApplied(); err != nil {
+		return err
+	}
 	configPath, keyPath, pathErr := utils.GetConfigFilePath()
 	if pathErr != nil {
 		return fmt.Errorf("get config file path failed: %w", pathErr)
@@ -429,11 +451,14 @@ func (o *ExecOptions) RunContext(ctx context.Context) (retErr error) {
 	}
 	var connector *ssh.Connector
 	if o.Interactive {
-		connector = newCLIConnectorWithAdapterOptions(provider, adpOpts, ssh.WithLogger(logger.DefaultLogger()), ssh.WithInteractionHandler(o.interaction))
+		connector = newCLIConnectorWithAdapterOptions(provider, o.interaction, adpOpts, ssh.WithLogger(logger.DefaultLogger()))
 	} else {
 		connector = newNonInteractiveConnector(provider, adpOpts, ssh.WithLogger(logger.DefaultLogger()))
 	}
 	defer func() {
+		if handoffErr := o.handoffConnectorMasters(ctx, connector); handoffErr != nil {
+			logger.DefaultLogger().Debugf("mux: master handoff failed: %v", handoffErr)
+		}
 		joinConnectorCloseError(&retErr, connector)
 	}()
 
@@ -516,13 +541,21 @@ func (o *ExecOptions) runInteractive(
 
 	var execErr error
 	if o.Sudo && o.execution.enabled() {
-		execution, optErr := o.cliExecutionConfig()
+		cliExec, optErr := o.cliExecutionConfig()
+		if optErr != nil {
+			return optErr
+		}
+		execution, optErr := ssh.ResolveExecution(cliExec, client.Execution())
 		if optErr != nil {
 			return optErr
 		}
 		execErr = client.RunInteractiveWithSudoExecution(ctx, cmd, execution)
-	} else if o.commandPlan != nil {
-		execErr = client.RunInteractivePlan(ctx, *o.commandPlan)
+	} else if o.commandPlan != nil || o.execution.enabled() {
+		plan, planErr := o.resolveCommandPlan(cmd, client.Execution())
+		if planErr != nil {
+			return fmt.Errorf("[%s] %w", task.host, planErr)
+		}
+		execErr = client.RunInteractivePlan(ctx, plan)
 	} else if o.Sudo {
 		execErr = client.RunInteractiveWithSudo(ctx, cmd)
 	} else {
@@ -558,11 +591,24 @@ func (o *ExecOptions) buildAdapterOptions(tasks []execHostTask, cfg *config.Conf
 	// A global policy applies to proxy-jump nodes, which do not have a task
 	// specific override. Explicit task credentials remain scoped to their target.
 	adpOpts = append(adpOpts, adapter.WithGlobalSessionAuth(adapter.SessionAuth{Remember: remember}))
+	o.targetNodeIDs = make([]string, 0, len(tasks))
 	for _, task := range tasks {
+		o.targetNodeIDs = append(o.targetNodeIDs, task.nodeID)
 		adpOpts = append(adpOpts, adapter.WithSessionAuthOverride(task.nodeID, adapter.SessionAuth{
-			Password: task.pass, Passphrase: task.passphrase, KeyPath: task.keyPath, Remember: remember, OTP: o.OTP,
+			Password: task.pass, Passphrase: task.passphrase, KeyPath: task.keyPath, Remember: remember, OTP: o.OTP, ProxyJump: o.JumpHost,
 		}))
 	}
+
+	// Session Control Overrides (Tier 1)
+	sessionCtrlOpts, err := o.control.ToSessionControlOptions()
+	if err != nil {
+		return nil, err
+	}
+	if !sessionCtrlOpts.IsEmpty() {
+		for _, task := range tasks {
+			adpOpts = append(adpOpts, adapter.WithSessionControlOverride(task.nodeID, sessionCtrlOpts))
+		}
+	}
 	if reg, regErr := utils.GetCredentialRegistry(cfg); regErr != nil {
 		return nil, fmt.Errorf("initialize credential resolver: %w", regErr)
 	} else if reg != nil {
@@ -629,9 +675,15 @@ func (o *ExecOptions) getOrCreateNode(ctx context.Context, repository *config.Re
 	return res.NodeID, updated, nil
 }
 
+func (o *ExecOptions) isPlannedCommandMismatch(execCmd string, isScript bool) bool {
+	return isScript ||
+		(o.commandPlan != nil && o.commandPlan.Command() != execCmd) ||
+		(o.commandPlan == nil && o.Command != execCmd)
+}
+
 func (o *ExecOptions) executeTask(ctx context.Context, connector *ssh.Connector, t execHostTask, execCmd string, isScript bool, totalTasks int, stdoutMu *sync.Mutex) (retErr error) {
-	if o.commandPlan != nil {
-		if isScript || o.commandPlan.Command() != execCmd {
+	if o.commandPlan != nil || (!isScript && o.execution.enabled()) {
+		if o.isPlannedCommandMismatch(execCmd, isScript) {
 			return fmt.Errorf("[%s] command changed after execution planning", t.host)
 		}
 		if !o.Sudo && !o.Stream && o.OutDir == "" {
@@ -688,35 +740,19 @@ func (o *ExecOptions) executeTask(ctx context.Context, connector *ssh.Connector,
 }
 
 func (o *ExecOptions) cliExecutionConfig() (*ssh.ExecutionConfig, error) {
-	if !o.execution.enabled() && !o.execution.noLoginSet {
-		return nil, nil
-	}
-	if o.execution.noLoginSet {
-		// --no-login keeps its legacy meaning of explicitly selecting Bash.
-		opts, err := o.execution.commandOptions(o.NoLoginShell)
-		if err != nil {
-			return nil, err
-		}
-		login := opts.Login == ssh.LoginEnabled
-		return &ssh.ExecutionConfig{Interpreter: opts.Interpreter, LaunchDialect: opts.LaunchDialect, Login: &login}, nil
-	}
-	if (o.execution.interpreterSet && o.execution.interpreter == "") || (o.execution.dialectSet && o.execution.dialect == "") {
-		return nil, fmt.Errorf("explicit interpreter and launch dialect must not be empty")
-	}
-	// Missing fields must stay missing: script shebangs and node configuration
-	// resolve them later. Selecting a dialect is not selecting server.
-	cliExec := &ssh.ExecutionConfig{}
-	if o.execution.interpreterSet {
-		cliExec.Interpreter = ssh.Interpreter(o.execution.interpreter)
-	}
-	if o.execution.dialectSet {
-		cliExec.LaunchDialect = ssh.LaunchDialect(o.execution.dialect)
+	return o.execution.toExecutionConfig(o.NoLoginShell)
+}
+
+func usesWindowsCommandPlanAdapter(execCfg *ssh.ExecutionConfig) bool {
+	if execCfg == nil {
+		return false
 	}
-	if o.execution.loginSet {
-		login := o.execution.login
-		cliExec.Login = &login
+	switch execCfg.Interpreter {
+	case ssh.InterpreterPowerShell, ssh.InterpreterPwsh, ssh.InterpreterCmd:
+		return true
+	default:
+		return false
 	}
-	return cliExec, cliExec.Validate()
 }
 
 func (o *ExecOptions) dispatchTaskExecution(ctx context.Context, client *ssh.Client, execCmd string, isScript bool, outputOpts []ssh.RunOption) (string, error) {
@@ -739,19 +775,36 @@ func (o *ExecOptions) dispatchTaskExecution(ctx context.Context, client *ssh.Cli
 		return client.RunScriptWithExecution(ctx, execCmd, resolvedExec, scriptOpts...)
 
 	}
-	if o.commandPlan != nil {
+	if o.commandPlan != nil || o.execution.enabled() {
 		if o.Sudo {
-			execution, err := o.cliExecutionConfig()
+			cliExec, err := o.cliExecutionConfig()
+			if err != nil {
+				return "", err
+			}
+			execution, err := ssh.ResolveExecution(cliExec, client.Execution())
 			if err != nil {
 				return "", err
 			}
 			return client.RunWithSudoExecution(ctx, execCmd, execution, outputOpts...)
 		}
+		plan, err := o.resolveCommandPlan(execCmd, client.Execution())
+		if err != nil {
+			return "", err
+		}
 		// The interpreter has already been applied by PlanCommand. Send its
 		// frozen payload unchanged through the existing output-mode adapters.
 		opts := append([]ssh.RunOption{}, outputOpts...)
 		opts = append(opts, ssh.WithInterpreter(ssh.InterpreterServer), ssh.WithLaunchDialect(ssh.LaunchUnknown))
-		return client.Run(ctx, o.commandPlan.Payload(), opts...)
+		return client.Run(ctx, plan.Payload(), opts...)
+	}
+	if !o.Sudo && !o.execution.noLoginSet && !o.NoLoginShell && usesWindowsCommandPlanAdapter(client.Execution()) {
+		plan, err := ssh.PlanCommand(execCmd, client.Execution().CommandOptions())
+		if err != nil {
+			return "", fmt.Errorf("plan command: %w", err)
+		}
+		opts := append([]ssh.RunOption{}, outputOpts...)
+		opts = append(opts, ssh.WithInterpreter(ssh.InterpreterServer), ssh.WithLaunchDialect(ssh.LaunchUnknown))
+		return client.Run(ctx, plan.Payload(), opts...)
 	}
 	cmdOpts := o.commandRunOptions(client.Execution(), outputOpts)
 	if o.Sudo {
@@ -859,6 +912,9 @@ func sanitizeHostForFilename(host string) string {
 }
 
 func (o *ExecOptions) buildTasksFromTags(provider config.ConfigProvider) ([]execHostTask, error) {
+	if err := o.ensureGeneralOptionsApplied(); err != nil {
+		return nil, err
+	}
 	var tasks []execHostTask
 	nodes := provider.GetNodesByTag(o.Tag)
 	if len(nodes) == 0 {
@@ -892,6 +948,9 @@ func (o *ExecOptions) buildTasksFromTags(provider config.ConfigProvider) ([]exec
 }
 
 func (o *ExecOptions) buildTasksFromHosts(ctx context.Context, repository *config.Repository) ([]execHostTask, []error, error) {
+	if err := o.ensureGeneralOptionsApplied(); err != nil {
+		return nil, nil, err
+	}
 	var tasks []execHostTask
 	var hostErrs []error
 	hosts, err := utils.ParseHosts(o.Host, o.HostFile, "")
@@ -978,6 +1037,14 @@ func appendExecAlias(slice []string, val string) ([]string, bool) {
 }
 
 func (o *ExecOptions) updateNodeFromHostInfo(ctx context.Context, nodeID string, repository *config.Repository, addr utils.HostInfo) (bool, error) {
+	if strings.HasPrefix(nodeID, config.OpenSSHNodePrefix) {
+		if o.JumpHost != "" {
+			if _, err := repository.ResolveProxyJumpChain(o.JumpHost); err != nil {
+				return false, fmt.Errorf("resolve exec proxy jump %q failed: %w", o.JumpHost, err)
+			}
+		}
+		return false, nil
+	}
 	view := repository.View()
 	ref, ok := view.NodeRefs[nodeID]
 	if !ok {