diff --git a/tests/pytests/integration/ssh/test_deploy.py b/tests/pytests/integration/ssh/test_deploy.py
index 8512813f6e60..9b2779bf7343 100644
--- a/tests/pytests/integration/ssh/test_deploy.py
+++ b/tests/pytests/integration/ssh/test_deploy.py
@@ -28,6 +28,119 @@ def thin_dir(salt_ssh_cli):
         shutil.rmtree(thin_dir_path, ignore_errors=True)
 
 
+@pytest.fixture(scope="module")
+def invalid_json_exe_mod(salt_run_cli, base_env_state_tree_root_dir):
+    module_contents = r"""
+import os
+import sys
+
+
+def __virtual__():
+    return "whoops"
+
+
+def test():
+    data = '{\n  "local": {\n    "whoops": "hrhrhr"\n  }\n}'
+    ctr = 0
+    for line in data.splitlines():
+        sys.stdout.write(line)
+        if ctr == 3:
+            print("Warning: Chaos is not a letter")
+        ctr += 1
+    sys.stdout.flush()
+    os._exit(0)
+"""
+    module_dir = base_env_state_tree_root_dir / "_modules"
+    module_tempfile = pytest.helpers.temp_file("whoops.py", module_contents, module_dir)
+    try:
+        with module_tempfile:
+            ret = salt_run_cli.run("saltutil.sync_modules")
+            assert ret.returncode == 0
+            assert "modules.whoops" in ret.data
+            yield
+    finally:
+        ret = salt_run_cli.run("saltutil.sync_modules")
+        assert ret.returncode == 0
+
+
+@pytest.fixture(scope="module")
+def invalid_return_exe_mod(salt_run_cli, base_env_state_tree_root_dir):
+    module_contents = r"""
+import json
+import os
+import sys
+
+
+def __virtual__():
+    return "whoopsiedoodle"
+
+
+def test(wrapped=True):
+    data = "Chaos is a ladder though"
+    if wrapped:
+        data = {"local": {"no_return_key_present": data}}
+    else:
+        data = {"no_local_key_present": data}
+
+    print(json.dumps(data))
+    sys.stdout.flush()
+    os._exit(0)
+"""
+    module_dir = base_env_state_tree_root_dir / "_modules"
+    module_tempfile = pytest.helpers.temp_file(
+        "whoopsiedoodle.py", module_contents, module_dir
+    )
+    try:
+        with module_tempfile:
+            ret = salt_run_cli.run("saltutil.sync_modules")
+            assert ret.returncode == 0
+            assert "modules.whoopsiedoodle" in ret.data
+            yield
+    finally:
+        ret = salt_run_cli.run("saltutil.sync_modules")
+        assert ret.returncode == 0
+
+
+@pytest.fixture(scope="module")
+def remote_exception_wrap_mod(salt_master):
+    module_contents = r"""
+def __virtual__():
+    return "check_exception"
+
+
+def failure():
+    # This should raise an exception
+    ret = __salt__["disk.usage"]("c")
+    return f"Probably got garbage: {ret}"
+"""
+    module_dir = pathlib.Path(salt_master.config["extension_modules"]) / "wrapper"
+    module_tempfile = pytest.helpers.temp_file(
+        "check_exception.py", module_contents, module_dir
+    )
+    with module_tempfile:
+        yield
+
+
+@pytest.fixture(scope="module")
+def remote_parsing_failure_wrap_mod(salt_master, invalid_json_exe_mod):
+    module_contents = r"""
+def __virtual__():
+    return "check_parsing"
+
+
+def failure(mod):
+    # This should raise an exception
+    ret = __salt__[f"{mod}.test"]()
+    return f"Probably got garbage: {ret}"
+"""
+    module_dir = pathlib.Path(salt_master.config["extension_modules"]) / "wrapper"
+    module_tempfile = pytest.helpers.temp_file(
+        "check_parsing.py", module_contents, module_dir
+    )
+    with module_tempfile:
+        yield
+
+
 def test_ping(salt_ssh_cli):
     """
     Test a simple ping
@@ -122,3 +235,65 @@ def test_retcode_exe_run_exception(salt_ssh_cli):
     assert isinstance(ret.data, dict)
     assert ret.data["stderr"].endswith("Exception: hehehe")
     assert ret.data["retcode"] == 1
+
+
+@pytest.mark.usefixtures("invalid_json_exe_mod")
+def test_retcode_json_decode_error(salt_ssh_cli):
+    """
+    Verify salt-ssh exits with a non-zero exit code when
+    it cannot decode the output of a command.
+    """
+    ret = salt_ssh_cli.run("whoops.test")
+    assert ret.returncode == EX_AGGREGATE
+    assert isinstance(ret.data, dict)
+    assert (
+        ret.data["stdout"]
+        == '{  "local": {    "whoops": "hrhrhr"  }Warning: Chaos is not a letter\n}'
+    )
+    assert ret.data["retcode"] == 0
+
+
+@pytest.mark.usefixtures("invalid_return_exe_mod")
+def test_retcode_invalid_return(salt_ssh_cli):
+    """
+    Verify salt-ssh exits with a non-zero exit code when
+    the decoded command output is invalid.
+    """
+    ret = salt_ssh_cli.run("whoopsiedoodle.test", "false")
+    assert ret.returncode == EX_AGGREGATE
+    assert isinstance(ret.data, dict)
+    assert ret.data["stdout"] == '{"no_local_key_present": "Chaos is a ladder though"}'
+    assert ret.data["retcode"] == 0
+
+
+@pytest.mark.usefixtures("remote_exception_wrap_mod")
+def test_wrapper_unwrapped_command_exception(salt_ssh_cli):
+    """
+    Verify salt-ssh does not return unexpected exception output to wrapper modules.
+    """
+    ret = salt_ssh_cli.run("check_exception.failure")
+    assert ret.data
+    assert "Probably got garbage" not in ret.data
+    assert ret.returncode == EX_AGGREGATE
+
+
+@pytest.mark.usefixtures("remote_parsing_failure_wrap_mod", "invalid_json_exe_mod")
+def test_wrapper_unwrapped_command_parsing_failure(salt_ssh_cli):
+    """
+    Verify salt-ssh does not return unexpected unparsable output to wrapper modules.
+    """
+    ret = salt_ssh_cli.run("check_parsing.failure", "whoops")
+    assert ret.data
+    assert "Probably got garbage" not in ret.data
+    assert ret.returncode == EX_AGGREGATE
+
+
+@pytest.mark.usefixtures("remote_parsing_failure_wrap_mod", "invalid_return_exe_mod")
+def test_wrapper_unwrapped_command_invalid_return(salt_ssh_cli):
+    """
+    Verify salt-ssh does not return unexpected unparsable output to wrapper modules.
+    """
+    ret = salt_ssh_cli.run("check_parsing.failure", "whoopsiedoodle")
+    assert ret.data
+    assert "Probably got garbage" not in ret.data
+    assert ret.returncode == EX_AGGREGATE
diff --git a/tests/pytests/integration/ssh/test_mine.py b/tests/pytests/integration/ssh/test_mine.py
index 7bb5bb10822a..878d461f49d1 100644
--- a/tests/pytests/integration/ssh/test_mine.py
+++ b/tests/pytests/integration/ssh/test_mine.py
@@ -8,6 +8,31 @@
 ]
 
 
+@pytest.fixture(scope="module", autouse=True)
+def pillar_tree(base_env_pillar_tree_root_dir):
+    top_file = """
+    base:
+      'localhost':
+        - mine
+      '127.0.0.1':
+        - mine
+    """
+    mine_pillar_file = """
+    mine_functions:
+      disk.usage:
+        - c
+    """
+    top_tempfile = pytest.helpers.temp_file(
+        "top.sls", top_file, base_env_pillar_tree_root_dir
+    )
+    mine_tempfile = pytest.helpers.temp_file(
+        "mine.sls", mine_pillar_file, base_env_pillar_tree_root_dir
+    )
+
+    with top_tempfile, mine_tempfile:
+        yield
+
+
 @pytest.fixture(autouse=True)
 def thin_dir(salt_ssh_cli):
     try:
@@ -29,3 +54,13 @@ def test_ssh_mine_get(salt_ssh_cli):
     assert "localhost" in ret.data
     assert "args" in ret.data["localhost"]
     assert ret.data["localhost"]["args"] == ["itworked"]
+
+
+def test_ssh_mine_get_error(salt_ssh_cli):
+    """
+    Test that a mine function returning an error is not
+    included in the output.
+    """
+    ret = salt_ssh_cli.run("mine.get", "localhost", "disk.usage")
+    assert ret.returncode == 0
+    assert not ret.data
diff --git a/tests/pytests/integration/ssh/test_state.py b/tests/pytests/integration/ssh/test_state.py
index 5f9bfb45e9f4..0e5446c59c01 100644
--- a/tests/pytests/integration/ssh/test_state.py
+++ b/tests/pytests/integration/ssh/test_state.py
@@ -200,6 +200,30 @@ def pillar_tree_render_fail(base_env_pillar_tree_root_dir):
         yield
 
 
+@pytest.fixture(scope="class")
+def state_tree_render_module_exception(base_env_state_tree_root_dir):
+    top_file = """
+    base:
+      'localhost':
+        - fail_module_exception
+      '127.0.0.1':
+        - fail_module_exception
+    """
+    state_file = r"""
+    This should fail being rendered:
+      test.show_notification:
+        - text: {{ salt["disk.usage"]("c") | yaml_dquote }}
+    """
+    top_tempfile = pytest.helpers.temp_file(
+        "top.sls", top_file, base_env_state_tree_root_dir
+    )
+    state_tempfile = pytest.helpers.temp_file(
+        "fail_module_exception.sls", state_file, base_env_state_tree_root_dir
+    )
+    with top_tempfile, state_tempfile:
+        yield
+
+
 @pytest.mark.slow_test
 def test_state_with_import(salt_ssh_cli, state_tree):
     """
