Add Dagster observe-only mode for dbt Platform jobs (#121)

* feat: add dbt platform observe-only mode

* fix: type dbt observe-only asset definitions

Author: C00ldudeNoonan

.env.example

@@ -117,6 +117,9 @@ DBT_PLATFORM_TOKEN=your-dbt-platform-service-token
 DBT_PLATFORM_ACCESS_URL=https://cloud.getdbt.com
 DBT_PLATFORM_ADHOC_JOB_NAME=economic-data-dagster-adhoc
 DBT_PLATFORM_POLL_INTERVAL_SECONDS=60
+# Set true in production when dbt Platform owns scheduled builds and Dagster
+# should only observe completed dbt Platform runs.
+DBT_PLATFORM_OBSERVE_ONLY=false
 # DBT_PROJECT_DIR=/path/to/dbt_project  # Optional, auto-detected
 
 # =============================================================================

docs/dbt_project/dbt-platform-fusion-onboarding.md

@@ -150,6 +150,7 @@ execution modes:
 | --- | --- | --- |
 | Local CLI | `local` | Uses `DbtCliResource` and the bundled `dbt_project/`. |
 | dbt Platform | `dbt_platform` | Uses `DbtCloudWorkspace` to load dbt Platform artifacts and launch ad hoc dbt Platform runs. |
+| dbt Platform observe-only | `dbt_platform` + `DBT_PLATFORM_OBSERVE_ONLY=true` | Loads dbt Platform asset specs and polls completed dbt Platform jobs, but does not register Dagster dbt jobs or launch ad hoc dbt Platform runs. |
 
 Configure these Dagster environment variables after the dbt Platform project and
 environment exist:
@@ -159,11 +160,34 @@ environment exist:
 | `DBT_ORCHESTRATION_MODE` | Set to `dbt_platform` to switch Dagster from local dbt CLI execution to dbt Platform. |
 | `DBT_PLATFORM_ACCOUNT_ID` | dbt Platform account id. |
 | `DBT_PLATFORM_PROJECT_ID` | dbt Platform project id. |
-| `DBT_PLATFORM_ENVIRONMENT_ID` | dbt Platform environment id Dagster should use for ad hoc runs. |
+| `DBT_PLATFORM_ENVIRONMENT_ID` | dbt Platform environment id Dagster should load and observe. Use the production environment when dbt Platform owns scheduled production builds and freshness jobs. |
 | `DBT_PLATFORM_TOKEN` | dbt Platform service token. |
 | `DBT_PLATFORM_ACCESS_URL` | dbt Platform URL, default `https://cloud.getdbt.com`. |
 | `DBT_PLATFORM_ADHOC_JOB_NAME` | Optional ad hoc job name Dagster creates/uses in dbt Platform. |
 | `DBT_PLATFORM_POLL_INTERVAL_SECONDS` | Optional polling interval for external dbt Platform run observations. |
+| `DBT_PLATFORM_OBSERVE_ONLY` | Optional strict production mode. Set to `true` when Dagster should observe dbt Platform jobs without exposing Dagster dbt asset jobs. |
+
+## Dagster Observe-Only Mode
+
+Use observe-only mode for production once dbt Platform owns the scheduled dbt
+jobs:
+
+```bash
+DBT_ORCHESTRATION_MODE=dbt_platform
+DBT_PLATFORM_ENVIRONMENT_ID=<production-environment-id>
+DBT_PLATFORM_OBSERVE_ONLY=true
+```
+
+In this mode Dagster loads dbt model asset specs from the configured dbt
+Platform environment and registers the dbt Cloud polling sensor. The sensor
+polls completed dbt Platform runs for that project and environment, skips
+Dagster-created ad hoc jobs, reads each run's `run_results.json`, and records
+model-level Dagster asset materializations. Dagster does not register the
+`dbt_*_models_job` jobs in this mode, so scheduled `dbt build` and
+`dbt source freshness` runs remain owned by dbt Platform.
+
+Keep observe-only disabled in local development or in a dedicated CI/ad hoc
+environment if you still want Dagster to trigger dbt Platform runs.
 
 Cutover path:
 
@@ -172,7 +196,9 @@ Cutover path:
 3. Set `DBT_ORCHESTRATION_MODE=dbt_platform` and configure the `DBT_PLATFORM_*` variables in Dagster.
 4. Validate Dagster code location load; Dagster should fetch the manifest from dbt Platform.
 5. Materialize a small dbt asset selection from Dagster to confirm ad hoc dbt Platform runs work.
-6. Remove local dbt project bundling and `dbt deps` fallback after the API path is stable.
+6. For production, set `DBT_PLATFORM_ENVIRONMENT_ID` to the production environment and `DBT_PLATFORM_OBSERVE_ONLY=true`.
+7. Confirm the `dbt_cloud_<account>_<project>_<environment>__run_status_sensor` sensor records asset materializations from the dbt Platform production build job.
+8. Remove local dbt project bundling and `dbt deps` fallback after the API path is stable.
 
 ## Remaining Manual Checklist
 
@@ -186,5 +212,7 @@ Cutover path:
 - [ ] Configure PR, production build, and source freshness jobs.
 - [ ] Add dbt Platform PR check and `Dagster CI / Test Dagster` to GitHub branch protection.
 - [ ] Add dbt Platform GitHub repository secrets/variables for Dagster CI.
-- [ ] Decide final Dagster orchestration cutover date.
+- [ ] Set `DBT_PLATFORM_OBSERVE_ONLY=true` in production Dagster after dbt Platform schedules are live.
+- [ ] Record the production dbt Platform environment id, production build job id, and source freshness job id in deployment notes.
 - [ ] Validate Dagster can load dbt Platform asset specs and trigger an ad hoc run.
+- [ ] Validate Dagster observes a completed dbt Platform production build and records model-level asset materializations.

macro_agents/src/macro_agents/defs/transformation/__init__.py

@@ -1,7 +1,10 @@
+from typing import cast
+
 import dagster as dg
 
 from macro_agents.defs.transformation.dbt import (
     dbt_cloud_polling_sensor,
+    dbt_platform_observe_only,
     dbt_resource,
     full_dbt_assets,
 )
@@ -11,6 +14,8 @@
     financial_conditions_index,
 )
 
