feat: add cost_optimization_features to job definitions

Add support for cost_optimization_features field (state_aware_orchestration,
efficient_testing) so dbt Fusion users can configure cost optimization
natively in YAML job definitions.

Closes #196

Author: b-per

docs/changelog.md

@@ -1,6 +1,10 @@
 
 To see the details of all changes, head to the GitHub repo
 
+### 1.16
+
+- Add support for `cost_optimization_features` in job definitions. Valid values are `state_aware_orchestration` and `efficient_testing`. This allows for dbt users on the Fusion engine to configure cost optimization natively in their YAML job definitions.
+
 ### 1.14
 
 - Add applied job IDs to `sync --json` output. The JSON now includes an `applied` section with `job_id` for each operation and an `apply_success` flag. See [JSON output](advanced_config/json_output.md) for details.

example_jobs_file/jobs.yml

@@ -26,6 +26,8 @@ jobs:
       git_provider_webhook: false
       github_webhook: false
       schedule: true
+    cost_optimization_features:
+      - state_aware_orchestration
 
   job2:
     account_id: 43791

src/dbt_jobs_as_code/schemas/job.py

@@ -110,6 +110,12 @@ class JobDefinition(BaseModel):
         json_schema_extra={"enum": ["scheduled", "merge", "ci", "other"]},
         default="scheduled",
     )
+    cost_optimization_features: Optional[List[str]] = Field(
+        default=None,
+        json_schema_extra={
+            "items": {"enum": ["state_aware_orchestration", "efficient_testing"]},
+        },
+    )
     triggers_on_draft_pr: bool = False
     job_completion_trigger_condition: Optional[JobCompletionTriggerCondition] = None
     custom_environment_variables: List[CustomEnvironmentVariable] = Field(
@@ -270,9 +276,7 @@ class JobMissingFields(JobDefinition):
     # when adding fields we also need to update the test for pytest
 
     # TODO: Add to JobDefinition model when the feature is out
-    integration_id: Optional[int] = None
     force_node_selection: Optional[bool] = True
-    cost_optimization_features: list[str]
 
     # Unneeded read-only fields
     raw_dbt_version: Optional[str] = None

src/dbt_jobs_as_code/schemas/load_job_schema.json

@@ -329,6 +329,27 @@
           "title": "Job Type",
           "type": "string"
         },
+        "cost_optimization_features": {
+          "anyOf": [
+            {
+              "items": {
+                "type": "string"
+              },
+              "type": "array"
+            },
+            {
+              "type": "null"
+            }
+          ],
+          "default": null,
+          "items": {
+            "enum": [
+              "state_aware_orchestration",
+              "efficient_testing"
+            ]
+          },
+          "title": "Cost Optimization Features"
+        },
         "triggers_on_draft_pr": {
           "default": false,
           "title": "Triggers On Draft Pr",

tests/exporter/test_export.py

@@ -70,6 +70,7 @@ def test_export_jobs_yml(capsys):
     run_compare_changes: false
     compare_changes_flags: --select state:modified
     job_type: scheduled
+    cost_optimization_features:
     triggers_on_draft_pr: false
     job_completion_trigger_condition:
     custom_environment_variables: []

tests/loader/conftest.py

@@ -206,6 +206,7 @@ def expected_config_dict():
             "job1": {
                 "account_id": 43791,
                 "compare_changes_flags": "--select state:modified",
+                "cost_optimization_features": None,
                 "custom_environment_variables": [],
                 "dbt_version": None,
                 "deferring_environment_id": None,
@@ -248,6 +249,7 @@ def expected_config_dict():
             "job2": {
                 "account_id": 43791,
                 "compare_changes_flags": "--select state:modified",
+                "cost_optimization_features": None,
                 "custom_environment_variables": [
                     {
                         "display_value": None,

tests/schemas/test_job.py

@@ -239,3 +239,92 @@ def test_json_schema_schedule_accepted_for_all_types(self, json_schema, job_type
         """Providing schedule is always valid regardless of job_type."""
         instance = self._config_instance(job_type=job_type, include_schedule=True)
         validate(instance=instance, schema=json_schema)
+
+
+class TestCostOptimizationFeatures:
+    """Tests for cost_optimization_features field on JobDefinition."""
+
+    def test_pydantic_defaults_to_none(self):
+        job = JobDefinition(**{**BASE_JOB_DATA, "schedule": {"cron": "0 0 * * *"}})
+        assert job.cost_optimization_features is None
+
+    def test_pydantic_accepts_valid_features(self):
+        job = JobDefinition(
+            **{
+                **BASE_JOB_DATA,
+                "schedule": {"cron": "0 0 * * *"},
+                "cost_optimization_features": ["state_aware_orchestration"],
+            }
+        )
+        assert job.cost_optimization_features == ["state_aware_orchestration"]
+
+    def test_pydantic_accepts_multiple_features(self):
+        job = JobDefinition(
+            **{
+                **BASE_JOB_DATA,
+                "schedule": {"cron": "0 0 * * *"},
+                "cost_optimization_features": [
+                    "state_aware_orchestration",
+                    "efficient_testing",
+                ],
+            }
+        )
+        assert job.cost_optimization_features == [
+            "state_aware_orchestration",
+            "efficient_testing",
+        ]
+
+    def test_pydantic_accepts_empty_list(self):
+        job = JobDefinition(
+            **{
+                **BASE_JOB_DATA,
+                "schedule": {"cron": "0 0 * * *"},
+                "cost_optimization_features": [],
+            }
+        )
+        assert job.cost_optimization_features == []
+
+    def test_payload_includes_cost_optimization_features(self):
+        job = JobDefinition(
+            **{
+                **BASE_JOB_DATA,
+                "schedule": {"cron": "0 0 * * *"},
+                "cost_optimization_features": ["state_aware_orchestration"],
+            }
+        )
+        payload = json.loads(job.to_payload())
+        assert payload["cost_optimization_features"] == ["state_aware_orchestration"]
+
+    def test_payload_includes_null_when_not_set(self):
+        job = JobDefinition(**{**BASE_JOB_DATA, "schedule": {"cron": "0 0 * * *"}})
+        payload = json.loads(job.to_payload())
+        assert payload["cost_optimization_features"] is None
+
+    @pytest.fixture
+    def json_schema(self):
+        return json.loads(generate_config_schema())
+
+    def test_json_schema_accepts_valid_features(self, json_schema):
+        instance = {
+            "jobs": {
+                "test_job": {
+                    **BASE_JOB_DATA,
+                    "schedule": {"cron": "0 0 * * *"},
+                    "cost_optimization_features": ["state_aware_orchestration"],
+                }
+            }
+        }
+        validate(instance=instance, schema=json_schema)
+
+    def test_json_schema_rejects_invalid_feature(self, json_schema):
+        instance = {
+            "jobs": {
+                "test_job": {
+                    **BASE_JOB_DATA,
+                    "schedule": {"cron": "0 0 * * *"},
+                    "cost_optimization_features": ["not_a_real_feature"],
+                }
+            }
+        }
+        with pytest.raises(JsonSchemaValidationError):
+            validate(instance=instance, schema=json_schema)