From 7358dbb788162f8dac885e9edec60b2c31f33091 Mon Sep 17 00:00:00 2001
From: Chenyu Li <chenyu.li@dbtlabs.com>
Date: Fri, 18 Oct 2024 14:28:12 -0700
Subject: [PATCH] preserve previous behavior

---
 core/dbt/task/run.py                          |  4 +-
 core/dbt/task/runnable.py                     |  1 -
 .../adapter/hooks/test_on_run_hooks.py        | 42 +++++++++++++++++--
 3 files changed, 41 insertions(+), 6 deletions(-)

diff --git a/core/dbt/task/run.py b/core/dbt/task/run.py
index d9fa902d265..ade14209747 100644
--- a/core/dbt/task/run.py
+++ b/core/dbt/task/run.py
@@ -768,8 +768,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"):
diff --git a/core/dbt/task/runnable.py b/core/dbt/task/runnable.py
index a37308f81ea..7ad9d87e10d 100644
--- a/core/dbt/task/runnable.py
+++ b/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
             ):
diff --git a/tests/functional/adapter/hooks/test_on_run_hooks.py b/tests/functional/adapter/hooks/test_on_run_hooks.py
index a078af582f9..627d4b0779c 100644
--- a/tests/functional/adapter/hooks/test_on_run_hooks.py
+++ b/tests/functional/adapter/hooks/test_on_run_hooks.py
@@ -162,7 +162,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 {
@@ -192,9 +192,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