fix: [AI-8671] accept dbt Fusion's capitalized freshness statuses in sources v1-v3

The dbt Fusion engine serializes source-freshness `status` with its Rust
variant names -- `"Pass"` / `"Warn"` / `"Error"` -- while dbt-core and the
published `sources/v3.json` schema Fusion itself stamps into the artifact
both use the lowercase forms. Every result row therefore failed both members
of the `results` union, the whole `sources.json` raised a `ValidationError`,
and the ingestion worker silently dropped it. Nine of harvestgroup's ten
production environments have zero source-freshness rows as a result.

This is the residual of the AI-7675 work: PR #106/#108 added the `_missing_`
forward-compat shim to the `run_results` status enums but explicitly left the
freshness enums, and the `sources` parsers entirely, untouched.

- Make `Status1` a `str, Enum` whose `_missing_` case-folds to the canonical
  lowercase member first, then falls back to the same forward-compat
  pseudo-member used by the `run_results` shim for unknown statuses.
- Case-folding rather than adding PascalCase members is deliberate: the
  extractor persists `status.value`, and every dbt-core-backed tenant already
  writes lowercase into the same table.
- Leave the runtime-error-only `Status` enum strict. Fusion has no
  runtime-error variant, and loosening it would let a row missing a required
  field fall silently into the field-less branch instead of erroring.
- Apply to v1-v3 rather than v3 alone, mirroring how #108 had to follow #106
  across the older schemas.

Verified against 432 real production artifacts spanning all nine Fusion
environments and 13 Fusion builds (preview.190 -> .210), run through parse
AND the worker's own `extract_sources`: 432/432 parsed, 10,893 freshness rows
extracted, every capitalized input landing as its lowercase counterpart.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

Author: shreyastelkar

src/vendor/dbt_artifacts_parser/parsers/sources/sources_v1.py

@@ -38,12 +38,33 @@ class SourceFreshnessRuntimeError(BaseParserModel):
     status: Status
 
 
-class Status1(Enum):
+class Status1(str, Enum):
     pass_ = "pass"
     warn = "warn"
     error = "error"
     runtime_error = "runtime error"
 
+    @classmethod
+    def _missing_(cls, value):
+        # The dbt Fusion engine serializes freshness statuses with its Rust
+        # variant names -- "Pass" / "Warn" / "Error" -- while the published
+        # sources schema it stamps into the artifact, and every dbt-core
+        # release, use the lowercase forms. Fold case first so a Fusion
+        # artifact resolves to the canonical lowercase member and `.value`
+        # stays stable for downstream storage.
+        if isinstance(value, str):
+            folded = value.casefold()
+            for member in cls:
+                if member.value.casefold() == folded:
+                    return member
+        # Forward-compatibility: surface any other unknown status as a real
+        # member so downstream `.value` access keeps working instead of failing
+        # validation and silently dropping the entire sources.json.
+        member = str.__new__(cls, value)
+        member._name_ = str(value)
+        member._value_ = value
+        return member
+
 
 class Period(Enum):
     minute = "minute"

src/vendor/dbt_artifacts_parser/parsers/sources/sources_v2.py

@@ -38,12 +38,33 @@ class SourceFreshnessRuntimeError(BaseParserModel):
     status: Status
 
 
-class Status1(Enum):
+class Status1(str, Enum):
     pass_ = "pass"
     warn = "warn"
     error = "error"
     runtime_error = "runtime error"
 
+    @classmethod
+    def _missing_(cls, value):
+        # The dbt Fusion engine serializes freshness statuses with its Rust
+        # variant names -- "Pass" / "Warn" / "Error" -- while the published
+        # sources schema it stamps into the artifact, and every dbt-core
+        # release, use the lowercase forms. Fold case first so a Fusion
+        # artifact resolves to the canonical lowercase member and `.value`
+        # stays stable for downstream storage.
+        if isinstance(value, str):
+            folded = value.casefold()
+            for member in cls:
+                if member.value.casefold() == folded:
+                    return member
+        # Forward-compatibility: surface any other unknown status as a real
+        # member so downstream `.value` access keeps working instead of failing
+        # validation and silently dropping the entire sources.json.
+        member = str.__new__(cls, value)
+        member._name_ = str(value)
+        member._value_ = value
+        return member
+
 
 class Period(Enum):
     minute = "minute"

