fix: store source freshness age for dbt v2 results

integration_tests/dbt_project/macros/test_get_freshness_result_age.sql

@@ -0,0 +1,4 @@
+{% macro test_get_freshness_result_age(result_dict) %}
+    {% set result = elementary.get_freshness_result_age(result_dict) %}
+    {{ return(result) }}
+{% endmacro %}

integration_tests/tests/test_dbt_artifacts/test_get_freshness_result_age.py

@@ -0,0 +1,37 @@
+import json
+
+import pytest
+from dbt_project import DbtProject
+
+
+@pytest.mark.parametrize(
+    "result_dict,expected_output",
+    [
+        # dbt-core
+        ({"age": 120.5}, 120.5),
+        ({"age": 0}, 0),
+        # dbt-fusion: no `age`, the value is in max_loaded_at_time_ago_in_s
+        ({"max_loaded_at_time_ago_in_s": 88.0}, 88.0),
+        ({"age": None, "max_loaded_at_time_ago_in_s": 88.0}, 88.0),
+        ({}, None),
+    ],
+)
+def test_get_freshness_result_age(
+    dbt_project: DbtProject, result_dict, expected_output
+):
+    """get_freshness_result_age reads `age` (dbt-core) and falls back to
+    `max_loaded_at_time_ago_in_s` (dbt-fusion)."""
+    result = dbt_project.dbt_runner.run_operation(
+        "elementary_tests.test_get_freshness_result_age",
+        macro_args={"result_dict": result_dict},
+    )
+    # When the macro returns None, log_macro_results doesn't log anything,
+    # so run_operation returns an empty list
+    if not result:
+        actual_output = None
+    else:
+        actual_output = json.loads(result[0])
+    assert actual_output == expected_output, (
+        f"get_freshness_result_age({result_dict!r}) returned {actual_output!r}, "
+        f"expected {expected_output!r}"
+    )

macros/edr/dbt_artifacts/upload_source_freshness.sql

@@ -41,7 +41,9 @@
             "status": result_dict.status,
             "max_loaded_at": result_dict.max_loaded_at,
             "snapshotted_at": result_dict.snapshotted_at,
-            "max_loaded_at_time_ago_in_s": result_dict.age,
+            "max_loaded_at_time_ago_in_s": elementary.get_freshness_result_age(
+                result_dict
+            ),
             "criteria": result_dict.node.get("freshness", {}),
             "adapter_response": result_dict.adapter_response,
             "timing": result_dict.timing,
@@ -51,6 +53,18 @@
     ) %}
 {% endmacro %}
 
+{#
+    dbt-core exposes the time since the last load as `age`; dbt-fusion has no
+    `age` and exposes it as `max_loaded_at_time_ago_in_s` (the sources.json name).
+#}
+{% macro get_freshness_result_age(result_dict) %}
+    {% set age = result_dict.get("age") %}
+    {% if age is none %}
+        {% set age = result_dict.get("max_loaded_at_time_ago_in_s") %}
+    {% endif %}
+    {% do return(age) %}
+{% endmacro %}
+
 {% macro flatten_source_freshness(node_dict) %}
     {% set compile_timing = {} %}
     {% set execute_timing = {} %}