diff --git a/CHANGELOG.md b/CHANGELOG.md
index 543d2ea54..76fdef28a 100644
--- a/CHANGELOG.md
+++ b/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
 
diff --git a/fluid_build/build_runners/base.py b/fluid_build/build_runners/base.py
index c0d3dd047..fbbc3b384 100644
--- a/fluid_build/build_runners/base.py
+++ b/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
diff --git a/fluid_build/build_runners/dbt/runner.py b/fluid_build/build_runners/dbt/runner.py
index ee9c0fd02..bad6a915d 100644
--- a/fluid_build/build_runners/dbt/runner.py
+++ b/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)
 
diff --git a/fluid_build/build_runners/python/runner.py b/fluid_build/build_runners/python/runner.py
index fff454c9c..aee8e8f73 100644
--- a/fluid_build/build_runners/python/runner.py
+++ b/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")
diff --git a/fluid_build/cli/generate_schedule.py b/fluid_build/cli/generate_schedule.py
index e027a71c9..a155d7e60 100644
--- a/fluid_build/cli/generate_schedule.py
+++ b/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):
diff --git a/fluid_build/cli/scaffold_composer.py b/fluid_build/cli/scaffold_composer.py
index 3ccae0730..a7553e9f0 100644
--- a/fluid_build/cli/scaffold_composer.py
+++ b/fluid_build/cli/scaffold_composer.py
@@ -19,8 +19,11 @@
 import os
 import re
 import shlex
+from typing import Any, Dict, List
 
+from .._console import cprint
 from ..providers.common.codegen_utils import py_str_literal, sanitize_identifier
+from ..util.contract import has_only_unscheduled_builds, is_unscheduled_build
 from ._acquisition_stage_ext import _validate_cron as _validate_cron_grammar
 from ._common import CLIError, load_contract_with_overlay
 from ._logging import info
