diff --git a/changelogs/fragments/doctospec.yml b/changelogs/fragments/doctospec.yml
new file mode 100644
index 00000000000..4929385484c
--- /dev/null
+++ b/changelogs/fragments/doctospec.yml
@@ -0,0 +1,4 @@
+minor_changes:
+  - Expanded documentation with the ability to specify paramter relations, like required_if, mutually_exclusive, required_toghether, etc
+  - Added function that will take a module documentation and transform into an argspec
+  - Action plugins base now can use doc_to_spec functino to validate parameters on controller (no need to round trip)
diff --git a/lib/ansible/module_utils/parsing/docs_to_spec.py b/lib/ansible/module_utils/parsing/docs_to_spec.py
new file mode 100644
index 00000000000..8b4084e0c68
--- /dev/null
+++ b/lib/ansible/module_utils/parsing/docs_to_spec.py
@@ -0,0 +1,164 @@
+# Copyright: Contributors to the Ansible project
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+"""
+Utilities to create module/plugin specs from their documentation.
+
+# example usage:
+
+    #prep
+    argpsec = get_options_from_docs(doc.get('options', {}))
+    restrictions = get_restrictions_from_doc(doc.get('restrictions', {}))
+
+    # do
+    validated = validate_spec(argspec, restrictions, task_params)
+
+    # error handle
+    if valided.error_messages:
+        raise ActionFail({'msg': 'Validation of arguments failed:\n%s' % '\n'.join(validated.error_messages), 'argument_errors': validatec.error_messages})
+
+    # get info
+    final_params = valided.validated_parameters
+    no_log_values = valided._no_log_values
+    aliases = validated._aliases
+
+'''
+example of DOCUMENTATION with requirements:
+
+    options:
+         ...jkk
+    notes:
+         ...
+    requirements:
+         ...
+    restrictions:
+        # mutually_exclusive
+       - description: You cannot use 'a' and 'b' at the same time
+         exclusive: a, b
+       - description: You cannot use 'c' and 'x' at the same time
+         exclusive: c, x
+
+        # required_together
+       - description: 'a' and 'b' required together
+         together: a, b
+
+        # required_one_of
+       - description: at least one of a or b is required
+         one_of: a, b
+
+         # required_if
+       - description: if x is set to y, a,b and c are required
+         required: [a,b,c]
+         if: x
+         equals: y
+
+         # required_by
+       - required: x
+         description: x is required if b or c are set
+         by: [b,c]
+'''
+"""
+from __future__ import annotations
+
+import typing as _t
+
+from ansible.module_utils.common.arg_spec import ArgumentSpecValidator
+from ansible.module_utils.common._collections_compat import Sequence
+from ansible.module_utils.six import string_types
+
+
+ARGS_DOCS_KEYS = ("aliases", "choices", "default", "elements", "no_log", "required", "type")
+
+
+def option_to_spec(option, deprecate=None) -> dict:
+    """ convert from doc option entry to spec arg definition """
+
+    # use known common keys to copy data
+    spec = {name: option[name] for name in ARGS_DOCS_KEYS if name in option}
+
+    # handle suboptions
+    if "suboptions" in option:
+        add_options_from_doc(spec, option["suboptions"], deprecate)
+
+        for sub in spec["options"].values():
+            # check if we need to apply_defults
+            if "default" in sub or "fallback" in sub:
+                spec["apply_defaults"] = True
+
+    # Use passed-in function to handle deprecations
+    if deprecate is not None and 'deprecated' in option:
+        deprecate(**option['deprecated'])
+
+    return spec
+
+
+def restriction_to_spec(r):
+    """ read documented restriction and create spec restriction """
+
+    name = None
+    rest = None  # normally a list except for 'required_by'
+    if 'required' in r:
+        if 'by' in r:
+            name = 'required_by'
+            rest = {r['required']: r['by']}
+        elif 'if' in r:
+            name = 'required_if'
+            rest = [r['if'], r['equals'], r['required']]
+    else:
+        for ding in ('exclusive', 'together', 'one_of'):
+            if ding in r:
+
+                if isinstance(r[ding], string_types):
+                    rest = r[ding].spit(',')
+                elif not isinstance(r[ding], Sequence):
+                    raise TypeError('must be a list!')
+                else:
+                    rest = r[ding]
+
+                if len(rest) < 2:
+                    raise TypeError('must have multiple elements')
+
+                if ding == 'exclusive':
+                    name = 'mutually_exclusive'
+                else:
+                    name = 'required_%s' % ding
+                break
+            else:
+                raise Exception('unknown restriction!')
+    return rest
+
+
+def add_options_from_doc(argspec, options, deprecate=None):
+    """ Add option doc entries into argspec """
+    for n, o in sorted(options.items()):
+        argspec[n] = option_to_spec(o, deprecate)
+
+
+def get_options_from_doc(options, deprecate=None):
+    """ Add option doc entries into argspec """
+    argspec = {}
+    add_options_from_doc(argspec, options, deprecate)
+    return argspec
+
+
+def add_restrictions_from_doc(restrict_args, restrictions):
+    """ add restriction doc entries into argspec """
+    for r in restrictions:
+        restrict_args.append(restriction_to_spec(r))
+
+
+def get_restrictions_from_doc(restrictions):
+    """ add restriction doc entries into argspec """
+    reargs = {}
+    add_restrictions_from_doc(reargs, restrictions)
+    return reargs
+
+
+def validate_spec(spec, restrictions, task_args):
+
+    validator = ArgumentSpecValidator(spec, **restrictions)
+    return validator.validate(task_args)
+
+
+def validate_spec_from_plugin(plugin):
+    # take plugin object (name?), get docs and process with above
+    pass
diff --git a/lib/ansible/plugins/action/__init__.py b/lib/ansible/plugins/action/__init__.py
index e296398622b..ba7fd5173d2 100644
--- a/lib/ansible/plugins/action/__init__.py
+++ b/lib/ansible/plugins/action/__init__.py
@@ -47,6 +47,7 @@ if t.TYPE_CHECKING:
     from ansible.playbook.play_context import PlayContext
     from ansible.playbook.task import Task
     from ansible.plugins.connection import ConnectionBase
