Support parsing of fusion logs for warnings in get_job_run_error (#737)

## Summary
<!-- Provide a brief description of the changes in this PR -->
With fusion sf GA right around the corner, Benoit said tis time to add
support for fusion log parsing of warnings. This should be stable now!
This is specific to the Admin API tool, `get_job_run_error` which
optionally returns warnings as well.

## What Changed
<!-- Describe the changes made in this PR -->
- Add `WarningFetcher._extract_fusion_log_warnings` and
`WarningFetcher.is_fusion_logs` check
- Parses fusion-specific warn format in logs

## Why
<!-- Explain the motivation for these changes -->

## Related Issues
<!-- Link any related issues using #issue_number -->
Closes #447 


## Checklist
- [x] I have performed a self-review of my code
- [x] I have made corresponding changes to the documentation (in
https://github.com/dbt-labs/docs.getdbt.com) if required -- Mention it
here
- [x] I have added tests that prove my fix is effective or that my
feature works
- [x] New and existing unit tests pass locally with my changes

## Validation

1. Tested in MCP inspector (trust me bro)
2. Tested with Claude Cpde

<img width="863" height="478" alt="Screenshot 2026-04-21 at 10 39 37 PM"
src="https://github.com/user-attachments/assets/aef6c6cf-3309-46dd-8a3b-57619e657c17"
/>
<img width="862" height="451" alt="Screenshot 2026-04-21 at 10 40 04 PM"
src="https://github.com/user-attachments/assets/b7f795ba-fc6b-46e7-a938-df98480ba190"
/>

Author: jairus-m

.changes/unreleased/Enhancement or New Feature-20260421-221040.yaml

@@ -0,0 +1,3 @@
+kind: Enhancement or New Feature
+body: Support parsing of fusion logs for warnings
+time: 2026-04-21T22:10:40.238804-07:00

src/dbt_mcp/dbt_admin/run_artifacts/parser.py

@@ -26,6 +26,16 @@
 
 logger = logging.getLogger(__name__)
 
+# Matches ANSI color/style escape sequences (ie \x1b[33;1m) so they can be
+# stripped from log output before pattern matching on plain text
+_ANSI_ESCAPE = re.compile(r"\x1b\[[0-9;]*m")
+# dbt<####> is the Fusion event code (ie dbt1088) to catch warnings. Warnings without this code
+# will not be matched (can upate if Fusion changes its log format)
+_FUSION_WARN = re.compile(
+    r"^\d{2}:\d{2}:\d{2}\s+WARN\s+Warn\s+dbt\d+:\s+.+$",
+    re.MULTILINE,
+)
+
 
 class JobRunFetcher:
     """Base class for fetching and parsing dbt Cloud job run artifacts.
@@ -537,20 +547,42 @@ async def _check_source_freshness_warnings(
             result_status="warn",
         )
 
+    def _is_fusion_logs(self, clean_logs: str) -> bool:
+        """Detect whether logs are from dbt Fusion (vs dbt Core/Platform)."""
+        return "Fusion version:" in clean_logs
+
+    def _extract_fusion_log_warnings(self, clean_logs: str) -> list[OutputResultSchema]:
+        """Extract warnings from dbt Fusion logs.
+
+        Fusion warnings are self-contained single lines:
+            HH:MM:SS      WARN         Warn dbtXXXX: <message>
+        """
+        return [
+            OutputResultSchema(
+                unique_id=None,
+                relation_name=None,
+                message=match.group(0).strip(),
+                status="warn",
+            )
+            for match in _FUSION_WARN.finditer(clean_logs)
+        ]
+
     def _extract_log_warnings(self, step: RunStepSchema) -> list[OutputResultSchema]:
-        """Extract warnings from step logs, matching on [WARNING] log entries."""
+        """Extract warnings from step logs (supports dbt Core and dbt Fusion formats)."""
         if not step.logs:
             return []
 
-        if "[WARNING]" not in step.logs:
-            return []
+        # Remove ANSI color codes to focus on text
+        clean_logs = _ANSI_ESCAPE.sub("", step.logs)
 
-        # TODO: Fusion logs differ from core (currently core only)
-        # Need to explore and implement a solution for this
+        # dbt Fusion: WARN with single line
+        if self._is_fusion_logs(clean_logs):
+            return self._extract_fusion_log_warnings(clean_logs)
+
+        # dbt Core: [WARNING] inline marker with possible continuation lines
+        if "[WARNING]" not in clean_logs:
+            return []
 
-        # Remove ANSI color codes to focus on text
-        ansi_escape = re.compile(r"\x1b\[[0-9;]*m")
-        clean_logs = ansi_escape.sub("", step.logs)
         lines = clean_logs.split("\n")
         warnings = []
         warning_pattern = r"\[WARNING\]"  # Standard warning log pattern in dbt Platform

tests/unit/dbt_admin/run_artifacts/test_warning_fetcher.py

@@ -129,6 +129,37 @@
                 "log_warnings": 1,
             },
         ),
+        # Fusion log warnings extracted from WARN-prefixed log entries
+        (
+            {
+                "id": 500,
+                "status": 10,
+                "is_cancelled": False,
+                "finished_at": "2024-01-01T12:00:00Z",
+                "run_steps": [
+                    {
+                        "index": 1,
+                        "name": "Invoke dbt with `dbt run`",
+                        "status": 10,
+                        "finished_at": "2024-01-01T12:00:00Z",
+                        "logs": (
+                            "04:00:14      INFO \x1b[1m     Running\x1b[0m Fusion version: 2.0.0-preview.173\n"
+                            "04:00:14      WARN \x1b[33;1m        Warn\x1b[0m \x1b[1mdbt1088\x1b[0m: Updated version available for zendesk@1.4.1: 1.5.1\n"
+                            "04:00:15      WARN \x1b[33;1m        Warn\x1b[0m \x1b[1mdbt1088\x1b[0m: Updated version available for tiktok_ads@1.1.0: 1.2.0\n"
+                            "04:00:16      INFO \x1b[1;32m   Installed\x1b[0m 2 packages"
+                        ),
+                    }
+                ],
+            },
+            [None],  # No run_results.json available
+            True,
+            {
+                "total_warnings": 2,
+                "test_warnings": 0,
+                "freshness_warnings": 0,
+                "log_warnings": 2,
+            },
+        ),
     ],
 )
 async def test_warning_scenarios(
@@ -184,3 +215,67 @@ async def mock_get_artifact(account_id, run_id, artifact_path, step=None):  # no
 
     assert result["has_warnings"] == expected_has_warnings
     assert result["summary"] == expected_counts
+
+
+@pytest.mark.parametrize(
+    "clean_logs,expected",
+    [
+        # Fusion banner present → detected as Fusion
+        ("04:00:14      INFO      Running Fusion version: 2.0.0-preview.173\n", True),
+        # Core logs — no Fusion banner
+        ("10:00:00 [WARNING] Deprecated function usage detected\n", False),
+        # Contains "Fusion" in a model name but not the banner — must not false-positive
+        ("10:00:00 INFO model fusion_events completed successfully\n", False),
+    ],
+)
+def test_is_fusion_logs(mock_client, admin_config, clean_logs, expected):
+    fetcher = WarningFetcher(
+        run_id=1,
+        run_details={
+            "id": 1,
+            "status": 10,
+            "is_cancelled": False,
+            "finished_at": "2024-01-01T00:00:00Z",
+            "run_steps": [],
+        },
+        client=mock_client,
+        admin_api_config=admin_config,
+    )
+    assert fetcher._is_fusion_logs(clean_logs) == expected
+
+
+@pytest.mark.parametrize(
+    "clean_logs,expected_messages",
+    [
+        # No WARN lines → empty result
+        (
+            "04:00:14      INFO      Running Fusion version: 2.0.0-preview.173\n"
+            "04:00:16      INFO      Installed 2 packages\n",
+            [],
+        ),
+        # Single WARN line → one result with the full line as the message
+        (
+            "04:00:14      WARN      Warn dbt1088: Updated version available for zendesk@1.4.1: 1.5.1\n",
+            [
+                "04:00:14      WARN      Warn dbt1088: Updated version available for zendesk@1.4.1: 1.5.1"
+            ],
+        ),
+    ],
+)
+def test_extract_fusion_log_warnings(
+    mock_client, admin_config, clean_logs, expected_messages
+):
+    fetcher = WarningFetcher(
+        run_id=1,
+        run_details={
+            "id": 1,
+            "status": 10,
+            "is_cancelled": False,
+            "finished_at": "2024-01-01T00:00:00Z",
+            "run_steps": [],
+        },
+        client=mock_client,
+        admin_api_config=admin_config,
+    )
+    results = fetcher._extract_fusion_log_warnings(clean_logs)
+    assert [r.message for r in results] == expected_messages