feat: a continuous trigger on a batch build warns at validate; scaffold-composer schedules only builds a scheduler starts

CHANGELOG.md

@@ -46,6 +46,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - A product whose every build becomes `continuous` (or runs outside forge-cli) gets no
   `schedule/` artifacts, so pipeline stage 11 skips `fluid schedule-sync`, and a DAG
   deployed for it earlier stays in the scheduler: remove that DAG by hand.
+- `fluid scaffold-composer` writes no DAG for a contract whose every build runs outside
+  forge-cli (`engine: flink`, `properties.language: flink_sql`) or continuously and that
+  declares no `orchestration.tasks`, where it used to write one. A DAG it wrote for such a
+  contract before stays in `--out-dir` and in Cloud Composer if you deployed it: delete
+  both. The command names the file when it finds it under the name it writes now.
 - An environment overlay cannot turn a scheduled build continuous: overlays cannot
   delete the base's `cron` or `schedule`, which a continuous trigger refuses. Declare the
   trigger in each environment's overlay instead of in the base.
@@ -61,15 +66,26 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   `timetable` next to it, as Databricks Jobs refuses a continuous trigger next to any
   other, and the error names the field and the trigger type that takes it. It is not
   Airflow's `@continuous`, which starts batch runs back to back. 0.7.5 is unchanged.
-  Declare it on a build that runs outside forge-cli or on a connector: the dbt and python
-  runners refuse it ("Unknown trigger type"), as they refuse `event` and `dataset`, and
-  `fluid scaffold-composer` still writes its nightly DAG for any contract.
+  Declare it on a build that runs outside forge-cli or on a connector: on a build forge-cli
+  runs as a batch run, `fluid validate` warns and `--strict` fails. The dbt and python
+  runners refuse the trigger and say why (the python runner takes every build no other
+  runner does, such as one with no engine or a local `spark` script), and an inline-SQL
+  build or a `duckdb`, `airbyte`, `meltano` or `dlt` acquisition runs only when `fluid
+  apply` runs builds. A continuous `spark`, `glue` or `custom` build without inline SQL
+  that names a `runtime.platform` other than `local`, or inline SQL on a platform with no
+  embedded-SQL executor, gets no warning: `fluid apply --mode amend-and-build` leaves it
+  to that platform, as it does a Flink job.
 
 ### Changed
 
 - A schema error from a `not: {required: [...]}` rule now reads `<field> is not allowed
   here`, or the rule's own `description`, instead of quoting the whole object and the
   subschema it must not match.
+- `fluid scaffold-composer` takes the DAG's cron from the first build a scheduler starts,
+  so a Flink job's own cron no longer schedules it, and writes no DAG for a contract whose
+  every build runs outside forge-cli or continuously, unless it declares
+  `orchestration.tasks`: the test `fluid generate schedule` applies before it writes no
+  files.
 
 ### Fixed
 

fluid_build/build_runners/base.py

@@ -38,7 +38,11 @@
 from fluid_build._console import error as console_error
 from fluid_build._contract_loader import CLIError, load_contract_with_overlay
 from fluid_build.util.binding_paths import ENV_PLACEHOLDER_RE
-from fluid_build.util.contract import describe_externally_run_build, is_externally_run_build
+from fluid_build.util.contract import (
+    describe_externally_run_build,
+    is_continuous_build,
+    is_externally_run_build,
+)
 
 LOG = logging.getLogger("fluid.build_runners")
 
@@ -142,6 +146,10 @@ def is_embedded_sql_build(build: Dict[str, Any]) -> bool:
 #: behaviour, so contracts that never declared one are unaffected.
 LOCAL_SQL_PLATFORMS = frozenset({"", "local", "duckdb"})
 
