From a945889ed00ca0e82a038516db120b98c67439d8 Mon Sep 17 00:00:00 2001
From: Speculator55005 <50082482+fas89@users.noreply.github.com>
Date: Wed, 21 Oct 2026 20:31:05 +0200
Subject: [PATCH 01/11] feat(validate): warn on trigger.type continuous on a
 build forge-cli runs as a batch run

A continuous trigger (schema 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
duckdb, airbyte, meltano and dlt acquisition builds to completion: the dbt and
python runners refuse the trigger, so such a build never runs, and the others
run once per fluid apply. fluid validate now says so as a warning, so plain
validate passes and --strict fails:

  builds[0] (id 'transform'): trigger.type continuous on engine 'dbt':
  forge-cli runs dbt builds as batch runs, which end, and its dbt runner
  refuses a continuous trigger: forge-cli never runs this build. Use
  trigger.type schedule, or run the job on a platform that keeps it running.

The check follows run_builds_from_args' dispatch order, so it names the runner
that gets the build: a dbt build with inline SQL runs on DuckDB. The Kafka
Connect and Debezium connectors, whose runners declare
RunnerCapability.STREAMING, and builds that run outside forge-cli get no
warning, and neither do spark, glue and custom builds: forge-cli has no runner
of its own for them, and their platform may keep them running. A crash in the
check is a warning that says the contract was not checked for it.

Borrowed: Spark's UnsupportedOperationChecker.checkForContinuous checks a
continuous trigger against what the query can run before it starts, and
Databricks refuses Trigger.Continuous where it cannot keep a job running
(INFINITE_STREAMING_TRIGGER_NOT_SUPPORTED), naming the trigger.
---
 fluid_build/build_runners/base.py |  96 ++++++++++++++++-
 fluid_build/cli/validate.py       |  18 ++++
 tests/test_continuous_trigger.py  | 173 ++++++++++++++++++++++++++++++
 3 files changed, 286 insertions(+), 1 deletion(-)

diff --git a/fluid_build/build_runners/base.py b/fluid_build/build_runners/base.py
index c0d3dd04..d20ba1c5 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")
 
@@ -205,6 +209,96 @@ def is_acquisition_build(build: Dict[str, Any]) -> bool:
     return engine in ACQUISITION_ENGINES
 
 
+#: 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"})
+
+
+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 once, with ``refusing_runner`` None. The checks follow the
+    dispatcher's order, so they name the runner that gets the build. A build
+    forge-cli has no runner of its own for (spark, glue, custom, or no engine)
+    is None: it can be a streaming job its own platform keeps running.
+    """
+    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):
+        return "inline-SQL builds", None
+    if is_dbt_build(build):
+        return "dbt builds", "dbt"
+    if name == "python":
+        return "python builds", "python"
+    return None
+
+
+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, and an inline-SQL build or a duckdb, airbyte, meltano or
+    dlt acquisition runs once per ``fluid apply``. A warning, not an error:
+    plain ``fluid validate`` passes, and ``--strict`` fails.
+
+    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 once per fluid apply, and no scheduler starts it"
+        warnings.append(
+            f"{where}: trigger.type continuous on {on}: forge-cli runs {runs} as batch runs, "
+            f"which end{outcome}. Use trigger.type schedule, or run the job on a platform that "
+            "keeps it running."
+        )
+    return warnings
+
+
 # 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
diff --git a/fluid_build/cli/validate.py b/fluid_build/cli/validate.py
index 2fc56268..24c482fa 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 the others run once
+        # per fluid apply. 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/tests/test_continuous_trigger.py b/tests/test_continuous_trigger.py
index 2d5e5cc2..9fa8e574 100644
--- a/tests/test_continuous_trigger.py
+++ b/tests/test_continuous_trigger.py
@@ -236,3 +236,176 @@ 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 run the job on a platform that keeps it running."
+
+
+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 _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_a_duckdb_acquisition_runs_once_per_apply(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 once per fluid apply, and "
+            f"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"),
+            pytest.param(_build("sql", properties=_SQL), "inline-SQL builds", id="sql"),
+            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")
+            ],
+        ],
+    )
+    def test_every_batch_runner_is_warned(self, build: Dict[str, Any], says: str) -> None:
+        (warning,) = _warnings({"builds": [build]})
+        assert says in warning
+        assert warning.endswith(_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"
+            ),
+            # No runner of its own: a Spark or Glue streaming job can run on
+            # its own platform.
+            pytest.param(_build("spark"), id="spark"),
+            pytest.param(_build("glue"), id="glue"),
+            pytest.param(_build("custom"), id="custom"),
+            pytest.param(_build(), id="no-engine-no-sql"),
+            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_build_without_an_id_is_named_by_index(self) -> None:
+        build = _build("python")
+        del build["id"]
+        (warning,) = _warnings({"builds": [_build("spark"), 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
+    }
+
+
+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 result.stdout
+
+    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]

From 9d4e910771c93e1d16fc2c2b3566dcd33df512d7 Mon Sep 17 00:00:00 2001
From: Speculator55005 <50082482+fas89@users.noreply.github.com>
Date: Wed, 21 Oct 2026 20:31:11 +0200
Subject: [PATCH 02/11] fix(build-runners): the dbt and python runners refuse
 trigger.type continuous by name

They printed "Unknown trigger type: continuous" and "Supported types: manual,
schedule", as for a typo. They now say that forge-cli runs the build as a
batch run, which ends, while a continuous trigger says the job's platform
keeps it running, and what to use instead. The exit code stays 1. The dbt
runner refuses before it resolves a profile or prepares the dbt command.
---
 fluid_build/build_runners/dbt/runner.py    | 16 +++++++++
 fluid_build/build_runners/python/runner.py | 16 +++++++++
 tests/test_continuous_trigger.py           | 42 ++++++++++++++++++++++
 3 files changed, 74 insertions(+)

diff --git a/fluid_build/build_runners/dbt/runner.py b/fluid_build/build_runners/dbt/runner.py
index ee9c0fd0..3e81af23 100644
--- a/fluid_build/build_runners/dbt/runner.py
+++ b/fluid_build/build_runners/dbt/runner.py
@@ -45,6 +45,7 @@
 
 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
@@ -939,6 +940,21 @@ 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;
+    # ``fluid validate`` warns about it (base.continuous_trigger_warnings).
+    if is_continuous_build(build):
+        cprint(
+            "   ❌ trigger.type continuous: forge-cli runs dbt builds as batch runs, which "
+            "end, and a continuous trigger says the job's platform keeps it running."
+        )
+        cprint(
+            "   Use trigger.type manual or schedule, or run the job on a platform that keeps "
+            "it running."
+        )
+        cprint(f"{'=' * 80}")
+        return 1
+
     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 fff454c9..bad8182c 100644
--- a/fluid_build/build_runners/python/runner.py
+++ b/fluid_build/build_runners/python/runner.py
@@ -34,6 +34,7 @@
 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
 
@@ -99,6 +100,21 @@ 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. ``fluid validate`` warns about it
+    # (base.continuous_trigger_warnings).
+    if is_continuous_build(build):
+        cprint(
+            "   ❌ trigger.type continuous: forge-cli runs a build's Python script as a batch "
+            "run, which ends, and a continuous trigger says the job's platform keeps it running."
+        )
+        cprint(
+            "   Use trigger.type manual or schedule, or run the job on a platform that keeps "
+            "it running."
+        )
+        cprint(f"{'=' * 80}")
+        return 1
+
     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/tests/test_continuous_trigger.py b/tests/test_continuous_trigger.py
index 9fa8e574..02bc39f1 100644
--- a/tests/test_continuous_trigger.py
+++ b/tests/test_continuous_trigger.py
@@ -409,3 +409,45 @@ def test_json_lists_the_warning(self, tmp_path: Path) -> None:
         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

From 458cc574a1cb92b4922c4a008c45a1811461a945 Mon Sep 17 00:00:00 2001
From: Speculator55005 <50082482+fas89@users.noreply.github.com>
Date: Wed, 21 Oct 2026 20:31:18 +0200
Subject: [PATCH 03/11] fix(scaffold-composer): no DAG for a contract no
 scheduler starts, and the cron of a scheduled build

fluid scaffold-composer wrote a DAG for any contract: one whose every build is
continuous or runs outside forge-cli got a nightly 0 2 * * * DAG, or the cron a
Flink job declares for its own platform. It now writes no DAG for such a
contract and says so, as fluid generate schedule does, and names a DAG it
wrote before, which it leaves in place. Otherwise the DAG takes the cron of
the first build a scheduler starts.

has_only_unscheduled_builds() moves to util/contract.py, so both commands
decide with one predicate.
---
 fluid_build/cli/generate_schedule.py | 12 +-----
 fluid_build/cli/scaffold_composer.py | 59 +++++++++++++++++++++------
 fluid_build/util/contract.py         |  8 ++++
 tests/test_continuous_trigger.py     | 61 ++++++++++++++++++++++++++++
 4 files changed, 117 insertions(+), 23 deletions(-)

diff --git a/fluid_build/cli/generate_schedule.py b/fluid_build/cli/generate_schedule.py
index e027a71c..a155d7e6 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 3ccae073..ae948c43 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,53 @@ 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 _report_nothing_to_schedule(dag_path: str) -> None:
+    """Say why no DAG is written, as ``fluid generate schedule`` does, and name
+    a DAG written for the product before: left in place, it keeps scheduling
+    the product wherever it was deployed."""
+    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 os.path.exists(dag_path):
+        cprint(
+            f"A DAG written for this product before is still at {dag_path}: delete it, and "
+            "remove it from Cloud Composer if you deployed it there.",
+            markup=False,
+        )
+
+
 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 +216,13 @@ 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, as `fluid generate
+        # schedule` says, and gets no DAG.
+        if has_only_unscheduled_builds(c):
+            _report_nothing_to_schedule(os.path.join(os.path.abspath(args.out_dir), f"{dag_id}.py"))
+            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/util/contract.py b/fluid_build/util/contract.py
index fd9922bf..aec0bc5a 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 02bc39f1..f7103a30 100644
--- a/tests/test_continuous_trigger.py
+++ b/tests/test_continuous_trigger.py
@@ -451,3 +451,64 @@ def test_another_unknown_trigger_keeps_its_message(
         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]) -> "tuple[int, Path]":
+        import logging
+        from types import SimpleNamespace
+
+        from fluid_build.cli.scaffold_composer import run
+
+        contract = tmp_path / "contract.fluid.yaml"
+        contract.write_text(
+            yaml.safe_dump({"id": "net.instances", "builds": list(builds)}), "utf-8"
+        )
+        out_dir = tmp_path / "dags"
+        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
+
+    def test_a_dag_written_before_is_named_and_kept(
+        self, tmp_path: Path, capsys: pytest.CaptureFixture
+    ) -> None:
+        stale = tmp_path / "dags" / "net_instances.py"
+        stale.parent.mkdir()
+        stale.write_text("# nightly\n", "utf-8")
+        rc, _ = self._run(tmp_path, _connector({"type": "continuous"}))
+        assert rc == 0
+        assert stale.read_text("utf-8") == "# nightly\n"
+        out = " ".join(capsys.readouterr().out.split())
+        assert "net_instances.py" in out and "delete it" in out
+
+    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_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")

From 0aaea9f82ada42b62e98d708599cab28de631d31 Mon Sep 17 00:00:00 2001
From: Speculator55005 <50082482+fas89@users.noreply.github.com>
Date: Wed, 21 Oct 2026 20:31:18 +0200
Subject: [PATCH 04/11] docs(changelog): a continuous trigger on a batch build
 warns; scaffold-composer schedules only scheduled builds

---
 CHANGELOG.md | 17 ++++++++++++++---
 1 file changed, 14 insertions(+), 3 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 543d2ea5..843603a3 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -46,6 +46,10 @@ 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, where it
+  used to write one. A DAG it wrote for such a contract before stays in `--out-dir`, which
+  the command names, and in Cloud Composer if you deployed it: delete both.
 - 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 +65,22 @@ 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, and an inline-SQL build or a `duckdb`, `airbyte`,
+  `meltano` or `dlt` acquisition runs once per `fluid apply`. A `spark`, `glue` or `custom`
+  build gets no warning: forge-cli has no runner of its own for it, and its platform may
+  keep it running.
 
 ### 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, as `fluid generate schedule` writes
+  no files for one.
 
 ### Fixed
 

From c474547ae58968688f5667a92dda080960b294bb Mon Sep 17 00:00:00 2001
From: Speculator55005 <50082482+fas89@users.noreply.github.com>
Date: Wed, 21 Oct 2026 20:34:50 +0200
Subject: [PATCH 05/11] fix(scaffold-composer): print the old DAG's path whole,
 so it copies

Rich folded a path longer than the terminal across lines, so the path of a DAG
written before came out broken where it is meant to be copied. The line is
printed with soft_wrap, and the terminal wraps it.
---
 fluid_build/cli/scaffold_composer.py | 3 +++
 tests/test_continuous_trigger.py     | 4 ++--
 2 files changed, 5 insertions(+), 2 deletions(-)

diff --git a/fluid_build/cli/scaffold_composer.py b/fluid_build/cli/scaffold_composer.py
index ae948c43..bd8048eb 100644
--- a/fluid_build/cli/scaffold_composer.py
+++ b/fluid_build/cli/scaffold_composer.py
@@ -197,10 +197,13 @@ def _report_nothing_to_schedule(dag_path: str) -> None:
         "(trigger.type: continuous), so there is nothing to schedule. No DAG written."
     )
     if os.path.exists(dag_path):
+        # soft_wrap: the terminal wraps the line, so the path is never broken
+        # by an inserted newline and copies whole.
         cprint(
             f"A DAG written for this product before is still at {dag_path}: delete it, and "
             "remove it from Cloud Composer if you deployed it there.",
             markup=False,
+            soft_wrap=True,
         )
 
 
diff --git a/tests/test_continuous_trigger.py b/tests/test_continuous_trigger.py
index f7103a30..3c9c8d8f 100644
--- a/tests/test_continuous_trigger.py
+++ b/tests/test_continuous_trigger.py
@@ -492,8 +492,8 @@ def test_a_dag_written_before_is_named_and_kept(
         rc, _ = self._run(tmp_path, _connector({"type": "continuous"}))
         assert rc == 0
         assert stale.read_text("utf-8") == "# nightly\n"
-        out = " ".join(capsys.readouterr().out.split())
-        assert "net_instances.py" in out and "delete it" in out
+        # Whole, as printed: a path folded across lines does not copy.
+        assert f"is still at {stale}: delete it" in capsys.readouterr().out
 
     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.

From c2aa246d94a1f889160f5aac82ba361a876afb3b Mon Sep 17 00:00:00 2001
From: Speculator55005 <50082482+fas89@users.noreply.github.com>
Date: Wed, 21 Oct 2026 21:21:55 +0200
Subject: [PATCH 06/11] fix(build-runners): fluid validate and fluid apply
 agree on a continuous build

A defect scan found the warning and the dispatcher disagreeing. A continuous
spark, glue or custom build, or an ingestion engine declared without
pattern: acquisition, got no warning, while fluid apply --mode amend-and-build
handed it to the python runner, which refused it. Inline SQL on a platform
with no embedded-SQL executor, such as databricks, was warned as running once
per apply, while the executor refuses that platform.

Now one predicate decides both. A continuous build that no forge-cli runner
takes is left to its platform, as a Flink job is: the dispatcher prints that
there is nothing for forge-cli to run, counts it under "Run outside forge-cli"
and records it as skipped with reason runs_outside_forge, and validate gives
no warning. A build with no engine goes to the python runner, so it is warned
like a python build. EMBEDDED_SQL_PLATFORMS is the one list of the platforms
that run inline SQL, read by the executor and by the warning. A test runs the
dispatcher over every engine and pattern and checks that validate warns
exactly when a batch runner gets the build.

The dbt and python runners print one refusal, refuse_continuous_trigger, and
it gives the same remedy as the warning ("Use trigger.type schedule or
manual, ..."). The run-once warning says the build runs only when fluid apply
runs builds: plain fluid apply does not run an acquisition. The --strict
output assertion no longer depends on the terminal width.
---
 fluid_build/build_runners/base.py          | 110 ++++++++++++++++---
 fluid_build/build_runners/dbt/runner.py    |  16 +--
 fluid_build/build_runners/python/runner.py |  15 +--
 fluid_build/cli/validate.py                |   4 +-
 tests/test_continuous_trigger.py           | 117 +++++++++++++++++++--
 5 files changed, 213 insertions(+), 49 deletions(-)

diff --git a/fluid_build/build_runners/base.py b/fluid_build/build_runners/base.py
index d20ba1c5..4e870886 100644
--- a/fluid_build/build_runners/base.py
+++ b/fluid_build/build_runners/base.py
@@ -146,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.
@@ -209,6 +213,12 @@ 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
@@ -223,10 +233,14 @@ def _batch_runner_for(build: Dict[str, Any]) -> Optional[Tuple[str, Optional[str
     ``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 once, with ``refusing_runner`` None. The checks follow the
-    dispatcher's order, so they name the runner that gets the build. A build
-    forge-cli has no runner of its own for (spark, glue, custom, or no engine)
-    is None: it can be a streaming job its own platform keeps running.
+    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; a build with no engine goes to the python runner. None for a
+    connector, which stays running, and for a build no forge-cli runner takes:
+    a spark, glue or custom engine, an ingestion engine without ``pattern:
+    acquisition``, or inline SQL on a platform with no embedded-SQL executor.
+    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
@@ -243,11 +257,13 @@ def _batch_runner_for(build: Dict[str, Any]) -> Optional[Tuple[str, Optional[str
     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 == "python":
-        return "python builds", "python"
+    if name in ("python", ""):
+        return ("python builds" if name else "builds with no engine"), "python"
     return None
 
 
@@ -258,9 +274,12 @@ def continuous_trigger_warnings(contract: Dict[str, Any]) -> List[str]:
     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, and an inline-SQL build or a duckdb, airbyte, meltano or
-    dlt acquisition runs once per ``fluid apply``. A warning, not an error:
-    plain ``fluid validate`` passes, and ``--strict`` fails.
+    runs such a build (or one with no engine, which the python runner takes),
+    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
+    no forge-cli runner takes 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
@@ -290,15 +309,73 @@ def continuous_trigger_warnings(contract: Dict[str, Any]) -> List[str]:
                 "runs this build"
             )
         else:
-            outcome = ": this build runs once per fluid apply, and no scheduler starts it"
+            outcome = (
+                ": this build runs only when fluid apply runs builds, and no scheduler starts it"
+            )
         warnings.append(
             f"{where}: trigger.type continuous on {on}: forge-cli runs {runs} as batch runs, "
-            f"which end{outcome}. Use trigger.type schedule, or run the job on a platform that "
-            "keeps it running."
+            f"which end{outcome}. {CONTINUOUS_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 no forge-cli runner takes.
+
+    That is a spark, glue or custom engine, an ingestion engine declared
+    without ``pattern: acquisition``, or inline SQL on a platform with no
+    embedded-SQL executor (:func:`_batch_runner_for` is None for each). Its
+    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 no forge-cli runner takes 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
@@ -746,13 +823,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",
@@ -1302,6 +1379,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 3e81af23..bad6a915 100644
--- a/fluid_build/build_runners/dbt/runner.py
+++ b/fluid_build/build_runners/dbt/runner.py
@@ -48,7 +48,7 @@
 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,
@@ -941,19 +941,9 @@ def execute_dbt_build(
     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;
-    # ``fluid validate`` warns about it (base.continuous_trigger_warnings).
+    # running, and a dbt run ends: refused before a profile is prepared.
     if is_continuous_build(build):
-        cprint(
-            "   ❌ trigger.type continuous: forge-cli runs dbt builds as batch runs, which "
-            "end, and a continuous trigger says the job's platform keeps it running."
-        )
-        cprint(
-            "   Use trigger.type manual or schedule, or run the job on a platform that keeps "
-            "it running."
-        )
-        cprint(f"{'=' * 80}")
-        return 1
+        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 bad8182c..aee8e8f7 100644
--- a/fluid_build/build_runners/python/runner.py
+++ b/fluid_build/build_runners/python/runner.py
@@ -37,6 +37,7 @@
 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")
 
@@ -101,19 +102,9 @@ def execute_build(
     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. ``fluid validate`` warns about it
-    # (base.continuous_trigger_warnings).
+    # running, and the script's run ends.
     if is_continuous_build(build):
-        cprint(
-            "   ❌ trigger.type continuous: forge-cli runs a build's Python script as a batch "
-            "run, which ends, and a continuous trigger says the job's platform keeps it running."
-        )
-        cprint(
-            "   Use trigger.type manual or schedule, or run the job on a platform that keeps "
-            "it running."
-        )
-        cprint(f"{'=' * 80}")
-        return 1
+        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)
diff --git a/fluid_build/cli/validate.py b/fluid_build/cli/validate.py
index 24c482fa..254427f8 100644
--- a/fluid_build/cli/validate.py
+++ b/fluid_build/cli/validate.py
@@ -903,8 +903,8 @@ def _run_contract_rules(
         # 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 the others run once
-        # per fluid apply. A warning, not an error: plain validate passes and
+        # 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
diff --git a/tests/test_continuous_trigger.py b/tests/test_continuous_trigger.py
index 3c9c8d8f..b64d5e42 100644
--- a/tests/test_continuous_trigger.py
+++ b/tests/test_continuous_trigger.py
@@ -241,7 +241,7 @@ def test_generate_schedule_has_nothing_to_schedule(self, tmp_path: Path) -> None
 # ── a continuous trigger on a build forge-cli runs as a batch run ──────────
 
 _SQL = {"sql": "SELECT 1"}
-_REMEDY = "Use trigger.type schedule, or run the job on a platform that keeps it running."
+_REMEDY = "Use trigger.type schedule or manual, or run the job on a platform that keeps it running."
 
 
 def _build(engine: Any = None, trigger: str = "continuous", **extra: Any) -> Dict[str, Any]:
@@ -251,6 +251,10 @@ def _build(engine: Any = None, trigger: str = "continuous", **extra: Any) -> Dic
     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
 
@@ -269,11 +273,11 @@ def test_a_dbt_build_is_never_run(self) -> None:
             f"forge-cli never runs this build. {_REMEDY}"
         ]
 
-    def test_a_duckdb_acquisition_runs_once_per_apply(self) -> None:
+    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 once per fluid apply, and "
-            f"no scheduler starts it. {_REMEDY}"
+            "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(
@@ -281,7 +285,19 @@ def test_a_duckdb_acquisition_runs_once_per_apply(self) -> None:
         [
             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",
@@ -313,12 +329,16 @@ def test_every_batch_runner_is_warned(self, build: Dict[str, Any], says: str) ->
             pytest.param(
                 _build("sql", properties={"language": "flink_sql", **_SQL}), id="flink-sql"
             ),
-            # No runner of its own: a Spark or Glue streaming job can run on
-            # its own platform.
+            # No forge-cli runner takes these: fluid apply leaves them to their
+            # platform, as it does a Flink job.
             pytest.param(_build("spark"), id="spark"),
             pytest.param(_build("glue"), id="glue"),
             pytest.param(_build("custom"), id="custom"),
-            pytest.param(_build(), id="no-engine-no-sql"),
+            pytest.param(_build("kafka-connect"), id="connector-without-acquisition"),
+            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"),
         ],
@@ -372,6 +392,87 @@ def test_streaming_engines_are_the_runners_that_declare_streaming() -> None:
     }
 
 
+_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("flink", properties=_SQL), id="flink"),
+    pytest.param(_build("sql", properties={"language": "flink_sql", **_SQL}), id="flink-sql"),
+]
+
+
+@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.
+    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
+    connector = called == ["acquisition"] and build["engine"] in base.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())
+
+
 class TestValidateCli:
     @staticmethod
     def _write(tmp_path: Path) -> None:
@@ -398,7 +499,7 @@ 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 result.stdout
+        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)

From 2dfd33de8266faff58430ff44cd6b6ee4a1d35db Mon Sep 17 00:00:00 2001
From: Speculator55005 <50082482+fas89@users.noreply.github.com>
Date: Wed, 21 Oct 2026 21:22:03 +0200
Subject: [PATCH 07/11] fix(scaffold-composer): name only a DAG it wrote, and
 keep the DAG for declared tasks

The no-DAG message told the operator to delete any file at <out-dir>/<dag_id>.py,
and dag_id is a lossy clean-up of the contract id, so it could name another
team's DAG. It now names the file only when it carries the template's
tags=["FLUID"] and this dag_id, as the current template quotes it or as the
template before #215 did; a file it did not write is not mentioned.

A contract whose every build is unscheduled but which hand-declares
orchestration.tasks still gets its DAG, as fluid generate schedule still
writes files for one. The path prints with emoji=False beside markup=False, so
an :emoji: code in --out-dir prints as written.
---
 fluid_build/cli/scaffold_composer.py | 47 ++++++++++++++++------
 tests/test_continuous_trigger.py     | 58 +++++++++++++++++++++-------
 2 files changed, 80 insertions(+), 25 deletions(-)

diff --git a/fluid_build/cli/scaffold_composer.py b/fluid_build/cli/scaffold_composer.py
index bd8048eb..a7553e9f 100644
--- a/fluid_build/cli/scaffold_composer.py
+++ b/fluid_build/cli/scaffold_composer.py
@@ -188,21 +188,40 @@ def _extract_cron(c: dict) -> object:
     return "0 2 * * *"
 
 
-def _report_nothing_to_schedule(dag_path: str) -> None:
+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 written for the product before: left in place, it keeps scheduling
-    the product wherever it was deployed."""
+    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 os.path.exists(dag_path):
-        # soft_wrap: the terminal wraps the line, so the path is never broken
-        # by an inserted newline and copies whole.
+    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 written for this product before is still at {dag_path}: delete it, and "
-            "remove it from Cloud Composer if you deployed it there.",
+            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,
         )
 