+    from ansible.plugins.loader import PluginLoadContext
     from ansible.template import Templar
 
 
@@ -101,6 +102,28 @@ class ActionBase(ABC, _AnsiblePluginInfoMixin):
         # Backwards compat: self._display isn't really needed, just import the global display and use that.
         self._display = display
 
+        self._found: dict[str, PluginLoadContext] = {}
+
+    def _ensure_invocation(self, result):
+
+        need_invocation = bool(display.verbosity or C.DEFAULT_DEBUG)
+        if need_invocation and 'invocation' not in result:
+            result['invocation'] = {}
+
+        if isinstance(result.get('invocation'), dict):
+            # we have invocation dict, otherwise it would be censored string already
+            # action plugins/modules might have popuilated
+            if self._task.no_log:
+                result['invocation'] = "CENSORED: no_log is set for this task"
+            else:
+                result["invocation"]['module_args'] = self._task.args.copy()
+
+                # TODO: get no_log list from controller side argspec, example for copy
+                if result['invocation']['module_args'].get('content') is not None:
+                    result['invocation']['module_args']['content'] = 'CENSORED: no_log is set for this parameter'
+
+        return result
+
     @abstractmethod
     def run(self, tmp: str | None = None, task_vars: dict[str, t.Any] | None = None) -> dict[str, t.Any]:
         """ Action Plugins should implement this method to perform their
@@ -118,8 +141,7 @@ class ActionBase(ABC, _AnsiblePluginInfoMixin):
 
         * Module parameters.  These are stored in self._task.args
         """