+#: The platforms :func:`_execute_embedded_sql_build` runs inline SQL on: the
+#: local DuckDB engine, and Snowflake. It refuses any other declared platform.
+EMBEDDED_SQL_PLATFORMS = LOCAL_SQL_PLATFORMS | {"snowflake"}
+
 
 def embedded_sql_platform(build: Dict[str, Any]) -> str:
     """Return the lower-cased ``execution.runtime.platform`` of a build.
@@ -205,6 +213,188 @@ def is_acquisition_build(build: Dict[str, Any]) -> bool:
     return engine in ACQUISITION_ENGINES
 
 
+#: What to do instead of a continuous trigger on a batch build, in the
+#: ``fluid validate`` warning and in a runner's refusal alike.
+CONTINUOUS_REMEDY = (
+    "Use trigger.type schedule or manual, or run the job on a platform that keeps it running."
+)
+
+#: Acquisition engines whose runner leaves a job running after ``fluid apply``
+#: ends: the Kafka Connect or Debezium connector it deploys keeps streaming.
+#: They are the runners that declare ``RunnerCapability.STREAMING``; every
+#: other runner here runs a build to completion, a batch run.
+STREAMING_ACQUISITION_ENGINES = frozenset({"kafka-connect", "debezium"})
+
+#: Engines whose job can run on a platform the build names in
+#: ``execution.runtime.platform``, such as a Spark or Glue streaming job on
+#: Databricks or AWS Glue. Without such a platform the python runner runs the
+#: build's script locally, as for any engine without a runner of its own.
+PLATFORM_RUN_ENGINES = frozenset({"spark", "glue", "custom"})
+
+
+def _batch_runner_for(build: Dict[str, Any]) -> Optional[Tuple[str, Optional[str]]]:
+    """``(builds, refusing_runner)`` when :func:`run_builds_from_args` runs the
+    build to completion, else None.
+
+    ``builds`` names the builds that runner takes, for messages. The dbt and
+    python runners refuse a continuous trigger, and ``refusing_runner`` names
+    them; inline SQL and the duckdb, airbyte, meltano and dlt runners run the
+    build each time ``fluid apply`` runs it, with ``refusing_runner`` None.
+    The checks follow the dispatcher's order, so they name the runner that gets
+    the build. The python runner takes every build the others do not: python,
+    no engine, spark, glue, custom, and an ingestion engine declared without
+    ``pattern: acquisition``; it runs the build's script locally.
+
+    None for a connector, which stays running, and for a build forge-cli does
+    not run: inline SQL on a platform with no embedded-SQL executor, and a
+    spark, glue or custom build that declares a ``runtime.platform`` other
+    than the local one, where that platform runs the job. The dispatcher
+    leaves a continuous one of those to its platform
+    (:func:`_runs_outside_forge_continuously`).
+    """
+    engine, pattern, props = build.get("engine"), build.get("pattern"), build.get("properties")
+    # The dispatch predicates read these shapes, and fluid validate also reads
+    # a contract the schema refused.
+    if not (
+        isinstance(engine, (str, type(None)))
+        and isinstance(pattern, (str, type(None)))
+        and isinstance(props, (dict, type(None)))
+    ):
+        return None
+    name = (engine or "").strip().lower()
+    if is_externally_run_build(build):
+        return None
+    if is_acquisition_build(build):
+        return None if name in STREAMING_ACQUISITION_ENGINES else (f"{name} builds", None)
+    if is_embedded_sql_build(build):
+        if embedded_sql_platform(build) not in EMBEDDED_SQL_PLATFORMS:
+            return None
+        return "inline-SQL builds", None
+    if is_dbt_build(build):
+        return "dbt builds", "dbt"
+    if name in PLATFORM_RUN_ENGINES and embedded_sql_platform(build) not in LOCAL_SQL_PLATFORMS:
+        return None
+    if not name:
+        return "builds with no engine", "python"
+    if name in ACQUISITION_ENGINES:
+        return f"{name} builds without pattern: acquisition", "python"
+    return f"{name} builds", "python"
+
+
+def continuous_trigger_warnings(contract: Dict[str, Any]) -> List[str]:
+    """``fluid validate``'s warnings for continuous builds forge-cli runs as batch runs.
+
+    ``trigger.type: continuous`` (schema 0.7.6) says the job's platform keeps
+    it running, so no scheduler starts the build
+    (:func:`~fluid_build.util.contract.is_unscheduled_build`). A batch run
+    ends: the dbt and python runners refuse the trigger, so forge-cli never
+    runs such a build (the python runner takes every build no other runner
+    does), and an inline-SQL build or a duckdb, airbyte, meltano or dlt
+    acquisition runs only when ``fluid apply`` runs builds. A warning, not an
+    error: plain ``fluid validate`` passes, and ``--strict`` fails. A
+    continuous build that forge-cli does not run gets none: its platform keeps
+    it running, and ``fluid apply`` leaves it there
+    (:func:`_runs_outside_forge_continuously`).
+
+    Borrowed: Spark checks a continuous trigger against what the query can run
+    before it starts (``UnsupportedOperationChecker.checkForContinuous``), and
+    Databricks refuses ``Trigger.Continuous`` where it cannot keep a job
+    running (``INFINITE_STREAMING_TRIGGER_NOT_SUPPORTED``), naming the trigger.
+    """
+    builds = contract.get("builds")
+    if not isinstance(builds, list):
+        return []
+    warnings: List[str] = []
+    for index, build in enumerate(builds):
+        if not isinstance(build, dict) or not is_continuous_build(build):
+            continue
+        runner = _batch_runner_for(build)
+        if runner is None:
+            continue
+        runs, refuser = runner
+        bid = build.get("id")
+        where = (
+            f"builds[{index}] (id {bid!r})" if isinstance(bid, str) and bid else f"builds[{index}]"
+        )
+        engine = build.get("engine")
+        on = f"engine {engine!r}" if engine else "a build with no engine"
+        if refuser:
+            outcome = (
+                f", and its {refuser} runner refuses a continuous trigger: forge-cli never "
+                "runs this build"
+            )
+        else:
+            outcome = (
+                ": this build runs only when fluid apply runs builds, and no scheduler starts it"
+            )
+        remedy = CONTINUOUS_REMEDY
+        if refuser == "python" and str(engine or "").strip().lower() in PLATFORM_RUN_ENGINES:
+            # Naming that platform is what leaves the job to it.
+            remedy = remedy[:-1] + ", named in execution.runtime.platform."
+        warnings.append(
+            f"{where}: trigger.type continuous on {on}: forge-cli runs {runs} as batch runs, "
+            f"which end{outcome}. {remedy}"
+        )
+    return warnings
+
+
+def refuse_continuous_trigger(runs: str) -> int:
+    """Say why a batch runner refuses ``trigger.type: continuous``; return exit code 1.
+
+    ``runs`` says what the runner runs and that it ends, for example ``"dbt
+    builds as batch runs, which end"``. ``fluid validate`` warns about the same
+    build (:func:`continuous_trigger_warnings`).
+    """
+    cprint(
+        f"   ❌ trigger.type continuous: forge-cli runs {runs}, and a continuous trigger says "
+        "the job's platform keeps it running."
+    )
+    cprint(f"   {CONTINUOUS_REMEDY}")
+    cprint(f"{'=' * 80}")
+    return 1
+
+
+def _runs_outside_forge_continuously(build: Dict[str, Any]) -> bool:
+    """True for a continuous build that forge-cli does not run.
+
+    That is inline SQL on a platform with no embedded-SQL executor, or a
+    spark, glue or custom build that declares a ``runtime.platform`` other
+    than the local one (:func:`_batch_runner_for` is None for each). That
+    platform keeps the job running, so :func:`run_builds_from_args` leaves it
+    there as it does a Flink job, and ``fluid validate`` gives no warning. A
+    Kafka Connect or Debezium connector is not one: its runner deploys it.
+    """
+    if not is_continuous_build(build) or is_externally_run_build(build):
+        return False
+    if _batch_runner_for(build) is not None:
+        return False
+    engine, pattern = build.get("engine"), build.get("pattern")
+    connector = isinstance(engine, str) and isinstance(pattern, str) and is_acquisition_build(build)
+    return not connector
+
+
+def _report_continuous_outside_forge(build: Dict[str, Any], build_id: str, report: Any) -> None:
+    """Say that a continuous build forge-cli does not run is not run here.
+
+    Counted and recorded as :func:`_report_externally_run_build` counts and
+    records a Flink job: on its own summary line, and as skipped with
+    ``reason: runs_outside_forge`` in the run report.
+    """
+    engine = build.get("engine")
+    what = f"engine '{engine}'" if engine else "inline SQL"
+    platform = embedded_sql_platform(build)
+    # A local platform is forge-cli's own, so it cannot be the one that keeps
+    # the job running.
+    platform = "its own platform" if platform in LOCAL_SQL_PLATFORMS else platform
+    cprint(
+        f"\n⏭️  Build '{build_id}' - trigger.type continuous on {what}: {platform} keeps the job "
+        "running, outside forge-cli; there is nothing for forge-cli to run.",
+        markup=False,
+    )
+    if report is not None:
+        report.record_build(build_id=build_id, status="skipped", reason="runs_outside_forge")
+
+
 # Acquisition-engine registry: ``engine name → (module path, callable name)``.
 # Adding a new engine is one entry here + a runner module — no edits to
 # the dispatcher's switch chain. Modules are imported lazily so a missing
@@ -652,13 +842,13 @@ def _execute_embedded_sql_build(
         return 0
 
     if platform not in LOCAL_SQL_PLATFORMS:
-        if platform != "snowflake":
+        if platform not in EMBEDDED_SQL_PLATFORMS:
             cprint(
                 f"   ❌ Build '{build_id}' declares "
                 f"execution.runtime.platform: '{platform}', which has no "
                 "embedded-SQL executor. forge-cli will not downgrade a declared "
                 "platform to the local DuckDB engine. Supported: "
-                f"{', '.join(sorted(LOCAL_SQL_PLATFORMS - {''}))}, snowflake."
+                f"{', '.join(sorted(EMBEDDED_SQL_PLATFORMS - {''}))}."
             )
             LOG.error(
                 "embedded_sql_platform_unsupported build_id=%s platform=%s",
@@ -1208,6 +1398,11 @@ def run_builds_from_args(
             total_external += 1
             continue
 
+        if _runs_outside_forge_continuously(build):
+            _report_continuous_outside_forge(build, build_id, report)
+            total_external += 1
+            continue
+
         if is_acquisition_build(build):
             sample_rows = getattr(args, "sample_rows", None)
             # A plan's contract had its ``{{ env.* }}`` templates resolved

fluid_build/build_runners/dbt/runner.py

@@ -45,9 +45,10 @@
 
 from fluid_build._console import cprint, success
 from fluid_build._console import error as console_error
+from fluid_build.util.contract import is_continuous_build
 
 from .._path_safety import confine_to_workspace
-from ..base import SENSITIVE_ENV_KEY_RE, _resolve_env_placeholders
+from ..base import SENSITIVE_ENV_KEY_RE, _resolve_env_placeholders, refuse_continuous_trigger
 from .profiles import (
     _list_profile_targets,
     _load_dbt_project_config,
@@ -939,6 +940,11 @@ def execute_dbt_build(
         trigger_label = "schedule (manual apply override)"
     cprint(f"   Trigger: {trigger_label}")
 
+    # trigger.type continuous (schema 0.7.6) says the job's platform keeps it
+    # running, and a dbt run ends: refused before a profile is prepared.
+    if is_continuous_build(build):
+        return refuse_continuous_trigger("dbt builds as batch runs, which end")
+
     project_config = _load_dbt_project_config(project_dir)
     profiles_dir, temp_profiles_dir = resolve_dbt_profiles_dir(build, project_dir, project_config)
 

fluid_build/build_runners/python/runner.py

@@ -34,8 +34,10 @@
 from fluid_build._console import cprint, success
 from fluid_build._console import error as console_error
 from fluid_build.credentials.dotenv_store import environment_for_project_code
+from fluid_build.util.contract import is_continuous_build
 
 from .._path_safety import confine_to_workspace, resolve_workspace_root
+from ..base import refuse_continuous_trigger
 
 LOG = logging.getLogger("fluid.build_runners.python")
 
@@ -99,6 +101,11 @@ def execute_build(
         trigger_label = "schedule (manual apply override)"
     cprint(f"   Trigger: {trigger_label}")
 
+    # trigger.type continuous (schema 0.7.6) says the job's platform keeps it
+    # running, and the script's run ends.
+    if is_continuous_build(build):
+        return refuse_continuous_trigger("a build's Python script as a batch run, which ends")
+
     if trigger_type == "manual" or (trigger_type == "schedule" and force_run):
         iterations = 1 if trigger_type == "schedule" and force_run else trigger.get("iterations", 1)
         delay_from_contract = trigger.get("delaySeconds")

fluid_build/cli/generate_schedule.py

@@ -50,15 +50,6 @@
 SUBCOMMAND = "schedule"
 
 
-def _only_unscheduled_builds(contract: Dict[str, Any]) -> bool:
-    """True when the contract has builds and no scheduler starts any of them:
-    each runs outside forge-cli, or continuously."""
-    from fluid_build.util.contract import is_unscheduled_build
-
-    builds = [b for b in contract.get("builds") or [] if isinstance(b, dict)]
-    return bool(builds) and all(is_unscheduled_build(b) for b in builds)
-
-
 def register_subcommand(subparsers: argparse._SubParsersAction):
     """Register the schedule subcommand under ``fluid generate``."""
     p = subparsers.add_parser(
@@ -155,6 +146,7 @@ def run(args: Any, logger: logging.Logger) -> int:
 
         # --- Resolve scheduler ---
         from fluid_build.schedulers.airflow import fluid_apply
+        from fluid_build.util.contract import has_only_unscheduled_builds
 
         scheduler_name = getattr(args, "scheduler", None)
         if not scheduler_name:
@@ -170,7 +162,7 @@ def run(args: Any, logger: logging.Logger) -> int:
         # a build whose trigger is continuous) has nothing to schedule, which is
         # not an error: unless it hand-declares orchestration tasks, or names a
         # scheduler that does not exist.
-        if _only_unscheduled_builds(contract) and not fluid_apply.has_explicit_tasks(contract):
+        if has_only_unscheduled_builds(contract) and not fluid_apply.has_explicit_tasks(contract):
             from fluid_build.schedulers import has_scheduler
 
             if not scheduler_name or has_scheduler(scheduler_name):