@@ -220,10 +239,14 @@ def run(args, logger: logging.Logger) -> int:
         # 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, as `fluid generate
-        # schedule` says, and gets no DAG.
-        if has_only_unscheduled_builds(c):
-            _report_nothing_to_schedule(os.path.join(os.path.abspath(args.out_dir), f"{dag_id}.py"))
+        # 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
diff --git a/tests/test_continuous_trigger.py b/tests/test_continuous_trigger.py
index b64d5e42..78ad5e08 100644
--- a/tests/test_continuous_trigger.py
+++ b/tests/test_continuous_trigger.py
@@ -559,17 +559,18 @@ def test_another_unknown_trigger_keeps_its_message(
 
 class TestScaffoldComposer:
     @staticmethod
-    def _run(tmp_path: Path, *builds: Dict[str, Any]) -> "tuple[int, Path]":
+    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"
-        contract.write_text(
-            yaml.safe_dump({"id": "net.instances", "builds": list(builds)}), "utf-8"
-        )
-        out_dir = tmp_path / "dags"
+        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
 
@@ -584,17 +585,40 @@ def test_nothing_to_schedule_writes_no_dag(
         assert "continuously (trigger.type: continuous), so there is nothing to schedule." in out
         assert "No DAG written." in out
 
-    def test_a_dag_written_before_is_named_and_kept(
-        self, tmp_path: Path, capsys: pytest.CaptureFixture
+    @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:
-        stale = tmp_path / "dags" / "net_instances.py"
-        stale.parent.mkdir()
-        stale.write_text("# nightly\n", "utf-8")
+        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 stale.read_text("utf-8") == "# nightly\n"
-        # Whole, as printed: a path folded across lines does not copy.
-        assert f"is still at {stale}: delete it" in capsys.readouterr().out
+        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.
@@ -608,6 +632,14 @@ def test_the_cron_comes_from_a_build_a_scheduler_starts(self, tmp_path: Path) ->
         (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

From 34e2a82e50970ea601b3cebfc404a9814bd1c353 Mon Sep 17 00:00:00 2001
From: Speculator55005 <50082482+fas89@users.noreply.github.com>
Date: Wed, 21 Oct 2026 21:22:03 +0200
Subject: [PATCH 08/11] docs(changelog): validate and apply agree on continuous
 builds; scaffold-composer keeps DAGs for declared tasks

---
 CHANGELOG.md | 22 +++++++++++++---------
 1 file changed, 13 insertions(+), 9 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 843603a3..03867b48 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -47,9 +47,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   `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, where it
-  used to write one. A DAG it wrote for such a contract before stays in `--out-dir`, which
-  the command names, and in Cloud Composer if you deployed it: delete both.
+  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`, which the command names, and in Cloud Composer if
+  you deployed it: delete both.
 - 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.
@@ -67,10 +68,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   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: 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, and an inline-SQL build or a `duckdb`, `airbyte`,
-  `meltano` or `dlt` acquisition runs once per `fluid apply`. A `spark`, `glue` or `custom`
-  build gets no warning: forge-cli has no runner of its own for it, and its platform may
-  keep it running.
+  runners refuse the trigger and say why (the python runner also takes a build with no
+  engine), and an inline-SQL build or a `duckdb`, `airbyte`, `meltano` or `dlt`
+  acquisition runs only when `fluid apply` runs builds. A continuous build that no
+  forge-cli runner takes, such as a `spark`, `glue` or `custom` one, or inline SQL on a
+  platform with no embedded-SQL executor, gets no warning: `fluid apply --mode
+  amend-and-build` leaves it to its platform, as it does a Flink job.
 
 ### Changed
 
@@ -79,8 +82,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   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, as `fluid generate schedule` writes
-  no files for one.
+  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
 

From dc9c37cbe1f08a777d613082da58cfc9a04baa20 Mon Sep 17 00:00:00 2001
From: Speculator55005 <50082482+fas89@users.noreply.github.com>
Date: Wed, 21 Oct 2026 22:03:21 +0200
Subject: [PATCH 09/11] fix(build-runners): the python runner keeps every build
 it runs, continuous or not

The previous commit left a continuous spark, glue or custom build, and an
ingestion engine declared without pattern: acquisition, to "its platform",
and fluid apply exited 0. But the python runner runs these builds' scripts
locally when the trigger is manual: a dlt pipeline.py or a local PySpark
script ran, and a connector without the acquisition pattern is never
deployed. Calling them outside forge-cli reported success with nothing
running.

The python runner keeps them again and refuses a continuous trigger by name,
and fluid validate now warns about each, naming an ingestion engine's
missing pattern. fluid apply leaves a continuous build to its platform only
where forge-cli does not run it: inline SQL on a platform with no
embedded-SQL executor, and a spark, glue or custom build that names a
runtime.platform other than local (PLATFORM_RUN_ENGINES). A test now checks
that the python runner receives those local scripts, which the parity test
alone could not catch: it passed when both sides were silent.
---
 fluid_build/build_runners/base.py | 53 +++++++++++-------
 tests/test_continuous_trigger.py  | 89 ++++++++++++++++++++++++-------
 2 files changed, 105 insertions(+), 37 deletions(-)

diff --git a/fluid_build/build_runners/base.py b/fluid_build/build_runners/base.py
index 4e870886..157d1cf9 100644
--- a/fluid_build/build_runners/base.py
+++ b/fluid_build/build_runners/base.py
@@ -225,6 +225,12 @@ def is_acquisition_build(build: Dict[str, Any]) -> bool:
 #: 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
@@ -235,11 +241,15 @@ def _batch_runner_for(build: Dict[str, Any]) -> Optional[Tuple[str, Optional[str
     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; a build with no engine goes to the python runner. None for a
-    connector, which stays running, and for a build no forge-cli runner takes:
-    a spark, glue or custom engine, an ingestion engine without ``pattern:
-    acquisition``, or inline SQL on a platform with no embedded-SQL executor.
-    The dispatcher leaves a continuous one of those to its platform
+    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")
@@ -262,9 +272,13 @@ def _batch_runner_for(build: Dict[str, Any]) -> Optional[Tuple[str, Optional[str
         return "inline-SQL builds", None
     if is_dbt_build(build):
         return "dbt builds", "dbt"
-    if name in ("python", ""):
-        return ("python builds" if name else "builds with no engine"), "python"
-    return None
+    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]:
@@ -274,12 +288,13 @@ def continuous_trigger_warnings(contract: Dict[str, Any]) -> List[str]:
     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 (or one with no engine, which the python runner takes),
-    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
-    no forge-cli runner takes gets none: its platform keeps it running, and
-    ``fluid apply`` leaves it there (:func:`_runs_outside_forge_continuously`).
+    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
@@ -336,11 +351,11 @@ def refuse_continuous_trigger(runs: str) -> int:
 
 
 def _runs_outside_forge_continuously(build: Dict[str, Any]) -> bool:
-    """True for a continuous build that no forge-cli runner takes.
+    """True for a continuous build that forge-cli does not run.
 
-    That is a spark, glue or custom engine, an ingestion engine declared
-    without ``pattern: acquisition``, or inline SQL on a platform with no
-    embedded-SQL executor (:func:`_batch_runner_for` is None for each). Its
+    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.
@@ -355,7 +370,7 @@ def _runs_outside_forge_continuously(build: Dict[str, Any]) -> bool:
 
 
 def _report_continuous_outside_forge(build: Dict[str, Any], build_id: str, report: Any) -> None:
-    """Say that a continuous build no forge-cli runner takes is not run here.
+    """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
diff --git a/tests/test_continuous_trigger.py b/tests/test_continuous_trigger.py
index 78ad5e08..7f8cc0b0 100644
--- a/tests/test_continuous_trigger.py
+++ b/tests/test_continuous_trigger.py
@@ -313,6 +313,29 @@ def test_only_fluid_apply_runs_a_duckdb_acquisition(self) -> None:
                 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:
@@ -329,12 +352,12 @@ def test_every_batch_runner_is_warned(self, build: Dict[str, Any], says: str) ->
             pytest.param(
                 _build("sql", properties={"language": "flink_sql", **_SQL}), id="flink-sql"
             ),
-            # No forge-cli runner takes these: fluid apply leaves them to their
-            # platform, as it does a Flink job.
-            pytest.param(_build("spark"), id="spark"),
-            pytest.param(_build("glue"), id="glue"),
-            pytest.param(_build("custom"), id="custom"),
-            pytest.param(_build("kafka-connect"), id="connector-without-acquisition"),
+            # 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",
@@ -368,7 +391,7 @@ def test_any_shape_is_read_without_raising(self, contract: Dict[str, Any]) -> No
     def test_a_build_without_an_id_is_named_by_index(self) -> None:
         build = _build("python")
         del build["id"]
-        (warning,) = _warnings({"builds": [_build("spark"), build]})
+        (warning,) = _warnings({"builds": [_build("flink"), build]})
         assert warning.startswith("builds[1]: trigger.type continuous on engine 'python':")
 
 
@@ -413,21 +436,17 @@ def test_streaming_engines_are_the_runners_that_declare_streaming() -> None:
     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"),
 ]
 
 
-@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.
+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
 
@@ -467,12 +486,46 @@ def _run(*_args: Any, **_kwargs: Any) -> int:
         args, logging.getLogger("test"), force_run=True, contract={"id": "p", "builds": [build]}
     )
     assert rc == 0
-    connector = called == ["acquisition"] and build["engine"] in base.STREAMING_ACQUISITION_ENGINES
+    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:

From 1c877168fc25a0533c834e1e8c4a90ffcce3fae1 Mon Sep 17 00:00:00 2001
From: Speculator55005 <50082482+fas89@users.noreply.github.com>
Date: Wed, 21 Oct 2026 22:03:21 +0200
Subject: [PATCH 10/11] docs(changelog): which continuous builds fluid apply
 leaves to their platform; when scaffold-composer names an old DAG

---
 CHANGELOG.md | 17 +++++++++--------
 1 file changed, 9 insertions(+), 8 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 03867b48..62208a1a 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -49,8 +49,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - `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`, which the command names, and in Cloud Composer if
-  you deployed it: delete both.
+  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.
@@ -68,12 +68,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   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: 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 also takes a build with no
-  engine), and an inline-SQL build or a `duckdb`, `airbyte`, `meltano` or `dlt`
-  acquisition runs only when `fluid apply` runs builds. A continuous build that no
-  forge-cli runner takes, such as a `spark`, `glue` or `custom` one, or inline SQL on a
-  platform with no embedded-SQL executor, gets no warning: `fluid apply --mode
-  amend-and-build` leaves it to its platform, as it does a Flink job.
+  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 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
 

From e336b2fa12a202999189f9628a0393b66b5029dc Mon Sep 17 00:00:00 2001
From: Speculator55005 <50082482+fas89@users.noreply.github.com>
Date: Wed, 21 Oct 2026 22:33:28 +0200
Subject: [PATCH 11/11] fix(validate): a continuous spark, glue or custom build
 is told to name its platform

The round-3 review found the warning did not say what leaves such a build to
its platform: execution.runtime.platform. Its remedy now ends "or run the job
on a platform that keeps it running, named in execution.runtime.platform."
The CHANGELOG says the platform exception holds for a build without inline
SQL: inline SQL is checked first, and Snowflake has an executor.
---
 CHANGELOG.md                      |  8 ++++----
 fluid_build/build_runners/base.py |  6 +++++-
 tests/test_continuous_trigger.py  | 10 +++++++++-
 3 files changed, 18 insertions(+), 6 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 62208a1a..76fdef28 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -71,10 +71,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   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 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.
+  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
 
diff --git a/fluid_build/build_runners/base.py b/fluid_build/build_runners/base.py
index 157d1cf9..fbbc3b38 100644
--- a/fluid_build/build_runners/base.py
+++ b/fluid_build/build_runners/base.py
@@ -327,9 +327,13 @@ def continuous_trigger_warnings(contract: Dict[str, Any]) -> List[str]:
             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}. {CONTINUOUS_REMEDY}"
+            f"which end{outcome}. {remedy}"
         )
     return warnings
 
diff --git a/tests/test_continuous_trigger.py b/tests/test_continuous_trigger.py
index 7f8cc0b0..5e1b603a 100644
--- a/tests/test_continuous_trigger.py
+++ b/tests/test_continuous_trigger.py
@@ -242,6 +242,8 @@ def test_generate_schedule_has_nothing_to_schedule(self, tmp_path: Path) -> None
 
 _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]:
@@ -341,7 +343,8 @@ def test_only_fluid_apply_runs_a_duckdb_acquisition(self) -> None:
     def test_every_batch_runner_is_warned(self, build: Dict[str, Any], says: str) -> None:
         (warning,) = _warnings({"builds": [build]})
         assert says in warning
-        assert warning.endswith(_REMEDY)
+        platform_run = build.get("engine") in ("spark", "glue", "custom")
+        assert warning.endswith(_PLATFORM_REMEDY if platform_run else _REMEDY)
 
     @pytest.mark.parametrize(
         "build",
@@ -388,6 +391,11 @@ def test_any_shape_is_read_without_raising(self, contract: Dict[str, Any]) -> No
         # 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"]