preserve previous behavior

core/dbt/task/run.py

@@ -775,8 +775,8 @@ def after_run(self, adapter, results) -> None:
         extras = {
             "schemas": list({s for _, s in database_schema_set}),
             "results": [
-                r for r in results if r.thread_id != "main"
-            ],  # exclude hooks to preserve backwards compatibility
+                r for r in results if r.thread_id != "main" or r.status == RunStatus.Error
+            ],  # exclude that didn't fail to preserve backwards compatibility
             "database_schemas": list(database_schema_set),
         }
         with adapter.connection_named("master"):

core/dbt/task/runnable.py

@@ -512,7 +512,6 @@ def execute_with_hooks(self, selected_uids: AbstractSet[str]):
         self.started_at = time.time()
         try:
             before_run_status = self.before_run(adapter, selected_uids)
-
             if before_run_status == RunStatus.Success or (
                 not get_flags().skip_nodes_if_on_run_start_fails
             ):

tests/functional/adapter/hooks/test_on_run_hooks.py

@@ -170,7 +170,7 @@ def test_results(self, project):
         assert run_results["results"] == []
 
 
-class Test__HookContext:
+class Test__HookContext__HookSuccess:
     @pytest.fixture(scope="class")
     def project_config_update(self):
         return {
@@ -200,9 +200,45 @@ def macros(self):
     def models(self):
         return {"my_model.sql": "select 1"}
 
-    def test_results_in_context(self, project):
+    def test_results_in_context_success(self, project):
         results, log_output = run_dbt_and_capture(["--debug", "run"])
         assert "Thread ID: " in log_output
         assert "Thread ID: main" not in log_output
+        assert results[0].thread_id == "main"  # hook still exists in run results
+        assert "Num Results in context: 1" in log_output  # only model given hook was successful
+
+
+class Test__HookContext__HookFail:
+    @pytest.fixture(scope="class")
+    def project_config_update(self):
+        return {
+            "on-run-start": [
+                "select a as id",  # fail
+            ],
+            "on-run-end": [
+                '{{ log("Num Results in context: " ~ results|length)}}'
+                "{{ output_thread_ids(results) }}",
+            ],
+        }
+
+    @pytest.fixture(scope="class")
+    def macros(self):
+        return {
+            "log.sql": """
+{% macro output_thread_ids(results) %}
+    {% for result in results %}
+        {{ log("Thread ID: " ~ result.thread_id) }}
+    {% endfor %}
+{% endmacro %}
+"""
+        }
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {"my_model.sql": "select 1"}
+
+    def test_results_in_context_hook_fail(self, project):
+        results, log_output = run_dbt_and_capture(["--debug", "run"], expect_pass=False)
+        assert "Thread ID: main" in log_output
         assert results[0].thread_id == "main"
-        assert "Num Results in context: 1" in log_output
+        assert "Num Results in context: 2" in log_output  # failed hook and model