src/vendor/dbt_artifacts_parser/parsers/sources/sources_v3.py

@@ -38,12 +38,33 @@ class Results(BaseParserModel):
     status: Status
 
 
-class Status1(Enum):
+class Status1(str, Enum):
     pass_ = "pass"
     warn = "warn"
     error = "error"
     runtime_error = "runtime error"
 
+    @classmethod
+    def _missing_(cls, value):
+        # The dbt Fusion engine serializes freshness statuses with its Rust
+        # variant names -- "Pass" / "Warn" / "Error" -- while the published
+        # sources schema it stamps into the artifact, and every dbt-core
+        # release, use the lowercase forms. Fold case first so a Fusion
+        # artifact resolves to the canonical lowercase member and `.value`
+        # stays stable for downstream storage.
+        if isinstance(value, str):
+            folded = value.casefold()
+            for member in cls:
+                if member.value.casefold() == folded:
+                    return member
+        # Forward-compatibility: surface any other unknown status as a real
+        # member so downstream `.value` access keeps working instead of failing
+        # validation and silently dropping the entire sources.json.
+        member = str.__new__(cls, value)
+        member._name_ = str(value)
+        member._value_ = value
+        return member
+
 
 class Period(Enum):
     minute = "minute"

tests/test_vendor/test_sources_freshness_status.py