@@ -561,3 +585,152 @@ def test_retcode_state_sls_id_render_exception(self, salt_ssh_cli):
     def test_retcode_state_top_run_fail(self, salt_ssh_cli):
         ret = salt_ssh_cli.run("state.top", "top.sls")
         assert ret.returncode == EX_AGGREGATE
+
+
+@pytest.mark.slow_test
+@pytest.mark.usefixtures("state_tree_render_module_exception")
+class TestRenderModuleException:
+    """
+    Verify salt-ssh stops state execution and fails with a retcode > 0
+    when a state rendering fails because an execution module throws an exception.
+    """
+
+    def test_retcode_state_sls_render_module_exception(self, salt_ssh_cli):
+        ret = salt_ssh_cli.run("state.sls", "fail_module_exception")
+        self._assert_ret(ret, EX_AGGREGATE)
+
+    def test_retcode_state_highstate_render_module_exception(self, salt_ssh_cli):
+        ret = salt_ssh_cli.run("state.highstate")
+        self._assert_ret(ret, EX_AGGREGATE)
+
+    def test_retcode_state_sls_id_render_module_exception(self, salt_ssh_cli):
+        ret = salt_ssh_cli.run("state.sls_id", "foo", "fail_module_exception")
+        self._assert_ret(ret, EX_AGGREGATE)
+
+    def test_retcode_state_show_sls_render_module_exception(self, salt_ssh_cli):
+        ret = salt_ssh_cli.run("state.show_sls", "fail_module_exception")
+        self._assert_ret(ret, EX_AGGREGATE)
+
+    def test_retcode_state_show_low_sls_render_module_exception(self, salt_ssh_cli):
+        ret = salt_ssh_cli.run("state.show_low_sls", "fail_module_exception")
+        self._assert_ret(ret, EX_AGGREGATE)
+
+    def test_retcode_state_show_highstate_render_module_exception(self, salt_ssh_cli):
+        ret = salt_ssh_cli.run("state.show_highstate")
+        self._assert_ret(ret, EX_AGGREGATE)
+
+    def test_retcode_state_show_lowstate_render_module_exception(self, salt_ssh_cli):
+        ret = salt_ssh_cli.run("state.show_lowstate")
+        # state.show_lowstate exits with 0 for non-ssh as well
+        self._assert_ret(ret, 0)
+
+    def test_retcode_state_top_render_module_exception(self, salt_ssh_cli):
+        ret = salt_ssh_cli.run("state.top", "top.sls")
+        self._assert_ret(ret, EX_AGGREGATE)
+
+    def _assert_ret(self, ret, retcode):
+        assert ret.returncode == retcode
+        assert isinstance(ret.data, list)
+        assert ret.data
+        assert isinstance(ret.data[0], str)
+        # While these three should usually follow each other, there
+        # can be warnings in between that would break such a logic.
+        assert "Rendering SLS 'base:fail_module_exception' failed:" in ret.data[0]
+        assert "Problem running salt function in Jinja template:" in ret.data[0]
+        assert (
+            "Error running 'disk.usage': Invalid flag passed to disk.usage"
+            in ret.data[0]
+        )
+
+
+@pytest.fixture(scope="class")
+def state_tree_remote_exception_mod(salt_run_cli, base_env_state_tree_root_dir):
+    module_contents = r"""
+import os
+import sys
+
+
+def __virtual__():
+    return "whoops"
+
+
+def do_stuff(name):
+    print("Hi there, nice that you're trying to make me do stuff.", file=sys.stderr)
+    print("Traceback (most recent call last):", file=sys.stderr)
+    print('  File "/dont/treat/me/like/that.py" in buzz_off', file=sys.stderr)
+    print("ComplianceError: 'Outlaw detected'", file=sys.stderr)
+    sys.stdout.flush()
+    os._exit(0)
+"""
+    module_dir = base_env_state_tree_root_dir / "_states"
+    module_tempfile = pytest.helpers.temp_file("whoops.py", module_contents, module_dir)
+    try:
+        with module_tempfile:
+            ret = salt_run_cli.run("saltutil.sync_states")
+            assert ret.returncode == 0
+            assert "states.whoops" in ret.data
+            yield
+    finally:
+        ret = salt_run_cli.run("saltutil.sync_states")
+        assert ret.returncode == 0
+
+
+@pytest.fixture(scope="class")
+def state_tree_remote_exception(
+    base_env_state_tree_root_dir, state_tree_remote_exception_mod
+):
+    top_file = """
+    base:
+      'localhost':
+        - remote_stacktrace
+      '127.0.0.1':
+        - remote_stacktrace
+    """
+    state_file = r"""
+    This should be detected as failure:
+      whoops.do_stuff
+    """
+    top_tempfile = pytest.helpers.temp_file(
+        "top.sls", top_file, base_env_state_tree_root_dir
+    )
+    state_tempfile = pytest.helpers.temp_file(
+        "remote_stacktrace.sls", state_file, base_env_state_tree_root_dir
+    )
+    with top_tempfile, state_tempfile:
+        yield
+
+
+@pytest.mark.slow_test
+@pytest.mark.usefixtures("state_tree_remote_exception")
+class TestStateRunRemoteExceptionRetcode:
+    """
+    Verify salt-ssh does not report success when it cannot parse
+    the return value.
+    """
+
+    def test_retcode_state_sls_remote_exception(self, salt_ssh_cli):
+        ret = salt_ssh_cli.run("state.sls", "remote_stacktrace")
+        assert ret.returncode == EX_AGGREGATE
+        assert ret.data
+
+    def test_retcode_state_highstate_remote_exception(self, salt_ssh_cli):
+        ret = salt_ssh_cli.run("state.highstate")
+        assert ret.returncode == EX_AGGREGATE
+        assert ret.data
+
+    def test_retcode_state_sls_id_remote_exception(self, salt_ssh_cli):
+        ret = salt_ssh_cli.run(
+            "state.sls_id", "This should be detected as failure", "remote_stacktrace"
+        )
+        assert ret.returncode == EX_AGGREGATE
+        assert ret.data
+
+    def test_retcode_state_top_remote_exception(self, salt_ssh_cli):
+        ret = salt_ssh_cli.run("state.top", "top.sls")
+        assert ret.returncode == EX_AGGREGATE
+        assert ret.data
+
+    def test_retcode_state_single_remote_exception(self, salt_ssh_cli):
+        ret = salt_ssh_cli.run("state.single", "whoops.do_stuff", "now")
+        assert ret.returncode == EX_AGGREGATE
+        assert ret.data