diff --git a/products/warehouse_sources/backend/temporal/data_imports/pipelines/pipeline_v3/duckgres/enablement.py b/products/warehouse_sources/backend/temporal/data_imports/pipelines/pipeline_v3/duckgres/enablement.py
index ddec8f161f37..fa733fe1ad9a 100644
--- a/products/warehouse_sources/backend/temporal/data_imports/pipelines/pipeline_v3/duckgres/enablement.py
+++ b/products/warehouse_sources/backend/temporal/data_imports/pipelines/pipeline_v3/duckgres/enablement.py
@@ -86,10 +86,14 @@ def duckgres_sink_enablement() -> SinkEnablement | None:
             "id", "uuid", "organization_id"
         )
     }
+    # Filtered by the app DB's own org ids, never by the control plane's row.organization_id:
+    # that value is an external, unvalidated string (the CP has test/dev rows keyed by
+    # human-readable slugs, not UUIDs), and passing it straight into a UUID FK lookup
+    # raises ValidationError before the org_id match-up below ever runs.
     budgets = {
         str(org_id): sink_max_concurrency
         for org_id, sink_max_concurrency in DuckgresServer.objects.filter(
-            organization_id__in={row.organization_id for row in rows}
+            organization_id__in={org_id for _, org_id in team_info.values()}
         ).values_list("organization_id", "sink_max_concurrency")
     }
 
diff --git a/products/warehouse_sources/backend/temporal/data_imports/pipelines/pipeline_v3/duckgres/test_enablement.py b/products/warehouse_sources/backend/temporal/data_imports/pipelines/pipeline_v3/duckgres/test_enablement.py
index eac84be3128c..c9fad2a14267 100644
--- a/products/warehouse_sources/backend/temporal/data_imports/pipelines/pipeline_v3/duckgres/test_enablement.py
+++ b/products/warehouse_sources/backend/temporal/data_imports/pipelines/pipeline_v3/duckgres/test_enablement.py
@@ -110,6 +110,32 @@ def test_duckgres_sink_enablement_uses_memberships_and_carries_org_budgets(
     assert mock_feature_enabled.call_count == 2
 
 
+@pytest.mark.django_db
+@patch.object(enablement, "is_dev_mode", return_value=False)
+@patch.object(enablement.posthoganalytics, "feature_enabled")
+def test_duckgres_sink_enablement_ignores_non_uuid_control_plane_org_ids(
+    mock_feature_enabled: MagicMock, _mock_dev: MagicMock
+) -> None:
+    """The control plane can report extra rows for a team keyed by a non-UUID
+    organization_id (e.g. dev/test rows using human-readable slugs instead of a real
+    org id). Those must not crash the whole refresh via a Django UUID lookup — they
+    should just fail the org match-up and be skipped, like any other mismatched row."""
+    org = Organization.objects.create(name="Org")
+    team = Team.objects.create(organization=org)
+    DuckgresServer.objects.create(organization=org, host="h", username="root", password="x")
+    mock_feature_enabled.return_value = True
+
+    mismatched_row = _cp_row(team)
+    mismatched_row["org_id"] = "not-a-uuid-slug"
+
+    with _patch_all_rows([_cp_row(team), mismatched_row]):
+        result = enablement.duckgres_sink_enablement()
+
+    assert result is not None
+    assert result.team_ids == [team.id]
+    mock_feature_enabled.assert_called_once()
+
+
 @pytest.mark.django_db
 @patch.object(enablement, "is_dev_mode", return_value=False)
 @patch.object(enablement.posthoganalytics, "feature_enabled")