@@ -0,0 +1,127 @@
+"""Tests for the sources v1-v3 parsers, specifically the resilient freshness `Status1` enum.
+
+Regression coverage for the dbt Fusion engine emitting capitalized freshness
+statuses (``"Pass"`` / ``"Warn"`` / ``"Error"``) in sources.json. Every result row
+failed both members of the ``results`` union, so the ENTIRE sources.json raised a
+``ValidationError`` and was silently dropped during ingestion -- no source freshness
+ever reached Postgres for a Fusion-backed environment.
+
+The statuses must fold to their canonical lowercase members, because the extractor
+persists ``result.status.value`` and the rest of the platform (including every
+dbt-core-backed tenant already in the same table) uses the lowercase vocabulary.
+"""
+import pytest
+
+from vendor.dbt_artifacts_parser.parser import parse_sources
+from vendor.dbt_artifacts_parser.parsers.sources.sources_v1 import SourceFreshnessOutput as OutputV1
+from vendor.dbt_artifacts_parser.parsers.sources.sources_v1 import Status1 as StatusV1
+from vendor.dbt_artifacts_parser.parsers.sources.sources_v2 import SourceFreshnessOutput as OutputV2
+from vendor.dbt_artifacts_parser.parsers.sources.sources_v2 import Status1 as StatusV2
+from vendor.dbt_artifacts_parser.parsers.sources.sources_v3 import Results as RuntimeErrorV3
+from vendor.dbt_artifacts_parser.parsers.sources.sources_v3 import Results1 as OutputV3
+from vendor.dbt_artifacts_parser.parsers.sources.sources_v3 import Status1 as StatusV3
+
+V3_SCHEMA = "https://schemas.getdbt.com/dbt/sources/v3.json"
+
+# (output model, status enum) per schema version -- the enum is identical in all three.
+VERSIONS = [
+    pytest.param(OutputV1, StatusV1, id="v1"),
+    pytest.param(OutputV2, StatusV2, id="v2"),
+    pytest.param(OutputV3, StatusV3, id="v3"),
+]
+
+# Fusion's Rust variant name -> the canonical lowercase status dbt-core emits.
+FUSION_CASINGS = [("Pass", "pass"), ("Warn", "warn"), ("Error", "error")]
+
+
+def _output(status: str, unique_id: str = "source.proj.schema.tbl") -> dict:
+    """A complete freshness result -- the shape Fusion always emits."""
+    return {
+        "unique_id": unique_id,
+        "max_loaded_at": "2026-08-19T08:33:50.855920Z",
+        "snapshotted_at": "2026-08-19T17:00:33.833000Z",
+        "max_loaded_at_time_ago_in_s": 30402.0,
+        "status": status,
+        "criteria": {
+            "warn_after": {"count": 24, "period": "hour"},
+            "error_after": {"count": 48, "period": "hour"},
+        },
+        "adapter_response": {},
+        "timing": [],
+        "thread_id": "Thread-20",
+        "execution_time": 0.0,
+    }
+
+
+def _sources_v3(*statuses: str) -> dict:
+    return {
+        "metadata": {
+            "dbt_schema_version": V3_SCHEMA,
+            "dbt_version": "2.0.0-preview.210",
+            "invocation_id": "test-invocation-123",
+        },
+        "elapsed_time": 1.5,
+        "results": [_output(s, f"source.proj.sch.t{i}") for i, s in enumerate(statuses)],
+    }
+
+
+class TestFusionStatusCasing:
+    """Fusion's capitalized statuses must parse AND normalize to lowercase."""
+
+    @pytest.mark.parametrize(("model", "status_enum"), VERSIONS)
+    @pytest.mark.parametrize(("fusion", "canonical"), FUSION_CASINGS)
+    def test_capitalized_status_folds_to_canonical_member(self, model, status_enum, fusion, canonical):
+        result = model(**_output(fusion))
+        assert result.status is status_enum(canonical)
+        assert result.status.value == canonical
+
+    @pytest.mark.parametrize(("model", "status_enum"), VERSIONS)
+    def test_lowercase_statuses_unchanged(self, model, status_enum):
+        """dbt-core's existing lowercase vocabulary must keep resolving as before."""
+        for status in ("pass", "warn", "error", "runtime error"):
+            assert model(**_output(status)).status.value == status
+
+    @pytest.mark.parametrize(("model", "status_enum"), VERSIONS)
+    def test_unknown_future_status_parses(self, model, status_enum):
+        """Forward-compat: a status dbt has not shipped yet must not drop the file."""
+        assert model(**_output("some_future_status")).status.value == "some_future_status"
+
+
+class TestUnionResolutionIsNotLossy:
+    """A complete freshness row must NEVER resolve to the runtime-error branch.
+
+    ``SourcesV3.results`` is ``list[Union[Results, Results1]]`` and ``Results`` (the
+    runtime-error shape) requires only ``unique_id`` + ``status`` with ``extra="allow"``.
+    If a full row resolved there, every freshness field would be dropped -- turning a
+    loud parse failure into silent data loss, which is strictly worse.
+    """
+
+    @pytest.mark.parametrize(("fusion", "canonical"), FUSION_CASINGS)
+    def test_full_row_resolves_to_output_branch(self, fusion, canonical):
+        parsed = parse_sources(_sources_v3(fusion))
+        (result,) = parsed.results
+        assert isinstance(result, OutputV3)
+        assert result.status.value == canonical
+        assert result.max_loaded_at == "2026-08-19T08:33:50.855920Z"
+        assert result.criteria.error_after.count == 48
+
+    def test_runtime_error_row_still_resolves_to_runtime_error_branch(self):
+        """dbt-core emits a distinct, field-less shape for a failed freshness check."""
+        artifact = _sources_v3()
+        artifact["results"] = [
+            {
+                "unique_id": "source.proj.sch.broken",
+                "error": "Database Error: permission denied",
+                "status": "runtime error",
+            }
+        ]
+        (result,) = parse_sources(artifact).results
+        assert isinstance(result, RuntimeErrorV3)
+        assert result.status.value == "runtime error"
+
+
+class TestFullArtifactParse:
+    def test_fusion_artifact_parses_end_to_end(self):
+        """The whole-file failure this regression is about: mixed Fusion statuses."""
+        parsed = parse_sources(_sources_v3("Pass", "Error", "Warn", "Pass"))
+        assert [r.status.value for r in parsed.results] == ["pass", "error", "warn", "pass"]