feat: add job1_prod_integration job; remove seed_sources from job1_prod task graph

job1_prod now matches staging structure: health_check → extracts → transforms.
seed_sources moves to the new job1_prod_integration job, which mirrors the
staging integration test pattern: seed_sources → (run + run_sdp in parallel),
no validate task (prod data is synthetic/incremental, not deterministic fixtures).

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: andre-salvati

CLAUDE.md

@@ -131,7 +131,7 @@ On every push: install deps → unit tests → bundle validate → deploy to sta
 - `databricks.yml` prod target has `mode: production` → DABs refuses to deploy if deployer != run-as identity (the SP). A developer's local `make deploy env=prod` will fail by design.
 - CI deploys to prod only when `github.ref == 'refs/heads/main'`.
 - `run_as` and `permissions` on every staging/prod job are pinned to the service principal's `application_id` (numeric), wired by `_get_service_principal_id` in `sdk_generate_template_job.py`.
-- Prod-only features in `_build_job`: cron schedule, `JobEmailNotifications`, a `health_check` task running before any extract, and a `JobsHealthRule` on `RUN_DURATION_SECONDS > DURATION_WARNING_SECONDS` (30 min) so the `on_duration_warning_threshold_exceeded` email actually has an event to fire on.
+- Prod-only features in `_build_job`: cron schedule, `JobEmailNotifications`, a `health_check` task running before any extract, and a `JobsHealthRule` on `RUN_DURATION_SECONDS > DURATION_WARNING_SECONDS` (30 min) so the `on_duration_warning_threshold_exceeded` email actually has an event to fire on. `seed_sources` is **not** in `_build_job` — it lives in `_build_job_prod_integration` (the dedicated prod integration job: seed → run + run_sdp, no validate).
 - The wheel filename in `JobEnvironment.dependencies` is pinned to `_project_version()` (reads `pyproject.toml`) so a forgotten rebuild can't silently deploy an old wheel.
 - Every job sets `max_concurrent_runs=1` + `queue.enabled=true`: late runs queue instead of getting silently skipped. Retries (staging/prod only) back off `MIN_RETRY_INTERVAL_MS` (60s). Per-task `timeout_seconds` (constants near the top of `sdk_generate_template_job.py`) prevent one hung task from eating the whole job budget. `notification_settings.no_alert_for_canceled_runs / _skipped_runs` keeps deliberate cancellations off the on-call pager.
 

scripts/sdk_generate_template_job.py

@@ -195,22 +195,7 @@ def _build_job(environment: str, sp_id: str | None) -> dict:
             )
         )
 
-    # seed_sources runs only in prod: staging/dev use the integration test `setup` task
-    # to seed external_source with controlled data, so seed_sources would add noise there.
-    # In prod, seed_sources runs after health_check and before the extract tasks.
-    if environment == "prod":
-        tasks.append(
-            Task(
-                task_key="seed_sources",
-                **_retry_kwargs(retries),
-                timeout_seconds=TIMEOUT_EXTRACT_S,
-                environment_key="default",
-                depends_on=[TaskDependency(task_key="health_check")],
-                python_wheel_task=_wheel_task(),
-            )
-        )
-
-    extract_deps: list[TaskDependency] = [TaskDependency(task_key="seed_sources")] if environment == "prod" else []
+    extract_deps: list[TaskDependency] = [TaskDependency(task_key="health_check")] if environment == "prod" else []
 
     tasks.extend(
         [
@@ -386,6 +371,63 @@ def _build_job_integration_test(environment: str, sp_id: str | None) -> dict:
     return d
 
 
+def _build_job_prod_integration(sp_id: str | None) -> dict:
+    """Prod integration job: seed_sources → run (job1 batch) + run_sdp (pipeline) in parallel.
+
+    Mirrors the staging integration test structure but uses seed_sources instead of the
+    test-only setup task, and omits validate (prod data is synthetic/incremental, not
+    the deterministic fixtures that integration_validate checks).
+    """
+    tasks = [
+        Task(
+            task_key="seed_sources",
+            **_retry_kwargs(2),
+            timeout_seconds=TIMEOUT_EXTRACT_S,
+            environment_key="default",
+            python_wheel_task=_wheel_task(),
+        ),
+        Task(
+            task_key="run",
+            depends_on=[TaskDependency(task_key="seed_sources")],
+            run_job_task=RunJobTask(job_id=f"${{{f'resources.jobs.{JOB_NAME}.id'}}}"),
+        ),
+        Task(
+            task_key="run_sdp",
+            depends_on=[TaskDependency(task_key="seed_sources")],
+            pipeline_task=PipelineTask(
+                pipeline_id="${resources.pipelines.job1_sdp.id}",
+                full_refresh=False,
+            ),
+        ),
+    ]
+
+    job = Job(
+        name=f"{JOB_NAME}_${{bundle.target}}_integration",
+        timeout_seconds=3600,
+        max_concurrent_runs=1,
+        queue=QueueSettings(enabled=True),
+        notification_settings=JobNotificationSettings(
+            no_alert_for_canceled_runs=True,
+            no_alert_for_skipped_runs=True,
+        ),
+        parameters=[
+            JobParameterDefinition(name="log_level", default=DEFAULT_LOG_LEVEL),
+            JobParameterDefinition(name="quarantine_fail_ratio", default=PROD_QUARANTINE_FAIL_RATIO),
+            JobParameterDefinition(name="seed_date", default=""),
+        ],
+        tags=_tags("prod"),
+        environments=_environments(),
+        tasks=tasks,
+    )
+
+    d = job.as_dict()
+    d["deployment"] = {"kind": "BUNDLE"}
+    d["run_as"] = {"service_principal_name": sp_id}
+    d["permissions"] = [{"service_principal_name": sp_id, "level": "CAN_MANAGE"}]
+
+    return d
+
+
 def _resolve_catalog(environment: str) -> str:
     """Compute the target catalog at generation time, mirroring Config.__init__.
 
@@ -440,6 +482,8 @@ def main():
     jobs: dict = {JOB_NAME: _build_job(env, sp_id)}
     if env in ("dev", "staging"):
         jobs[f"{JOB_NAME}_integration_test"] = _build_job_integration_test(env, sp_id)
+    if env == "prod":
+        jobs[f"{JOB_NAME}_prod_integration"] = _build_job_prod_integration(sp_id)
 
     output: dict = {
         "resources": {