-        # does not default to {'changed': False, 'failed': False}, as it used to break async
-        result: dict[str, t.Any] = {}
+        result: dict[str, t.Any] = {'changed': False, 'failed': False}
 
         if tmp is not None:
             display.warning('ActionModule.run() no longer honors the tmp parameter. Action'
@@ -142,7 +164,7 @@ class ActionBase(ABC, _AnsiblePluginInfoMixin):
         if self._connection._shell.tmpdir is None and self._early_needs_tmp_path():
             self._make_tmp_path()
 
-        return result
+        return self._ensure_invocation(result)
 
     def validate_argument_spec(self, argument_spec=None,
                                mutually_exclusive=None,
@@ -253,6 +275,48 @@ class ActionBase(ABC, _AnsiblePluginInfoMixin):
             return True
         return False
 
+    def _get_module(self, module_name: str, module_args: dict | None = None) -> PluginLoadContext:
+
+        if module_name not in self._found:
+            split_module_name = module_name.split('.')
+            collection_name = '.'.join(split_module_name[0:2]) if len(split_module_name) > 2 else ''
+            leaf_module_name = resource_from_fqcr(module_name)
+
+            # Search module path(s) for named module.
+            for mod_type in self._connection.module_implementation_preferences:
+                # Check to determine if PowerShell modules are supported, and apply
+                # some fixes (hacks) to module name + args.
+                if mod_type == '.ps1':
+                    # FIXME: This should be temporary and moved to an exec subsystem plugin where we can define the mapping
+                    # for each subsystem.
+                    win_collection = 'ansible.windows'
+                    rewrite_collection_names = ['ansible.builtin', 'ansible.legacy', '']
+                    # async_status, win_stat, win_file, win_copy, and win_ping are not just like their
+                    # python counterparts but they are compatible enough for our
+                    # internal usage
+                    # NB: we only rewrite the module if it's not being called by the user (eg, an action calling something else)
+                    # and if it's unqualified or FQ to a builtin
+                    if leaf_module_name in ('stat', 'file', 'copy', 'ping') and \
+                            collection_name in rewrite_collection_names and self._task.action != module_name:
+                        module_name = '%s.win_%s' % (win_collection, leaf_module_name)
+                    elif leaf_module_name == 'async_status' and collection_name in rewrite_collection_names:
+                        module_name = '%s.%s' % (win_collection, leaf_module_name)
+
+                result = self._shared_loader_obj.module_loader.find_plugin_with_context(module_name, mod_type, collection_list=self._task.collections)
+                if result.resolved:
+                    self._found[module_name] = result
+                    break
+                else:
+                    if result.redirect_list and len(result.redirect_list) > 1:
+                        # take the last one in the redirect list, we may have successfully jumped through N other redirects
+                        target_module_name = result.redirect_list[-1]
+                        raise AnsibleError("The module {0} was redirected to {1}, which could not be loaded.".format(module_name, target_module_name))
+
+            else:  # This is a for-else: http://bit.ly/1ElPkyg
+                raise AnsibleError("The module %s was not found in configured module paths" % (module_name))
+
+        return self._found[module_name]
+
     def _configure_module(self, module_name, module_args, task_vars) -> tuple[_BuiltModule, str]:
         """
         Handles the loading and templating of the module code through the
@@ -263,9 +327,9 @@ class ActionBase(ABC, _AnsiblePluginInfoMixin):
         else:
             use_vars = task_vars
 
-        split_module_name = module_name.split('.')
-        collection_name = '.'.join(split_module_name[0:2]) if len(split_module_name) > 2 else ''
-        leaf_module_name = resource_from_fqcr(module_name)
+        # get module path and unquote args if needed
+        result = self._get_module(module_name, module_args)
+        module_path = result.plugin_resolved_path
 
         # Search module path(s) for named module.
         for mod_type in self._connection.module_implementation_preferences:
@@ -302,6 +366,15 @@ class ActionBase(ABC, _AnsiblePluginInfoMixin):
         else:  # This is a for-else: http://bit.ly/1ElPkyg
             raise AnsibleError("The module %s was not found in configured module paths" % (module_name))
 
+        # TODO: move this tweak down to the modules, (platform: windows attribute?), not extensible here
+        # Remove extra quotes surrounding path parameters before sending to module.
+        if resource_from_fqcr(module_name) in ['win_stat', 'win_file', 'win_copy', 'slurp'] and \
+                module_args and \
+                hasattr(self._connection._shell, '_unquote'):
+            for key in ('src', 'dest', 'path'):
+                if key in module_args:
+                    module_args[key] = self._connection._shell._unquote(module_args[key])
+
         # insert shared code and arguments into the module
         final_environment: dict[str, t.Any] = {}
         self._compute_environment_string(final_environment)
@@ -347,6 +420,15 @@ class ActionBase(ABC, _AnsiblePluginInfoMixin):
 
         return module_bits, module_path
 
+    def _get_argspec_from_docs(self, module_name):
+        resolved = self._get_module(module_name)
+        argspec = '' + resolved  # TODO: actuall call build function
+        return argspec
+
+    def _validate_args(self, module_name, module_args):
+        argspec = self._get_argspec_from_docs(module_name)
+        # TODO: validate(argpsec, module_args)
+
     def _compute_environment_string(self, raw_environment_out=None):
         """
         Builds the environment string to be used when executing the remote task.
diff --git a/lib/ansible/plugins/action/async_status.py b/lib/ansible/plugins/action/async_status.py
index 676fc9324ec..a7c02cee9ea 100644
--- a/lib/ansible/plugins/action/async_status.py
+++ b/lib/ansible/plugins/action/async_status.py
@@ -20,6 +20,10 @@ class ActionModule(ActionBase):
 
         results = super(ActionModule, self).run(tmp, task_vars)
 
+        # async works diff from other action plugins
+        for nope in ('changed', 'failed'):
+            del results[nope]
+
         validation_result, new_module_args = self.validate_argument_spec(
             argument_spec={
                 'jid': {'type': 'str', 'required': True},
diff --git a/lib/ansible/plugins/action/copy.py b/lib/ansible/plugins/action/copy.py
index d3c1f0b2fc4..658f97ac953 100644
--- a/lib/ansible/plugins/action/copy.py
+++ b/lib/ansible/plugins/action/copy.py
@@ -206,22 +206,19 @@ class ActionModule(ActionBase):
         # NOTE: adding invocation arguments here needs to be kept in sync with
         # any no_log specified in the argument_spec in the module.
         # This is not automatic.
-        # NOTE: do not add to this. This should be made a generic function for action plugins.
-        # This should also use the same argspec as the module instead of keeping it in sync.
-        if 'invocation' not in result:
-            if self._task.no_log:
-                result['invocation'] = "CENSORED: no_log is set"
-            else:
-                # NOTE: Should be removed in the future. For now keep this broken
-                # behaviour, have a look in the PR 51582
-                result['invocation'] = self._task.args.copy()
-                result['invocation']['module_args'] = self._task.args.copy()
+        # NOTE: DO NOT ADD TO THIS. There is a generic function for action plugins.
+        if self._task.no_log:
+            result['invocation'] = "CENSORED: no_log is set"
+        else:
+            # NOTE: Should be removed in the future. For now keep this broken
+            # behaviour, have a look in the PR 51582, deprecate once DT is available
+            # TAGS: DT
+            result['invocation'] = self._task.args.copy()
 
         if isinstance(result['invocation'], dict):
+            # NOTE: also deprecate and remove once we have DT
             if 'content' in result['invocation']:
                 result['invocation']['content'] = 'CENSORED: content is a no_log parameter'
-            if result['invocation'].get('module_args', {}).get('content') is not None:
-                result['invocation']['module_args']['content'] = 'VALUE_SPECIFIED_IN_NO_LOG_PARAMETER'
 
         return result