+DagsterAssetDefinition = dg.AssetsDefinition | dg.AssetSpec | dg.SourceAsset
+
 
 dbt_models_job = dg.define_asset_job(
     name="dbt_models_job",
@@ -108,11 +113,16 @@
 if dbt_cloud_polling_sensor is not None:
     transformation_sensors.append(dbt_cloud_polling_sensor)
 
+dbt_asset_definitions: list[DagsterAssetDefinition] = (
+    cast(list[DagsterAssetDefinition], full_dbt_assets)
+    if isinstance(full_dbt_assets, list)
+    else [cast(DagsterAssetDefinition, full_dbt_assets)]
+)
 
-defs = dg.Definitions(
-    assets=[full_dbt_assets, financial_conditions_index, fci_weights_config],
-    asset_checks=transformation_checks,
-    jobs=[
+dbt_jobs = (
+    []
+    if dbt_platform_observe_only
+    else [
         dbt_models_job,
         dbt_staging_models_job,
         dbt_staging_telemetry_models_job,
@@ -124,7 +134,18 @@
         dbt_agents_preprocess_models_job,
         dbt_data_quality_models_job,
         dbt_backtesting_models_job,
+    ]
+)
+
+
+defs = dg.Definitions(
+    assets=[
+        *dbt_asset_definitions,
+        financial_conditions_index,
+        fci_weights_config,
     ],
+    asset_checks=transformation_checks,
+    jobs=dbt_jobs,
     resources={"dbt": dbt_resource},
     sensors=transformation_sensors,
 )

macro_agents/src/macro_agents/defs/transformation/dbt.py

@@ -65,6 +65,11 @@ def _first_env(*names: str) -> str | None:
     return None
 
 
+def _truthy_env(*names: str) -> bool:
+    value = _first_env(*names)
+    return value is not None and value.lower() in {"1", "true", "t", "yes", "y", "on"}
+
+
 def _required_int_env(*names: str) -> int:
     value = _first_env(*names)
     if value is None:
@@ -86,7 +91,18 @@ def _required_env_var(*names: str) -> dg.EnvVar:
         raise RuntimeError(f"Set {joined_names} to enable dbt Platform orchestration.")
 
 
-if orchestration_mode in {"dbt_platform", "dbt_cloud", "platform", "cloud"}:
+dbt_platform_mode = orchestration_mode in {
+    "dbt_platform",
+    "dbt_cloud",
+    "platform",
+    "cloud",
+}
+dbt_platform_observe_only = dbt_platform_mode and _truthy_env(
+    "DBT_PLATFORM_OBSERVE_ONLY", "DBT_CLOUD_OBSERVE_ONLY"
+)
+
+
+if dbt_platform_mode:
     dbt_cloud_credentials = DbtCloudCredentials(
         account_id=_required_int_env("DBT_PLATFORM_ACCOUNT_ID", "DBT_CLOUD_ACCOUNT_ID"),
         token=_required_env_var("DBT_PLATFORM_TOKEN", "DBT_CLOUD_API_TOKEN"),
@@ -105,21 +121,34 @@ def _required_env_var(*names: str) -> dg.EnvVar:
         ),
     )
 
-    @dbt_cloud_assets(
-        workspace=dbt_resource,
-        exclude=DBT_NASDAQ_EXCLUDE,
-        dagster_dbt_translator=CustomizedDagsterDbtTranslator(),
-    )
-    def full_dbt_assets(context: dg.AssetExecutionContext, dbt: DbtCloudWorkspace):
-        yield from dbt.cli(
-            ["build"],
-            dagster_dbt_translator=CustomizedDagsterDbtTranslator(),
-            context=context,
-        ).wait()
+    dbt_platform_translator = CustomizedDagsterDbtTranslator()
+
+    if dbt_platform_observe_only:
+        full_dbt_assets = list(
+            dbt_resource.load_asset_specs(
+                select="fqn:*",
+                exclude=DBT_NASDAQ_EXCLUDE,
+                selector="",
+                dagster_dbt_translator=dbt_platform_translator,
+            )
+        )
+    else:
+
+        @dbt_cloud_assets(
+            workspace=dbt_resource,
+            exclude=DBT_NASDAQ_EXCLUDE,
+            dagster_dbt_translator=dbt_platform_translator,
+        )
+        def full_dbt_assets(context: dg.AssetExecutionContext, dbt: DbtCloudWorkspace):
+            yield from dbt.cli(
+                ["build"],
+                dagster_dbt_translator=dbt_platform_translator,
+                context=context,
+            ).wait()
 
     dbt_cloud_polling_sensor = build_dbt_cloud_polling_sensor(
         workspace=dbt_resource,
-        dagster_dbt_translator=CustomizedDagsterDbtTranslator(),
+        dagster_dbt_translator=dbt_platform_translator,
         minimum_interval_seconds=int(
             os.getenv("DBT_PLATFORM_POLL_INTERVAL_SECONDS", "60")
         ),

macro_agents/tests/test_dbt_platform_observe_only.py

@@ -0,0 +1,39 @@
+import importlib
+import sys
+
+import dagster as dg
+from dagster_dbt import DbtCloudWorkspace
+
+
+def test_dbt_platform_observe_only_registers_specs_without_dbt_jobs(monkeypatch):
+    monkeypatch.setenv("DBT_ORCHESTRATION_MODE", "dbt_platform")
+    monkeypatch.setenv("DBT_PLATFORM_OBSERVE_ONLY", "true")
+    monkeypatch.setenv("DBT_PLATFORM_ACCOUNT_ID", "1")
+    monkeypatch.setenv("DBT_PLATFORM_PROJECT_ID", "2")
+    monkeypatch.setenv("DBT_PLATFORM_ENVIRONMENT_ID", "3")
+    monkeypatch.setenv("DBT_PLATFORM_TOKEN", "test-token")
+
+    def fake_load_asset_specs(self, **kwargs):
+        return [
+            dg.AssetSpec(
+                key=dg.AssetKey("observed_dbt_model"),
+                group_name="staging",
+                kinds={"dbtcloud"},
+            )
+        ]
+
+    monkeypatch.setattr(DbtCloudWorkspace, "load_asset_specs", fake_load_asset_specs)
+
+    for module_name in [
+        "macro_agents.defs.transformation",
+        "macro_agents.defs.transformation.dbt",
+    ]:
+        sys.modules.pop(module_name, None)
+
+    transformation = importlib.import_module("macro_agents.defs.transformation")
+
+    assert transformation.dbt_jobs == []
+    assert transformation.dbt_asset_definitions[0].key == dg.AssetKey(
+        "observed_dbt_model"
+    )
+    assert transformation.dbt_cloud_polling_sensor is not None