@@ -157,30 +160,75 @@ def _build_pipeline_bash_commands(contract_path: str, provider: str) -> tuple[st
     return validate_cmd, plan_cmd, apply_cmd
 
 
-def _extract_cron(c: dict) -> object:
-    """Pull the trigger cron from a contract, tolerating both the
-    normalized ``builds: [...]`` list shape (what
-    ``load_contract_with_overlay`` emits) and the legacy singular
-    ``build: {...}`` mapping. Returns the default daily cron when absent."""
+def _builds(c: Dict[str, Any]) -> List[Dict[str, Any]]:
+    """The contract's builds: the normalized ``builds: [...]`` list (what
+    ``load_contract_with_overlay`` emits), then the legacy singular
+    ``build: {...}`` mapping."""
     builds = c.get("builds")
-    if isinstance(builds, list):
-        for b in builds:
-            if isinstance(b, dict):
-                cron = ((b.get("execution") or {}).get("trigger") or {}).get("cron")
-                if cron is not None:
-                    return cron
+    found = [b for b in builds if isinstance(b, dict)] if isinstance(builds, list) else []
     legacy = c.get("build")
     if isinstance(legacy, dict):
-        cron = ((legacy.get("execution") or {}).get("trigger") or {}).get("cron")
+        found.append(legacy)
+    return found
+
+
+def _extract_cron(c: dict) -> object:
+    """Pull the trigger cron of the first build a scheduler starts, from the
+    normalized ``builds: [...]`` list or the legacy singular ``build: {...}``
+    mapping. A build no scheduler starts (a Flink job, or a continuous trigger:
+    :func:`~fluid_build.util.contract.is_unscheduled_build`) is skipped, so a
+    cron it carries for its own platform does not schedule this DAG. Returns
+    the default daily cron when absent."""
+    for b in _builds(c):
+        if is_unscheduled_build(b):
+            continue
+        cron = ((b.get("execution") or {}).get("trigger") or {}).get("cron")
         if cron is not None:
             return cron
     return "0 2 * * *"
 
 
+def _wrote_dag(dag_path: str, dag_id: str) -> bool:
+    """True when ``dag_path`` is a DAG this command wrote for ``dag_id``: it
+    carries the template's ``tags=["FLUID"]`` and its ``dag_id``, quoted as
+    ``py_str_literal`` writes it or as the template did before it."""
+    if not os.path.isfile(dag_path):
+        return False
+    try:
+        with open(dag_path, encoding="utf-8", errors="replace") as f:
+            head = f.read(65536)
+    except OSError:
+        return False
+    return 'tags=["FLUID"]' in head and (
+        f"dag_id={py_str_literal(dag_id)}," in head or f'dag_id="{dag_id}",' in head
+    )
+
+
+def _report_nothing_to_schedule(dag_path: str, dag_id: str) -> None:
+    """Say why no DAG is written, as ``fluid generate schedule`` does, and name
+    a DAG this command wrote for the product before: left in place, it keeps
+    scheduling the product wherever it was deployed. A file of the same name
+    that it did not write is someone else's, and is not mentioned."""
+    cprint(
+        "Every build runs outside forge-cli (a Flink job, for example) or continuously "
+        "(trigger.type: continuous), so there is nothing to schedule. No DAG written."
+    )
+    if _wrote_dag(dag_path, dag_id):
+        # Printed as written: soft_wrap leaves wrapping to the terminal, so the
+        # path is never broken by an inserted newline and copies whole, and no
+        # [markup] or :emoji: code in it is rendered.
+        cprint(
+            f"A DAG this command wrote for the product before is still at {dag_path}: delete "
+            "it, and remove it from Cloud Composer if you deployed it there.",
+            markup=False,
+            emoji=False,
+            soft_wrap=True,
+        )
+
+
 def run(args, logger: logging.Logger) -> int:
     try:
         c = load_contract_with_overlay(args.contract, getattr(args, "env", None), logger)
-        cron = _validate_cron(_extract_cron(c))
         # SECURITY (path traversal + filename injection): the contract's
         # id/name is attacker-influenced and was previously only ``.``->``_``
         # cleaned, which leaves ``/`` (path traversal) and an absolute name
@@ -190,6 +238,17 @@ def run(args, logger: logging.Logger) -> int:
         # to ``_`` and strips leading digits, yielding a single safe path
         # component that doubles as a valid Python ``dag_id``.
         dag_id = sanitize_identifier(c.get("id") or c.get("name") or "fluid_product")
+        # No scheduler starts a Flink job or a continuous build, so a contract
+        # with no other build has nothing to schedule and gets no DAG. The
+        # test is `fluid generate schedule`'s, hand-declared orchestration
+        # tasks included, so the two commands agree on such a contract.
+        from fluid_build.schedulers.airflow.fluid_apply import has_explicit_tasks
+
+        if has_only_unscheduled_builds(c) and not has_explicit_tasks(c):
+            dag_path = os.path.join(os.path.abspath(args.out_dir), f"{dag_id}.py")
+            _report_nothing_to_schedule(dag_path, dag_id)
+            return 0
+        cron = _validate_cron(_extract_cron(c))
         # SECURITY (shell + Python-literal injection): the contract path is
         # interpolated into a generated ``bash_command`` that a shell runs at
         # task time AND into Python source Airflow imports at DAG-parse time.
diff --git a/fluid_build/cli/validate.py b/fluid_build/cli/validate.py
index 2fc56268c..254427f87 100644
--- a/fluid_build/cli/validate.py
+++ b/fluid_build/cli/validate.py
@@ -899,6 +899,24 @@ def _run_contract_rules(
             if args.verbose:
                 info(logger, f"Vector binding check skipped: {exc}")
 
+        # --- A continuous trigger on a batch build (WARNING) ---------------
+        # trigger.type continuous (0.7.6) says the job's platform keeps it
+        # running, so no scheduler starts the build. forge-cli runs dbt,
+        # python, inline-SQL and batch-acquisition builds to completion: the
+        # dbt and python runners refuse the trigger, and only fluid apply runs
+        # the others. A warning, not an error: plain validate passes and
+        # --strict fails. A crash in the check is reported, never swallowed.
+        try:
+            from fluid_build.build_runners.base import continuous_trigger_warnings
+
+            for msg in continuous_trigger_warnings(contract):
+                validation_result.add_warning(msg)
+        except Exception as exc:  # noqa: BLE001 — reported, never swallowed
+            validation_result.add_warning(
+                f"Continuous-trigger check could not run ({type(exc).__name__}: {exc}); "
+                "the contract was NOT checked for a continuous trigger on a batch build"
+            )
+
         # --- Packaging checks (RFC-packaging-modes.md file 9) ---------------
         # ``fluid validate`` had NO packaging awareness: a `shared` block with
         # no `pool` validated clean and then failed at `fluid plan` with an
diff --git a/fluid_build/util/contract.py b/fluid_build/util/contract.py
index fd9922bfd..aec0bc5aa 100644
--- a/fluid_build/util/contract.py
+++ b/fluid_build/util/contract.py
@@ -176,6 +176,14 @@ def is_unscheduled_build(build: Mapping[str, Any]) -> bool:
     return is_externally_run_build(build) or is_continuous_build(build)
 
 
+def has_only_unscheduled_builds(contract: Mapping[str, Any]) -> bool:
+    """True when the contract has builds and no scheduler starts any of them:
+    each runs outside forge-cli, or continuously (:func:`is_unscheduled_build`).
+    A contract with no builds is False: no build of it was left out."""
+    builds = [b for b in get_builds(contract) if isinstance(b, Mapping)]
+    return bool(builds) and all(is_unscheduled_build(b) for b in builds)
+
+
 def get_contract_version(contract: Mapping[str, Any]) -> Optional[str]:
     """
     Get the FLUID schema version from a contract.
diff --git a/tests/test_continuous_trigger.py b/tests/test_continuous_trigger.py
index 2d5e5cc2e..5e1b603a6 100644
--- a/tests/test_continuous_trigger.py
+++ b/tests/test_continuous_trigger.py
@@ -236,3 +236,473 @@ def test_generate_schedule_has_nothing_to_schedule(self, tmp_path: Path) -> None
         assert result.returncode == 0, result.stdout + result.stderr
         assert "continuously (trigger.type: continuous)" in " ".join(result.stdout.split())
         assert not (tmp_path / "sched").exists()
+
+
+# ── a continuous trigger on a build forge-cli runs as a batch run ──────────
+
+_SQL = {"sql": "SELECT 1"}
+_REMEDY = "Use trigger.type schedule or manual, or run the job on a platform that keeps it running."
+# A spark, glue or custom build is left to the platform it names.
+_PLATFORM_REMEDY = _REMEDY[:-1] + ", named in execution.runtime.platform."
+
+
+def _build(engine: Any = None, trigger: str = "continuous", **extra: Any) -> Dict[str, Any]:
+    build: Dict[str, Any] = {"id": "b", "execution": {"trigger": {"type": trigger}}, **extra}
+    if engine is not None:
+        build["engine"] = engine
+    return build
+
+
+def _runtime(platform: str = "databricks") -> Dict[str, Any]:
+    return {"trigger": {"type": "continuous"}, "runtime": {"platform": platform}}
+
+
+def _warnings(contract: Dict[str, Any]) -> list:
+    from fluid_build.build_runners.base import continuous_trigger_warnings
+
+    return continuous_trigger_warnings(contract)
+
+
+class TestBatchBuildWarning:
+    """A continuous trigger says the job's platform keeps it running, and no
+    scheduler starts the build; forge-cli runs dbt, python, inline-SQL and batch
+    acquisition builds to completion, so ``fluid validate`` warns."""
+
+    def test_a_dbt_build_is_never_run(self) -> None:
+        assert _warnings({"builds": [_build("dbt")]}) == [
+            "builds[0] (id 'b'): trigger.type continuous on engine 'dbt': forge-cli runs dbt "
+            "builds as batch runs, which end, and its dbt runner refuses a continuous trigger: "
+            f"forge-cli never runs this build. {_REMEDY}"
+        ]
+
+    def test_only_fluid_apply_runs_a_duckdb_acquisition(self) -> None:
+        assert _warnings({"builds": [_build("duckdb", pattern="acquisition")]}) == [
+            "builds[0] (id 'b'): trigger.type continuous on engine 'duckdb': forge-cli runs "
+            "duckdb builds as batch runs, which end: this build runs only when fluid apply runs "
+            f"builds, and no scheduler starts it. {_REMEDY}"
+        ]
+
+    @pytest.mark.parametrize(
+        "build, says",
+        [
+            pytest.param(_build("dbt-snowflake"), "its dbt runner refuses", id="dbt-adapter"),
+            pytest.param(_build(" Python "), "its python runner refuses", id="python"),
+            # The dispatcher hands a build with no engine to the python runner.
+            pytest.param(
+                _build(),
+                "on a build with no engine: forge-cli runs builds with no engine as batch runs, "
+                "which end, and its python runner refuses",
+                id="no-engine",
+            ),
+            pytest.param(_build("sql", properties=_SQL), "inline-SQL builds", id="sql"),
+            pytest.param(
+                _build("sql", properties=_SQL, execution=_runtime("snowflake")),
+                "inline-SQL builds",
+                id="sql-on-snowflake",
+            ),
+            pytest.param(
+                _build(pattern="embedded-logic", properties=_SQL),
+                "on a build with no engine: forge-cli runs inline-SQL builds",
+                id="inline-sql-no-engine",
+            ),
+            # The runner takes inline SQL ahead of dbt, so the build runs on DuckDB.
+            pytest.param(
+                _build("dbt", pattern="embedded-logic", properties=_SQL),
+                "on engine 'dbt': forge-cli runs inline-SQL builds",
+                id="dbt-with-inline-sql",
+            ),
+            *[
+                pytest.param(_build(e, pattern="acquisition"), f"runs {e} builds", id=e)
+                for e in ("airbyte", "meltano", "dlt")
+            ],
+            # The python runner runs the script of any build no other runner
+            # takes, locally, unless a spark, glue or custom build names a
+            # platform of its own.
+            *[
+                pytest.param(_build(e), f"runs {e} builds as batch runs", id=e)
+                for e in ("spark", "glue", "custom")
+            ],
+            pytest.param(
+                _build("spark", execution=_runtime("local")),
+                "runs spark builds as batch runs",
+                id="spark-on-local",
+            ),
+            pytest.param(
+                _build("dlt", pattern="hybrid-reference"),
+                "forge-cli runs dlt builds without pattern: acquisition as batch runs, which "
+                "end, and its python runner refuses",
+                id="dlt-without-acquisition",
+            ),
+            pytest.param(
+                _build("kafka-connect"),
+                "runs kafka-connect builds without pattern: acquisition",
+                id="connector-without-acquisition",
+            ),
+        ],
+    )
+    def test_every_batch_runner_is_warned(self, build: Dict[str, Any], says: str) -> None:
+        (warning,) = _warnings({"builds": [build]})
+        assert says in warning
+        platform_run = build.get("engine") in ("spark", "glue", "custom")
+        assert warning.endswith(_PLATFORM_REMEDY if platform_run else _REMEDY)
+
+    @pytest.mark.parametrize(
+        "build",
+        [
+            pytest.param(_build("kafka-connect", pattern="acquisition"), id="kafka-connect"),
+            pytest.param(_build("debezium", pattern="acquisition"), id="debezium"),
+            pytest.param(_build("flink", properties=_SQL), id="flink"),
+            pytest.param(
+                _build("sql", properties={"language": "flink_sql", **_SQL}), id="flink-sql"
+            ),
+            # forge-cli does not run these: fluid apply leaves them to the
+            # platform they name, as it does a Flink job.
+            *[
+                pytest.param(_build(e, execution=_runtime(p)), id=f"{e}-on-{p}")
+                for e, p in (("spark", "databricks"), ("glue", "aws"), ("custom", "kubernetes"))
+            ],
+            pytest.param(
+                _build("spark", pattern="embedded-logic", properties=_SQL, execution=_runtime()),
+                id="inline-sql-on-databricks",
+            ),
+            pytest.param(_build("dbt", trigger="schedule"), id="schedule"),
+            pytest.param({"id": "b", "engine": "dbt"}, id="no-trigger"),
+        ],
+    )
+    def test_no_warning(self, build: Dict[str, Any]) -> None:
+        assert _warnings({"builds": [build]}) == []
+
+    @pytest.mark.parametrize(
+        "contract",
+        [
+            pytest.param({"builds": "dbt"}, id="builds-str"),
+            pytest.param({"builds": None}, id="builds-null"),
+            pytest.param({"builds": ["dbt", None, 3]}, id="build-not-object"),
+            pytest.param({"builds": [_build(5)]}, id="engine-int"),
+            pytest.param({"builds": [_build("sql", pattern=["x"])]}, id="pattern-list"),
+            pytest.param(
+                {"builds": [_build(pattern="embedded-logic", properties=["SELECT 1"])]},
+                id="properties-list",
+            ),
+            pytest.param({"builds": [{"id": "b", "engine": "dbt", "execution": "x"}]}, id="exec"),
+        ],
+    )
+    def test_any_shape_is_read_without_raising(self, contract: Dict[str, Any]) -> None:
+        # fluid validate runs its contract rules on a contract the schema refused too.
+        assert _warnings(contract) == []
+
+    def test_a_spark_build_is_told_where_to_name_its_platform(self) -> None:
+        (warning,) = _warnings({"builds": [_build("spark")]})
+        assert warning.endswith(_PLATFORM_REMEDY)
+        assert _warnings({"builds": [_build("spark", execution=_runtime())]}) == []
+
+    def test_a_build_without_an_id_is_named_by_index(self) -> None:
+        build = _build("python")
+        del build["id"]
+        (warning,) = _warnings({"builds": [_build("flink"), build]})
+        assert warning.startswith("builds[1]: trigger.type continuous on engine 'python':")
+
+
+def test_streaming_engines_are_the_runners_that_declare_streaming() -> None:
+    # The kafka-connect and debezium runners leave a connector running; every
+    # other acquisition runner runs to completion. Read from the runners, so a
+    # new streaming runner cannot be warned about by mistake.
+    from fluid_build.api.runner import RunnerCapability
+    from fluid_build.build_runners.base import (
+        ACQUISITION_ENGINES,
+        STREAMING_ACQUISITION_ENGINES,
+        _declared_capabilities_for_engine,
+    )
+
+    declared = {engine: _declared_capabilities_for_engine(engine) for engine in ACQUISITION_ENGINES}
+    assert all(caps is not None for caps in declared.values()), declared
+    assert STREAMING_ACQUISITION_ENGINES == {
+        engine
+        for engine, caps in declared.items()
+        if caps is not None and RunnerCapability.STREAMING.value in caps
+    }
+
+
+_DISPATCHED = [
+    pytest.param(_build("dbt"), id="dbt"),
+    pytest.param(_build("dbt-snowflake"), id="dbt-adapter"),
+    pytest.param(_build("python"), id="python"),
+    pytest.param(_build(), id="no-engine"),
+    pytest.param(_build("sql", properties=_SQL), id="sql"),
+    pytest.param(_build("sql", properties=_SQL, execution=_runtime("snowflake")), id="snowflake"),
+    pytest.param(_build("spark", pattern="embedded-logic", properties=_SQL), id="spark-sql-local"),
+    pytest.param(
+        _build("spark", pattern="embedded-logic", properties=_SQL, execution=_runtime()),
+        id="spark-sql-databricks",
+    ),
+    *[
+        pytest.param(_build(engine, pattern="acquisition"), id=f"{engine}-acquisition")
+        for engine in ("duckdb", "airbyte", "meltano", "dlt", "kafka-connect", "debezium")
+    ],
+    pytest.param(_build("duckdb", pattern="hybrid-reference"), id="duckdb-not-acquisition"),
+    pytest.param(_build("kafka-connect"), id="kafka-connect-no-pattern"),
+    pytest.param(_build("spark", pattern="hybrid-reference"), id="spark"),
+    pytest.param(_build("glue"), id="glue"),
+    pytest.param(_build("custom"), id="custom"),
+    pytest.param(_build("spark", execution=_runtime()), id="spark-on-databricks"),
+    pytest.param(_build("glue", execution=_runtime("aws")), id="glue-on-aws"),
+    pytest.param(_build("dlt", pattern="hybrid-reference"), id="dlt-not-acquisition"),
+    pytest.param(_build("flink", properties=_SQL), id="flink"),
+    pytest.param(_build("sql", properties={"language": "flink_sql", **_SQL}), id="flink-sql"),
+]
+
+
+def _dispatch(build: Dict[str, Any], tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> list:
+    """The runners ``fluid apply --mode amend-and-build`` hands *build* to,
+    recorded instead of run."""
+    import argparse
+    import logging
+
+    from fluid_build.build_runners import base
+    from fluid_build.build_runners.dbt import runner as dbt_runner
+    from fluid_build.build_runners.python import runner as python_runner
+
+    called: list = []
+
+    def _record(name: str) -> Any:
+        def _run(*_args: Any, **_kwargs: Any) -> int:
+            called.append(name)
+            return 0
+
+        return _run
+
+    monkeypatch.setattr(base, "_execute_acquisition_build", _record("acquisition"))
+    monkeypatch.setattr(base, "_execute_embedded_sql_build", _record("inline-sql"))
+    monkeypatch.setattr(dbt_runner, "execute_dbt_build", _record("dbt"))
+    monkeypatch.setattr(python_runner, "execute_build", _record("python"))
+    (tmp_path / "contract.fluid.yaml").write_text("id: p\n", "utf-8")
+    (tmp_path / "ingest.py").write_text("print('ran')\n", "utf-8")
+    (tmp_path / "dbt_project.yml").write_text("name: p\nprofile: p\n", "utf-8")
+    args = argparse.Namespace(
+        contract=str(tmp_path / "contract.fluid.yaml"),
+        build_id=None,
+        dry_run=False,
+        fail_fast=False,
+        delay=0,
+        no_output=True,
+        sample_rows=None,
+        env=None,
+        mode="amend-and-build",
+        allow_skipped_builds=False,
+    )
+    rc = base.run_builds_from_args(
+        args, logging.getLogger("test"), force_run=True, contract={"id": "p", "builds": [build]}
+    )
+    assert rc == 0
+    return called
+
+
+@pytest.mark.parametrize("build", _DISPATCHED)
+def test_validate_warns_exactly_when_apply_runs_the_build_as_a_batch_run(
+    build: Dict[str, Any],
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture,
+) -> None:
+    # One predicate behind both: a warned build reaches a batch runner under
+    # fluid apply --mode amend-and-build, and any other reaches a connector's
+    # runner or none, as a Flink job does.
+    from fluid_build.build_runners.base import STREAMING_ACQUISITION_ENGINES
+
+    called = _dispatch(build, tmp_path, monkeypatch)
+    connector = called == ["acquisition"] and build["engine"] in STREAMING_ACQUISITION_ENGINES
+    assert bool(_warnings({"builds": [build]})) is (bool(called) and not connector), called
+    if not called:
+        assert "there is nothing for forge-cli to run" in " ".join(capsys.readouterr().out.split())
+
+
+@pytest.mark.parametrize(
+    "build",
+    [
+        pytest.param(_build("spark", pattern="hybrid-reference"), id="spark"),
+        pytest.param(_build("custom", execution=_runtime("local")), id="custom-on-local"),
+        pytest.param(_build("dlt", pattern="hybrid-reference"), id="dlt-not-acquisition"),
+        pytest.param(_build("kafka-connect"), id="kafka-connect-no-pattern"),
+    ],
+)
+def test_apply_hands_a_local_script_to_the_python_runner(
+    build: Dict[str, Any], tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    # The python runner runs these builds' scripts when the trigger is
+    # manual, so a continuous one is refused there, not passed off as a job
+    # some platform keeps running.
+    assert _dispatch(build, tmp_path, monkeypatch) == ["python"]
+
+
+class TestValidateCli:
+    @staticmethod
+    def _write(tmp_path: Path) -> None:
+        contract = _contract()
+        contract["builds"] = [
+            {
+                "id": "transform",
+                "engine": "dbt",
+                "pattern": "hybrid-reference",
+                "repository": "./dbt",
+                "execution": {"trigger": {"type": "continuous"}},
+            }
+        ]
+        (tmp_path / "contract.fluid.yaml").write_text(yaml.safe_dump(contract), "utf-8")
+
+    def test_plain_validate_warns_and_passes(self, tmp_path: Path) -> None:
+        self._write(tmp_path)
+        result = _run_cli("validate", "contract.fluid.yaml", "--offline", cwd=tmp_path)
+        assert result.returncode == 0, result.stdout + result.stderr
+        out = " ".join(result.stdout.split())
+        assert "builds[0] (id 'transform'): trigger.type continuous on engine 'dbt'" in out
+
+    def test_strict_fails(self, tmp_path: Path) -> None:
+        self._write(tmp_path)
+        result = _run_cli("validate", "contract.fluid.yaml", "--offline", "--strict", cwd=tmp_path)
+        assert result.returncode == 1, result.stdout + result.stderr
+        assert "warnings treated as errors (--strict)" in " ".join(result.stdout.split())
+
+    def test_json_lists_the_warning(self, tmp_path: Path) -> None:
+        self._write(tmp_path)
+        result = _run_cli(
+            "validate", "contract.fluid.yaml", "--offline", "--format", "json", cwd=tmp_path
+        )
+        assert result.returncode == 0, result.stdout + result.stderr
+        report = json.loads(result.stdout)
+        assert report["valid"] is True
+        assert [w for w in report["warnings"] if "trigger.type continuous" in w]
+
+
+class TestRunnerRefusal:
+    def test_the_dbt_runner_names_the_trigger(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture
+    ) -> None:
+        from fluid_build.build_runners.dbt import runner
+
+        def _not_reached(*_args: Any, **_kwargs: Any) -> Any:
+            raise AssertionError("a refused build must not prepare a dbt profile")
+
+        monkeypatch.setattr(runner, "resolve_dbt_profiles_dir", _not_reached)
+        assert runner.execute_dbt_build(_build("dbt"), tmp_path, tmp_path) == 1
+        out = " ".join(capsys.readouterr().out.split())
+        assert "trigger.type continuous: forge-cli runs dbt builds as batch runs" in out
+        assert "Unknown trigger type" not in out
+
+    def test_the_python_runner_names_the_trigger(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture
+    ) -> None:
+        from fluid_build.build_runners.python import runner
+
+        def _not_run(*_args: Any, **_kwargs: Any) -> Any:
+            raise AssertionError("a refused build must not run its script")
+
+        script = tmp_path / "ingest.py"
+        script.write_text("print('ran')\n", "utf-8")
+        monkeypatch.setattr(runner.subprocess, "run", _not_run)
+        assert runner.execute_build(_build("python"), script, tmp_path) == 1
+        out = " ".join(capsys.readouterr().out.split())
+        assert "trigger.type continuous: forge-cli runs a build's Python script" in out
+        assert "Unknown trigger type" not in out
+
+    def test_another_unknown_trigger_keeps_its_message(
+        self, tmp_path: Path, capsys: pytest.CaptureFixture
+    ) -> None:
+        from fluid_build.build_runners.python import runner
+
+        script = tmp_path / "ingest.py"
+        script.write_text("print('ran')\n", "utf-8")
+        assert runner.execute_build(_build("python", trigger="event"), script, tmp_path) == 1
+        assert "Unknown trigger type: event" in capsys.readouterr().out
+
+
+# ── scaffold-composer: a DAG only for builds a scheduler starts ──────────────
+
+
+class TestScaffoldComposer:
+    @staticmethod
+    def _run(
+        tmp_path: Path, *builds: Dict[str, Any], out: str = "dags", **top: Any
+    ) -> "tuple[int, Path]":
+        import logging
+        from types import SimpleNamespace
+
+        from fluid_build.cli.scaffold_composer import run
+
+        contract = tmp_path / "contract.fluid.yaml"
+        doc = {"id": "net.instances", "builds": list(builds), **top}
+        contract.write_text(yaml.safe_dump(doc), "utf-8")
+        out_dir = tmp_path / out
+        args = SimpleNamespace(contract=str(contract), env=None, out_dir=str(out_dir))
+        return run(args, logging.getLogger("test.scaffold_composer")), out_dir
+
+    def test_nothing_to_schedule_writes_no_dag(
+        self, tmp_path: Path, capsys: pytest.CaptureFixture
+    ) -> None:
+        flink = {"id": "j", "engine": "flink", "execution": {"trigger": {"cron": "0 5 * * *"}}}
+        rc, out_dir = self._run(tmp_path, _connector({"type": "continuous"}), flink)
+        assert rc == 0
+        assert not out_dir.exists()
+        out = " ".join(capsys.readouterr().out.split())
+        assert "continuously (trigger.type: continuous), so there is nothing to schedule." in out
+        assert "No DAG written." in out
+
+    @pytest.mark.parametrize("out", ["dags", "dags [/]:id:"])
+    def test_a_dag_it_wrote_before_is_named_and_kept(
+        self, tmp_path: Path, capsys: pytest.CaptureFixture, out: str
+    ) -> None:
+        scheduled = {"id": "t", "engine": "dbt", "execution": {"trigger": {"cron": "0 3 * * *"}}}
+        self._run(tmp_path, scheduled, out=out)
+        (dag,) = (tmp_path / out).glob("*.py")
+        written = dag.read_text("utf-8")
+        capsys.readouterr()
+        rc, _ = self._run(tmp_path, _connector({"type": "continuous"}), out=out)
+        assert rc == 0
+        assert dag.read_text("utf-8") == written
+        # As written and whole: no markup or emoji rendered, no fold to break a copy.
+        assert f"is still at {dag}: delete it" in capsys.readouterr().out
+
+    @pytest.mark.parametrize(
+        "content, named",
+        [
+            # The template before #215 quoted the dag_id with double quotes.
+            ('    dag_id="net_instances",\n    tags=["FLUID"]\n', True),
+            ("# hand-written by another team\n", False),
+            ('    dag_id="other_product",\n    tags=["FLUID"]\n', False),
+        ],
+    )
+    def test_only_a_file_it_wrote_is_named(
+        self, tmp_path: Path, capsys: pytest.CaptureFixture, content: str, named: bool
+    ) -> None:
+        existing = tmp_path / "dags" / "net_instances.py"
+        existing.parent.mkdir()
+        existing.write_text(content, "utf-8")
+        rc, _ = self._run(tmp_path, _connector({"type": "continuous"}))
+        assert rc == 0
+        assert existing.read_text("utf-8") == content
+        assert ("is still at" in capsys.readouterr().out) is named
+
+    def test_the_cron_comes_from_a_build_a_scheduler_starts(self, tmp_path: Path) -> None:
+        # The Flink job's cron is its platform's business, not this DAG's.
+        rc, out_dir = self._run(
+            tmp_path,
+            {"id": "j", "engine": "flink", "execution": {"trigger": {"cron": "0 5 * * *"}}},
+            _connector({"type": "continuous"}),
+            {"id": "t", "engine": "dbt", "execution": {"trigger": {"cron": "0 3 * * *"}}},
+        )
+        assert rc == 0
+        (dag,) = out_dir.glob("*.py")
+        assert "schedule='0 3 * * *'" in dag.read_text("utf-8")
+
+    def test_hand_declared_tasks_keep_the_dag(self, tmp_path: Path) -> None:
+        # As fluid generate schedule does: the contract declares what to run.
+        tasks = {"tasks": [{"taskId": "refresh", "action": "dbt_run"}]}
+        rc, out_dir = self._run(tmp_path, _connector({"type": "continuous"}), orchestration=tasks)
+        assert rc == 0
+        (dag,) = out_dir.glob("*.py")
+        assert "schedule='0 2 * * *'" in dag.read_text("utf-8")
+
+    def test_a_contract_with_no_builds_keeps_its_nightly_dag(self, tmp_path: Path) -> None:
+        rc, out_dir = self._run(tmp_path)
+        assert rc == 0
+        (dag,) = out_dir.glob("*.py")
+        assert "schedule='0 2 * * *'" in dag.read_text("utf-8")