From 265c4c9d21b380c41f00e08491462d0a5b0d9355 Mon Sep 17 00:00:00 2001 From: Speculator55005 <50082482+fas89@users.noreply.github.com> Date: Wed, 21 Oct 2026 10:06:48 +0200 Subject: [PATCH 01/12] feat(validate): a schema error from a not-required rule names the field, or the rule's description A `not: {required: [field]}` rule quoted the whole object and the subschema it must not match: "{'type': 'continuous', 'cron': '0 2 * * *'} should not be valid under {'required': ['cron']}". It now reads " is not allowed here", as it already did under sink.catalogAuth, or the rule's own `description` when it carries one, so a schema can say why and what to do instead. --- fluid_build/schema_manager.py | 15 +++++++++++---- 1 file changed, 11 insertions(+), 4 deletions(-) diff --git a/fluid_build/schema_manager.py b/fluid_build/schema_manager.py index 173c14418..4f3a483eb 100644 --- a/fluid_build/schema_manager.py +++ b/fluid_build/schema_manager.py @@ -72,17 +72,24 @@ def format_schema_error(error: Any) -> str: Every path that reports a FLUID-contract schema error formats it here (``fluid validate`` on a contract and on a bundle), so a secret pasted in place of a ``sink.catalogAuth`` variable name is not printed or saved. + + A ``not: {required: [field]}`` rule, a field the object's ``type`` does not + take, reads as `` is not allowed here``, or as the rule's own + ``description`` when it carries one: jsonschema's message would quote the + whole object and the subschema it must not match. """ + value = error.validator_value + if error.validator == "not" and isinstance(value, dict) and list(value) == ["required"]: + why = error.schema.get("description") if isinstance(error.schema, dict) else None + if isinstance(why, str) and why: + return why + return f"{', '.join(map(str, value['required']))} is not allowed here" if not any(part in _UNQUOTED_VALUE_PATHS for part in error.absolute_path): return str(error.message) if error.validator in _KEY_ONLY_VALIDATORS: return str(error.message) if error.validator == "pattern": return f"does not match {error.validator_value!r} (value not shown)" - value = error.validator_value - if error.validator == "not" and isinstance(value, dict) and list(value) == ["required"]: - # ``not: {required: [field]}``: a field the object's ``type`` does not take. - return f"{', '.join(map(str, value['required']))} is not allowed here" if error.validator == "not" and isinstance(value, dict) and list(value) == ["pattern"]: # ``not: {pattern: ...}``: a token-shaped variable name, or a ``${``. return f"must not match {value['pattern']!r} (value not shown)" From 60e38beadfb9fe9f0471e43e0959b98f463b682f Mon Sep 17 00:00:00 2001 From: Speculator55005 <50082482+fas89@users.noreply.github.com> Date: Wed, 21 Oct 2026 10:06:48 +0200 Subject: [PATCH 02/12] feat(schema): trigger.type continuous in the 0.7.6 preview, never scheduled A long-running Flink job, or a Kafka Connect or Debezium connector, had no trigger type: every one of them starts runs. `continuous` says the platform keeps the job running, so no scheduler starts it: scheduler synthesis (`fluid generate schedule`, the template-mode schedule) and the acquisition schedule-sync leave the build out, where a connector used to get a nightly 0 2 * * * task, and a contract with no other build has nothing to schedule. The `fluid apply` DAGs already render only schedule triggers. The schema refuses cron, schedule, datasets, event and timetable next to it, and each refusal names the field and the trigger type that takes it. 0.7.5 is unchanged. Borrowed: the name and the "no other trigger" rule from Databricks Jobs' continuous trigger (docs.databricks.com/jobs/triggers.html). Diverged on purpose from Airflow's @continuous (ContinuousTimetable), which starts batch runs back to back: for an always-on job that would loop a no-op DAG. --- fluid_build/cli/_acquisition_stage_ext.py | 5 + fluid_build/cli/generate_schedule.py | 23 +- fluid_build/schedulers/synthesis.py | 9 +- fluid_build/schemas/fluid-schema-0.7.6.json | 66 +++++- fluid_build/util/contract.py | 18 ++ tests/test_continuous_trigger.py | 226 ++++++++++++++++++++ 6 files changed, 330 insertions(+), 17 deletions(-) create mode 100644 tests/test_continuous_trigger.py diff --git a/fluid_build/cli/_acquisition_stage_ext.py b/fluid_build/cli/_acquisition_stage_ext.py index 208038539..81048e038 100644 --- a/fluid_build/cli/_acquisition_stage_ext.py +++ b/fluid_build/cli/_acquisition_stage_ext.py @@ -59,6 +59,7 @@ from fluid_build.build_runners._ids import ( validate_identifier as _validate_identifier, ) +from fluid_build.util.contract import is_continuous_build LOG = logging.getLogger("fluid.acquire.stage_ext") @@ -481,6 +482,10 @@ def schedule_sync_acquisition( out: List[ScheduleArtifact] = [] for build in acquisition_builds(contract): bid = _validate_identifier(build.get("id", ""), kind="build.id") + # A continuous connector keeps running on its own: a schedule beside it + # is refused by schema 0.7.6, and must not start a second copy here. + if is_continuous_build(build): + continue execution = build.get("execution", {}) or {} trigger = execution.get("trigger", {}) or {} schedule = trigger.get("schedule") diff --git a/fluid_build/cli/generate_schedule.py b/fluid_build/cli/generate_schedule.py index 8a72412fe..e027a71c9 100644 --- a/fluid_build/cli/generate_schedule.py +++ b/fluid_build/cli/generate_schedule.py @@ -50,12 +50,13 @@ SUBCOMMAND = "schedule" -def _only_externally_run_builds(contract: Dict[str, Any]) -> bool: - """True when the contract has builds and every one runs outside forge-cli.""" - from fluid_build.util.contract import is_externally_run_build +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_externally_run_build(b) for b in builds) + return bool(builds) and all(is_unscheduled_build(b) for b in builds) def register_subcommand(subparsers: argparse._SubParsersAction): @@ -165,16 +166,18 @@ def run(args: Any, logger: logging.Logger) -> int: # the operator names, so defaulting WHO renders it commits nothing. scheduler_name = fluid_apply.DEFAULT_ENGINE - # A contract whose every build runs on its own platform (a Flink job) - # 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_externally_run_builds(contract) and not fluid_apply.has_explicit_tasks(contract): + # A contract whose every build runs on its own platform (a Flink job, or + # 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): from fluid_build.schedulers import has_scheduler if not scheduler_name or has_scheduler(scheduler_name): cprint( - "Every build runs outside forge-cli (a Flink job, for example), so " - "there is nothing to schedule. No files written." + "Every build runs outside forge-cli (a Flink job, for example) or " + "continuously (trigger.type: continuous), so there is nothing to " + "schedule. No files written." ) return 0 diff --git a/fluid_build/schedulers/synthesis.py b/fluid_build/schedulers/synthesis.py index 3bd681f9d..739161190 100644 --- a/fluid_build/schedulers/synthesis.py +++ b/fluid_build/schedulers/synthesis.py @@ -25,7 +25,7 @@ import re from typing import Any, Dict, List, Optional -from fluid_build.util.contract import is_externally_run_build +from fluid_build.util.contract import is_unscheduled_build # Provider + engine → scheduler action mapping. # Keys are (provider, engine) tuples; values are the action string used @@ -114,12 +114,13 @@ def synthesize_orchestration_from_builds( The returned dict is suitable for injection into the contract before calling a scheduler's ``validate()`` and ``generate()`` methods. """ - # A Flink job runs continuously on its own platform: there is nothing to - # schedule for it, and a nightly task that "runs" it would only log its SQL. + # A Flink job, or any build whose trigger is continuous, runs on its own + # platform: there is nothing to schedule for it, and a nightly task that + # "runs" it would at best log its SQL and at worst start a second copy. builds: List[Dict[str, Any]] = [ b for b in contract.get("builds", []) or [] - if not (isinstance(b, dict) and is_externally_run_build(b)) + if not (isinstance(b, dict) and is_unscheduled_build(b)) ] if not builds: return {} diff --git a/fluid_build/schemas/fluid-schema-0.7.6.json b/fluid_build/schemas/fluid-schema-0.7.6.json index 3168d0746..eda67c969 100644 --- a/fluid_build/schemas/fluid-schema-0.7.6.json +++ b/fluid_build/schemas/fluid-schema-0.7.6.json @@ -4316,8 +4316,10 @@ "dependency", "dataset", "schedule_and_dataset", - "timetable" - ] + "timetable", + "continuous" + ], + "description": "NEW in v0.7.6: continuous, for a build that runs as a long-lived job on its own platform, such as an engine: flink job on the Flink Kubernetes Operator or AWS Managed Service for Apache Flink, or a kafka-connect or debezium connector. The platform keeps the job running, so no scheduler starts runs: fluid generate schedule, fluid schedule-sync and the fluid apply DAGs leave the build out. A continuous trigger takes no other trigger, so cron, schedule, datasets, event and timetable are refused next to it, as Databricks Jobs refuses a continuous trigger next to any other." }, "schedule": { "type": "string" @@ -4382,7 +4384,65 @@ "type": "object", "description": "Custom timetable configuration" } - } + }, + "allOf": [ + { + "if": { + "required": [ + "type" + ], + "properties": { + "type": { + "const": "continuous" + } + } + }, + "then": { + "allOf": [ + { + "description": "cron is not allowed with type continuous: a continuous job runs all the time, so its trigger takes no other trigger. Remove cron, or set type: schedule.", + "not": { + "required": [ + "cron" + ] + } + }, + { + "description": "schedule is not allowed with type continuous: a continuous job runs all the time, so its trigger takes no other trigger. Remove schedule, or set type: schedule.", + "not": { + "required": [ + "schedule" + ] + } + }, + { + "description": "datasets is not allowed with type continuous: a continuous job runs all the time, so its trigger takes no other trigger. Remove datasets, or set type: dataset.", + "not": { + "required": [ + "datasets" + ] + } + }, + { + "description": "event is not allowed with type continuous: a continuous job runs all the time, so its trigger takes no other trigger. Remove event, or set type: event.", + "not": { + "required": [ + "event" + ] + } + }, + { + "description": "timetable is not allowed with type continuous: a continuous job runs all the time, so its trigger takes no other trigger. Remove timetable, or set type: timetable.", + "not": { + "required": [ + "timetable" + ] + } + } + ] + } + } + ] }, "runtime": { "type": "object", diff --git a/fluid_build/util/contract.py b/fluid_build/util/contract.py index d3b30e0a0..fd9922bfd 100644 --- a/fluid_build/util/contract.py +++ b/fluid_build/util/contract.py @@ -158,6 +158,24 @@ def describe_externally_run_build(build: Mapping[str, Any]) -> str: return f"language '{props.get('language') if isinstance(props, Mapping) else None}'" +def is_continuous_build(build: Mapping[str, Any]) -> bool: + """True when the build's ``execution.trigger.type`` is ``continuous`` (new in + schema 0.7.6): its platform keeps the job running, so no scheduler starts + its runs. Named after Databricks Jobs' continuous trigger, not Airflow's + ``@continuous``, which starts batch runs back to back.""" + execution = build.get("execution") + trigger = execution.get("trigger") if isinstance(execution, Mapping) else None + kind = trigger.get("type") if isinstance(trigger, Mapping) else None + return isinstance(kind, str) and kind.strip().lower() == "continuous" + + +def is_unscheduled_build(build: Mapping[str, Any]) -> bool: + """True when no scheduler starts the build's runs: its job runs outside + forge-cli (:func:`is_externally_run_build`), or continuously + (:func:`is_continuous_build`).""" + return is_externally_run_build(build) or is_continuous_build(build) + + 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 new file mode 100644 index 000000000..0ab9125c2 --- /dev/null +++ b/tests/test_continuous_trigger.py @@ -0,0 +1,226 @@ +# Copyright 2024-2026 Agentics Transformation Ltd +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""``trigger.type: continuous`` (schema 0.7.6): a job its platform keeps running. + +A long-running Flink job, or a Kafka Connect or Debezium connector, has no +trigger type: schedule, event and the rest all start runs. ``continuous`` says +no scheduler starts this build, so every path that turns builds into scheduled +runs leaves it out, and the schema refuses another trigger next to it, as +Databricks Jobs refuses a continuous trigger next to any other. +""" + +from __future__ import annotations + +import copy +import json +import os +import subprocess +import sys +from pathlib import Path +from typing import Any, Dict + +import jsonschema +import pytest +import yaml + +from fluid_build.schema_manager import format_schema_error +from fluid_build.util.contract import is_continuous_build, is_unscheduled_build + +SCHEMAS = Path(__file__).resolve().parent.parent / "fluid_build" / "schemas" + + +def _connector(trigger: Dict[str, Any]) -> Dict[str, Any]: + return { + "id": "ingest", + "engine": "kafka-connect", + "pattern": "acquisition", + "execution": {"trigger": trigger}, + } + + +def _contract(fluid_version: str = "0.7.6", **trigger: Any) -> Dict[str, Any]: + """The requester's Flink job, with a continuous trigger.""" + return { + "fluidVersion": fluid_version, + "kind": "DataProduct", + "id": "network.service-instances-full", + "name": "Service instances (reconciled)", + "metadata": {"owner": {"team": "data-engineering"}}, + "builds": [ + { + "id": "reconcile", + "engine": "flink" if fluid_version == "0.7.6" else "sql", + "pattern": "embedded-logic", + "properties": { + "language": "flink_sql", + "sql": "INSERT INTO service_instances_full SELECT * FROM service_instances", + "parameters": {"parallelism": 2}, + }, + "execution": { + "trigger": {"type": "continuous", **trigger}, + "runtime": {"platform": "kubernetes"}, + }, + } + ], + "exposes": [ + { + "exposeId": "service_instances_full", + "kind": "topic", + "binding": { + "platform": "kafka", + "format": "kafka_topic", + "location": {"topic": "network.service-instances-full"}, + }, + "contract": {"schema": [{"name": "id", "type": "string"}]}, + } + ], + } + + +def _errors(version: str, contract: Dict[str, Any]) -> list: + schema = json.loads((SCHEMAS / f"fluid-schema-{version}.json").read_text(encoding="utf-8")) + validator = jsonschema.validators.validator_for(schema)(schema) + return [format_schema_error(e) for e in validator.iter_errors(contract)] + + +class TestSchema: + def test_0_7_6_accepts_a_continuous_trigger(self) -> None: + assert _errors("0.7.6", _contract()) == [] + + def test_0_7_5_is_unchanged(self) -> None: + # 0.7.6 is the preview; the stable version does not move. + assert any("'continuous'" in msg for msg in _errors("0.7.5", _contract("0.7.5"))) + + @pytest.mark.parametrize( + "field, value, remedy", + [ + ("cron", "0 2 * * *", "set type: schedule"), + ("schedule", "@daily", "set type: schedule"), + ("datasets", ["dataset://kafka/network.service-instances"], "set type: dataset"), + ("event", "inventory-updated", "set type: event"), + ("timetable", {"kind": "business-hours"}, "set type: timetable"), + ], + ) + def test_another_trigger_beside_it_is_refused( + self, field: str, value: Any, remedy: str + ) -> None: + # Left in, a scheduler that reads the field alone would start a + # second copy of a job that is already running. + assert _errors("0.7.6", _contract(**{field: value})) == [ + f"{field} is not allowed with type continuous: a continuous job runs all the " + f"time, so its trigger takes no other trigger. Remove {field}, or {remedy}." + ] + + @pytest.mark.parametrize( + "trigger", + [ + pytest.param({"type": "schedule", "cron": "0 2 * * *"}, id="schedule-cron"), + pytest.param({"cron": "0 2 * * *"}, id="no-type-cron"), + pytest.param({"type": "dataset", "datasets": ["dataset://x"]}, id="dataset"), + ], + ) + def test_other_trigger_types_keep_their_fields(self, trigger: Dict[str, Any]) -> None: + contract = _contract() + contract["builds"][0]["execution"]["trigger"] = trigger + assert _errors("0.7.6", contract) == [] + + +@pytest.mark.parametrize( + "build, continuous, unscheduled", + [ + pytest.param(_connector({"type": "continuous"}), True, True, id="continuous"), + pytest.param(_connector({"type": " Continuous "}), True, True, id="case-and-space"), + pytest.param(_connector({"type": "schedule"}), False, False, id="schedule"), + pytest.param({"id": "x", "engine": "flink"}, False, True, id="flink-no-trigger"), + pytest.param({"id": "x", "execution": None}, False, False, id="null-execution"), + pytest.param({"id": "x", "execution": {"trigger": "continuous"}}, False, False, id="str"), + ], +) +def test_classification(build: Dict[str, Any], continuous: bool, unscheduled: bool) -> None: + assert is_continuous_build(build) is continuous + assert is_unscheduled_build(build) is unscheduled + + +def test_a_scheduler_gets_no_task_for_a_continuous_connector() -> None: + # A Kafka Connect connector used to get a nightly 02:00 task. + from fluid_build.schedulers.synthesis import synthesize_orchestration_from_builds + + contract = {"id": "p", "builds": [_connector({"type": "continuous"})]} + assert synthesize_orchestration_from_builds(copy.deepcopy(contract), "airflow") == {} + contract["builds"].append({"id": "daily", "engine": "sql", "properties": {"sql": "SELECT 1"}}) + tasks = synthesize_orchestration_from_builds(contract, "airflow")["tasks"] + assert [t["taskId"] for t in tasks] == ["daily"] + + +def test_a_continuous_build_gets_no_fluid_apply_dag() -> None: + from fluid_build.schedulers.airflow.fluid_apply import has_scheduled_builds + + assert has_scheduled_builds({"builds": [_connector({"type": "continuous"})]}) is False + + +def test_schedule_sync_writes_nothing_for_a_continuous_connector(tmp_path: Path) -> None: + # This emitter reads ``schedule`` alone; the schema refuses the pair, but + # a contract that skipped validation must still not start a second copy. + from fluid_build.cli._acquisition_stage_ext import schedule_sync_acquisition + + contract = {"id": "p", "builds": [_connector({"type": "continuous", "schedule": "0 2 * * *"})]} + assert schedule_sync_acquisition(contract, tmp_path) == [] + contract["builds"][0]["execution"]["trigger"]["type"] = "schedule" + assert schedule_sync_acquisition(contract, tmp_path) + + +def _run_cli(*argv: str, cwd: Path) -> "subprocess.CompletedProcess[str]": + return subprocess.run( + [sys.executable, "-m", "fluid_build.cli", *argv], + cwd=str(cwd), + capture_output=True, + text=True, + env={k: v for k, v in os.environ.items() if not k.endswith("_API_KEY")}, + ) + + +class TestCli: + def test_validate_accepts_a_continuous_trigger(self, tmp_path: Path) -> None: + (tmp_path / "contract.fluid.yaml").write_text(yaml.safe_dump(_contract()), "utf-8") + result = _run_cli("validate", "contract.fluid.yaml", "--offline", "--strict", cwd=tmp_path) + assert result.returncode == 0, result.stdout + result.stderr + + def test_validate_names_the_field_and_the_fix(self, tmp_path: Path) -> None: + (tmp_path / "contract.fluid.yaml").write_text( + yaml.safe_dump(_contract(cron="0 2 * * *")), "utf-8" + ) + result = _run_cli("validate", "contract.fluid.yaml", "--offline", cwd=tmp_path) + assert result.returncode == 1 + out = " ".join(result.stdout.split()) + assert "builds[0].execution.trigger: cron is not allowed with type continuous" in out + assert "Remove cron, or set type: schedule." in out + + def test_generate_schedule_has_nothing_to_schedule(self, tmp_path: Path) -> None: + contract = _contract() + contract["builds"] = [_connector({"type": "continuous"})] + (tmp_path / "contract.fluid.yaml").write_text(yaml.safe_dump(contract), "utf-8") + result = _run_cli( + "generate", + "schedule", + "contract.fluid.yaml", + "-o", + "sched", + "--scheduler", + "airflow", + cwd=tmp_path, + ) + assert result.returncode == 0, result.stdout + result.stderr + assert "continuously (trigger.type: continuous)" in " ".join(result.stdout.split()) + assert not (tmp_path / "sched").exists() From 5cdde33ae3c9adba58acc070d07fe9c05a80aeaa Mon Sep 17 00:00:00 2001 From: Speculator55005 <50082482+fas89@users.noreply.github.com> Date: Wed, 21 Oct 2026 10:06:48 +0200 Subject: [PATCH 03/12] feat(validate): --strict fails on an overlay patch that lands on another entry The loader logged overlay_patch_retargeted, and a command's --strict reads its own findings, never the log, so `fluid validate --env prod --strict` passed a 0.7.5 overlay that patched the wrong build. `collect_overlay_warnings()` hands the warning to `fluid validate` and `fluid contract-validation`, which report it as their own warning: shown without --strict, failing with it, and listed in `--format json`. The loader logs a warning a block takes at DEBUG only, so the terminal shows it once, and does not mark it as said, so a later load outside a block still logs it. overlay_entry_added stays a log line: an entry only prod adds is often intended, and --strict could not be silenced on it. Borrowed: the context-variable collector already used by build_runners/kafka_connect/iceberg_sink_validation.py (_AT_RUN), and the shape of warnings.catch_warnings(record=True) under CPython 3.14's context_aware_warnings, which keeps one context's warnings out of another's. --- fluid_build/cli/contract_validation.py | 16 ++- fluid_build/cli/validate.py | 29 ++++- fluid_build/loader.py | 73 +++++++++++- tests/test_overlay_keyed_merge.py | 149 +++++++++++++++++++++++++ 4 files changed, 255 insertions(+), 12 deletions(-) diff --git a/fluid_build/cli/contract_validation.py b/fluid_build/cli/contract_validation.py index 2525e081d..56b29c6b0 100644 --- a/fluid_build/cli/contract_validation.py +++ b/fluid_build/cli/contract_validation.py @@ -292,11 +292,16 @@ def validate(self) -> ValidationReport: """Execute comprehensive contract validation.""" start_time = time.time() - # Load contract + # Load contract. A positional overlay patch that lands on another + # entry is only logged by the loader; collect it, so it is reported + # as a warning below and ``--strict`` fails on it. + from fluid_build.loader import collect_overlay_warnings + try: - self.contract = load_contract_with_overlay( - str(self.contract_path), self.env, self.logger - ) + with collect_overlay_warnings() as overlay_warnings: + self.contract = load_contract_with_overlay( + str(self.contract_path), self.env, self.logger + ) except FileNotFoundError: from .core import ContractNotFoundError @@ -330,6 +335,9 @@ def validate(self) -> ValidationReport: duration=0.0, data_checks_performed=self.check_data, ) + # The message already names the fix. + for warning in overlay_warnings: + self.report.add_issue("warning", "overlay", warning.message) # Step 1: Validate contract syntax against schema self._validate_contract_schema() diff --git a/fluid_build/cli/validate.py b/fluid_build/cli/validate.py index e1398d996..69f5bca65 100644 --- a/fluid_build/cli/validate.py +++ b/fluid_build/cli/validate.py @@ -106,6 +106,9 @@ def register(subparsers: argparse._SubParsersAction): # Strict validation (warnings as errors) fluid validate contract.fluid.yaml --strict + + # Strict, and fail when an overlay patch lands on another build, expose or consume + fluid validate contract.fluid.yaml --env prod --strict """, formatter_class=argparse.RawDescriptionHelpFormatter, ) @@ -301,10 +304,7 @@ def run(args, logger: logging.Logger) -> int: return _run_bundle_validation(contract_path, args, schema_manager, logger, start_time) # Load contract with overlay - try: - contract = load_contract_with_overlay(args.contract, getattr(args, "env", None), logger) - except Exception as e: - raise CLIError(1, "contract_load_failed", {"error": str(e)}) + contract, overlay_warnings = _load_contract_and_overlay_warnings(args, logger) # 0.7-only gate: pre-0.7 schemas (0.4.x / 0.5.x / 0.6.x) are no # longer supported anywhere in the pipeline. Reject them with an @@ -331,6 +331,9 @@ def run(args, logger: logging.Logger) -> int: logger=logger, ) + for warning in overlay_warnings: + validation_result.add_warning(warning) + # Show schema if requested if args.show_schema: _show_schema_info(target_version, schema_manager, args, logger) @@ -452,6 +455,24 @@ def run(args, logger: logging.Logger) -> int: raise CLIError(1, "cli_unhandled_exception", {"error": str(e)}) +def _load_contract_and_overlay_warnings(args, logger: logging.Logger) -> tuple[dict, list[str]]: + """Load ``args.contract`` with its ``--env`` overlay, and the overlay warnings to report. + + A positional overlay patch that lands on another entry is only logged by + the loader, and ``--strict`` reads the validation result, never the log. + :func:`fluid_build.loader.collect_overlay_warnings` hands it to validate + instead, so it is reported as a warning and ``--strict`` fails on it. + """ + from fluid_build.loader import collect_overlay_warnings + + try: + with collect_overlay_warnings() as overlay_warnings: + contract = load_contract_with_overlay(args.contract, getattr(args, "env", None), logger) + except Exception as e: + raise CLIError(1, "contract_load_failed", {"error": str(e)}) + return contract, [warning.message for warning in overlay_warnings] + + def _handle_list_versions(schema_manager: FluidSchemaManager, args, logger: logging.Logger) -> int: """Handle --list-versions flag.""" try: diff --git a/fluid_build/loader.py b/fluid_build/loader.py index 2cc227b27..bc0866180 100644 --- a/fluid_build/loader.py +++ b/fluid_build/loader.py @@ -15,12 +15,26 @@ # fluid_build/loader.py from __future__ import annotations +import contextlib +import contextvars import json import logging import os import threading from pathlib import Path -from typing import Any, Dict, List, Mapping, NamedTuple, Optional, Set, Tuple, Union +from typing import ( + AbstractSet, + Any, + Dict, + Iterator, + List, + Mapping, + NamedTuple, + Optional, + Set, + Tuple, + Union, +) from fluid_build.util.ref_confinement import ( REF_ROOT_ENV, @@ -404,15 +418,66 @@ def _key_fields(item: Any, keys: Tuple[str, ...]) -> Dict[str, str]: _NOTED_OVERLAY_MERGE_WARNINGS_LOCK = threading.Lock() +class OverlayWarning(NamedTuple): + """An overlay merge warning: its event name and its message.""" + + event: str + message: str + + +#: Overlay warnings ``fluid validate`` and ``fluid contract-validation`` report +#: as their own warnings, so that ``--strict`` fails on them: a positional patch +#: that landed on another entry. ``overlay_entry_added`` is often intended (a +#: build only prod runs) and ``--strict`` could not be silenced on it, so it +#: stays a log line. +STRICT_OVERLAY_WARNINGS = frozenset({"overlay_patch_retargeted"}) + +#: The events and the list of the :func:`collect_overlay_warnings` block that +#: is running, ``None`` outside one. A context variable, as +#: ``warnings.catch_warnings`` is under Python 3.14's ``context_aware_warnings``, +#: so a load in another thread or task does not add to this block's list. +_OVERLAY_WARNING_SINK: contextvars.ContextVar[ + Optional[Tuple[AbstractSet[str], List[OverlayWarning]]] +] = contextvars.ContextVar("overlay_warning_sink", default=None) + + +@contextlib.contextmanager +def collect_overlay_warnings( + events: AbstractSet[str] = STRICT_OVERLAY_WARNINGS, +) -> Iterator[List[OverlayWarning]]: + """Hand the overlay merge warnings named in ``events`` to the caller. + + The loader logs its warnings, and a command's ``--strict`` reads its own + findings, never the log. Inside this block a warning named in ``events`` is + appended to the yielded list instead, once, and logged at DEBUG only: the + caller reports it, so the terminal does not show it twice. A warning the + block takes is not marked as said, so a later load outside a block still + logs it. + """ + sink: List[OverlayWarning] = [] + token = _OVERLAY_WARNING_SINK.set((events, sink)) + try: + yield sink + finally: + _OVERLAY_WARNING_SINK.reset(token) + + def _warn_once(event: str, name: str, index: Optional[int], message: str, *args: Any) -> None: text = message % args + extra: Dict[str, Any] = {"event": event, "list": name} + if index is not None: + extra["index"] = index + collecting = _OVERLAY_WARNING_SINK.get() + if collecting is not None and event in collecting[0]: + warning = OverlayWarning(event, text) + if warning not in collecting[1]: + collecting[1].append(warning) + LOG.debug("%s", text, extra=extra) + return with _NOTED_OVERLAY_MERGE_WARNINGS_LOCK: if text in _NOTED_OVERLAY_MERGE_WARNINGS: return _NOTED_OVERLAY_MERGE_WARNINGS.add(text) - extra: Dict[str, Any] = {"event": event, "list": name} - if index is not None: - extra["index"] = index LOG.warning("%s", text, extra=extra) diff --git a/tests/test_overlay_keyed_merge.py b/tests/test_overlay_keyed_merge.py index 81531d66e..05ca7ba98 100644 --- a/tests/test_overlay_keyed_merge.py +++ b/tests/test_overlay_keyed_merge.py @@ -25,8 +25,12 @@ from __future__ import annotations import copy +import json import logging +import os import re +import subprocess +import sys from pathlib import Path from typing import Any, Dict @@ -335,3 +339,148 @@ def test_an_added_entry_names_its_contract_too( "overlay_entry_added: the overlay for p1", "overlay_entry_added: the overlay for p2", ] + + +# --strict: ``fluid validate`` and ``fluid contract-validation`` take a +# retargeted patch from the loader and report it as their own warning, so +# ``--strict`` fails on it. The loader used to log it, which --strict never read. + + +class TestCollectOverlayWarnings: + def test_the_block_takes_a_retargeted_patch_instead_of_logging_it( + self, caplog: pytest.LogCaptureFixture + ) -> None: + from fluid_build.loader import OverlayWarning, collect_overlay_warnings + + caplog.set_level(logging.DEBUG, logger="fluid.loader") + with collect_overlay_warnings() as taken: + _merge(_base("0.7.5"), PROD) + assert [w.event for w in taken] == ["overlay_patch_retargeted"] + assert isinstance(taken[0], OverlayWarning) + assert "names id transform but patches the base's builds[0], id backfill" in ( + taken[0].message + ) + # Said once, by the command: the log keeps it at DEBUG. + levels = [r.levelno for r in caplog.records if r.name == "fluid.loader"] + assert levels == [logging.DEBUG] + + def test_an_added_entry_stays_a_log_line(self, caplog: pytest.LogCaptureFixture) -> None: + # Adding a build only prod runs is often intended, and --strict could + # not be silenced on it. + from fluid_build.loader import collect_overlay_warnings + + caplog.set_level(logging.WARNING, logger="fluid.loader") + with collect_overlay_warnings() as taken: + _merge(_base("0.7.6"), {"builds": [{"id": "reindex", "engine": "spark"}]}) + assert taken == [] + assert _events(caplog) == ["overlay_entry_added"] + + def test_a_warning_the_block_took_is_still_logged_by_a_later_load( + self, caplog: pytest.LogCaptureFixture + ) -> None: + from fluid_build.loader import collect_overlay_warnings + + caplog.set_level(logging.WARNING, logger="fluid.loader") + with collect_overlay_warnings() as taken: + _merge(_base("0.7.5"), PROD) + _merge(_base("0.7.5"), PROD) + assert len(taken) == 1 + assert _events(caplog) == [] + _merge(_base("0.7.5"), PROD) + assert _events(caplog) == ["overlay_patch_retargeted"] + + def test_outside_a_block_nothing_is_collected(self) -> None: + from fluid_build.loader import _OVERLAY_WARNING_SINK, collect_overlay_warnings + + with collect_overlay_warnings(): + pass + assert _OVERLAY_WARNING_SINK.get() is None + + +def _write_product(tmp_path: Path, version: str, overlay: Dict[str, Any]) -> None: + contract = { + "fluidVersion": version, + "kind": "DataProduct", + "id": "network.service-instances", + "name": "Service instances", + "metadata": {"owner": {"team": "data-engineering"}}, + "builds": [{"id": b["id"], "engine": "custom", "description": b["id"]} for b in BUILDS], + "exposes": [ + { + "exposeId": "service_instances", + "kind": "topic", + "binding": { + "platform": "kafka", + "format": "kafka_topic", + "location": {"topic": "network.service-instances"}, + }, + "contract": {"schema": [{"name": "id", "type": "string"}]}, + } + ], + } + (tmp_path / "contract.fluid.yaml").write_text(yaml.safe_dump(contract), encoding="utf-8") + (tmp_path / "overlays").mkdir() + (tmp_path / "overlays" / "prod.yaml").write_text(yaml.safe_dump(overlay), encoding="utf-8") + + +#: Written against ``[ingest, transform]``: its second entry patches transform. +PROD_BY_POSITION = {"builds": [{}, {"id": "transform", "description": "8 executors"}]} + + +def _validate(tmp_path: Path, *argv: str) -> "subprocess.CompletedProcess[str]": + return subprocess.run( + [sys.executable, "-m", "fluid_build.cli", "validate", "contract.fluid.yaml"] + + ["--env", "prod", "--offline", *argv], + cwd=str(tmp_path), + capture_output=True, + text=True, + env={k: v for k, v in os.environ.items() if not k.endswith("_API_KEY")}, + ) + + +class TestValidateStrict: + def test_the_retargeted_patch_is_a_warning(self, tmp_path: Path) -> None: + _write_product(tmp_path, "0.7.5", PROD_BY_POSITION) + result = _validate(tmp_path) + assert result.returncode == 0, result.stdout + result.stderr + out = result.stdout + result.stderr + assert "Validation Warnings:" in result.stdout + # Reported by validate, not also logged by the loader. + assert out.count("overlay_patch_retargeted") == 1 + assert "the overlay's builds[1] names id transform but patches the base's builds[1], " in ( + " ".join(result.stdout.split()) + ) + + def test_strict_fails_on_it(self, tmp_path: Path) -> None: + _write_product(tmp_path, "0.7.5", PROD_BY_POSITION) + assert _validate(tmp_path, "--strict").returncode == 1 + + def test_strict_json_lists_it(self, tmp_path: Path) -> None: + _write_product(tmp_path, "0.7.5", PROD_BY_POSITION) + result = _validate(tmp_path, "--strict", "--format", "json") + assert result.returncode == 1 + warnings = json.loads(result.stdout)["warnings"] + assert [w.split(":")[0] for w in warnings] == ["overlay_patch_retargeted"] + + def test_keyed_merge_has_nothing_to_warn_about(self, tmp_path: Path) -> None: + overlay = {"builds": [{"id": "transform", "description": "8 executors"}]} + _write_product(tmp_path, "0.7.6", overlay) + result = _validate(tmp_path, "--strict") + assert result.returncode == 0, result.stdout + result.stderr + + def test_an_added_build_does_not_fail_strict(self, tmp_path: Path) -> None: + overlay = {"builds": [{"id": "reindex", "engine": "custom", "description": "reindex"}]} + _write_product(tmp_path, "0.7.6", overlay) + result = _validate(tmp_path, "--strict") + assert result.returncode == 0, result.stdout + result.stderr + + +def test_contract_validation_reports_it_as_a_warning(tmp_path: Path) -> None: + from fluid_build.cli.contract_validation import ContractValidator + + _write_product(tmp_path, "0.7.5", PROD_BY_POSITION) + report = ContractValidator( + tmp_path / "contract.fluid.yaml", env="prod", check_data=False, track_history=False + ).validate() + overlay = [w for w in report.get_warnings() if w.category == "overlay"] + assert [w.message.split(":")[0] for w in overlay] == ["overlay_patch_retargeted"] From 22d7fd950c3125577aa50215ff68e6a012f789cd Mon Sep 17 00:00:00 2001 From: Speculator55005 <50082482+fas89@users.noreply.github.com> Date: Wed, 21 Oct 2026 10:06:49 +0200 Subject: [PATCH 04/12] docs(changelog): trigger.type continuous, --strict on retargeted overlay patches, readable not-required errors --- CHANGELOG.md | 30 ++++++++++++++++++++++++++++++ 1 file changed, 30 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 99c7f209a..fd1b356cd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,36 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Upgrade notes + +- `fluid validate --env --strict` and `fluid contract-validation --env + --strict` now fail when an overlay patch lands on an entry that names another id + (`overlay_patch_retargeted`), which a 0.7.5 or older contract, merged by position, can + do. Without `--strict` both commands list it under their warnings, and the loader no + longer also logs it there. Put the overlay's entries in the base's order, or set + `fluidVersion: "0.7.6"` so overlays patch by id. `overlay_entry_added` stays a log line + and does not fail `--strict`. `fluid validate .tgz` does not see the warning: + `fluid bundle --env` merged the overlay before, so validate the contract file with + `--env`. + +### Added + +- **`trigger.type: continuous` (0.7.6 preview).** A build whose job its platform keeps + running, such as an `engine: flink` job or a `kafka-connect` or `debezium` connector, + can say so: `execution.trigger: {type: continuous}`. No scheduler starts it: `fluid + generate schedule`, `fluid schedule-sync` and the `fluid apply` DAGs leave the build out, + where a connector used to get a nightly `0 2 * * *` task. A contract with no other build + has nothing to schedule. The schema refuses `cron`, `schedule`, `datasets`, `event` and + `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. + +### Changed + +- A schema error from a `not: {required: [...]}` rule now reads ` is not allowed + here`, or the rule's own `description`, instead of quoting the whole object and the + subschema it must not match. + ## [0.21.0] — 2026-10-19 ### Upgrade notes From a182b5874ba85b2b597f39f97a03373db7e5a855 Mon Sep 17 00:00:00 2001 From: Speculator55005 <50082482+fas89@users.noreply.github.com> Date: Wed, 21 Oct 2026 10:33:14 +0200 Subject: [PATCH 05/12] fix(validate): every --strict that fails on a retargeted overlay patch shows it `fluid test --strict` reads ContractValidator's report, so it failed on the overlay warning without printing it: its table had no row for the category, and the loader now logs a collected warning at DEBUG only. It shows as an "Environment overlay" row. Workspace mode (`fluid validate --env prod --strict` with no contract) loaded each product outside the collector and passed; it now lists the warning under the product and fails it under --strict. `fluid validate --strict` says "Validation failed: warnings treated as errors (--strict)" when warnings alone fail it, after a summary that reads valid. The --help example fits 80 columns. --- fluid_build/cli/test.py | 6 +++ fluid_build/cli/validate.py | 76 +++++++++++++++++++++---------- tests/test_overlay_keyed_merge.py | 44 +++++++++++++++++- 3 files changed, 99 insertions(+), 27 deletions(-) diff --git a/fluid_build/cli/test.py b/fluid_build/cli/test.py index 9ba646dcd..a573b0ffc 100644 --- a/fluid_build/cli/test.py +++ b/fluid_build/cli/test.py @@ -439,6 +439,12 @@ def _warns(*categories: str) -> List: else: rows.append((_PASS, "Schema syntax", "Valid")) + # Environment overlay (only rendered when the --env overlay patched an + # entry it does not name): --strict fails on it, so it must show. + overlay_issues = [i for i in issues if i.category == "overlay"] + if overlay_issues: + rows.append((_WARN, "Environment overlay", "; ".join(e.message for e in overlay_issues))) + # Provider connectivity # The connection probe runs in provider detection, so it happens # under --no-data too. diff --git a/fluid_build/cli/validate.py b/fluid_build/cli/validate.py index 69f5bca65..fe0c7f817 100644 --- a/fluid_build/cli/validate.py +++ b/fluid_build/cli/validate.py @@ -107,7 +107,7 @@ def register(subparsers: argparse._SubParsersAction): # Strict validation (warnings as errors) fluid validate contract.fluid.yaml --strict - # Strict, and fail when an overlay patch lands on another build, expose or consume + # Strict: also fail when an overlay patch lands on another entry fluid validate contract.fluid.yaml --env prod --strict """, formatter_class=argparse.RawDescriptionHelpFormatter, @@ -304,7 +304,12 @@ def run(args, logger: logging.Logger) -> int: return _run_bundle_validation(contract_path, args, schema_manager, logger, start_time) # Load contract with overlay - contract, overlay_warnings = _load_contract_and_overlay_warnings(args, logger) + try: + contract, overlay_warnings = _load_with_overlay_warnings( + args.contract, getattr(args, "env", None), logger + ) + except Exception as e: + raise CLIError(1, "contract_load_failed", {"error": str(e)}) # 0.7-only gate: pre-0.7 schemas (0.4.x / 0.5.x / 0.6.x) are no # longer supported anywhere in the pipeline. Reject them with an @@ -455,8 +460,10 @@ def run(args, logger: logging.Logger) -> int: raise CLIError(1, "cli_unhandled_exception", {"error": str(e)}) -def _load_contract_and_overlay_warnings(args, logger: logging.Logger) -> tuple[dict, list[str]]: - """Load ``args.contract`` with its ``--env`` overlay, and the overlay warnings to report. +def _load_with_overlay_warnings( + contract_path: str, env: Optional[str], logger: logging.Logger +) -> tuple[dict, list[str]]: + """Load a contract with its ``--env`` overlay, and the overlay warnings to report. A positional overlay patch that lands on another entry is only logged by the loader, and ``--strict`` reads the validation result, never the log. @@ -465,11 +472,8 @@ def _load_contract_and_overlay_warnings(args, logger: logging.Logger) -> tuple[d """ from fluid_build.loader import collect_overlay_warnings - try: - with collect_overlay_warnings() as overlay_warnings: - contract = load_contract_with_overlay(args.contract, getattr(args, "env", None), logger) - except Exception as e: - raise CLIError(1, "contract_load_failed", {"error": str(e)}) + with collect_overlay_warnings() as overlay_warnings: + contract = load_contract_with_overlay(contract_path, env, logger) return contract, [warning.message for warning in overlay_warnings] @@ -1279,11 +1283,21 @@ def _output_text_results(result: ValidationResult, args, logger: logging.Logger) treat_warnings_as_errors = args.strict and has_warnings if has_errors or treat_warnings_as_errors: + if not has_errors: + _explain_strict_failure(result, args) return 1 else: return 0 +def _explain_strict_failure(result: ValidationResult, args: Any) -> None: + """Say why ``--strict`` exits 1 after a summary that reads valid.""" + if args.quiet: + _cprint_literal(f"ERROR: {len(result.warnings)} warning(s) treated as errors (--strict)") + else: + cprint("❌ Validation failed: warnings treated as errors (--strict)") + + # --------------------------------------------------------------------------- # Public helpers for other CLI commands (publish, apply, …) that need to # run FLUID schema validation on an already-loaded contract dict. @@ -1398,19 +1412,17 @@ def _try_workspace_validate( failed = 0 for product in products: try: - contract = _load_contract_for_workspace(product.contract_path, args, logger) + contract, overlay_warnings = _load_with_overlay_warnings( + str(product.contract_path), getattr(args, "env", None), logger + ) result = schema_manager.validate_contract( contract, strict=args.strict, offline_only=getattr(args, "offline", False), ) - if result.is_valid: - cprint(f" ✅ {product.name:<20} valid ({product.fluid_version or '?'})") + if _report_workspace_product(product, result, overlay_warnings, args.strict): passed += 1 else: - cprint(f" ❌ {product.name:<20} {len(result.errors)} error(s)") - for err in result.errors[:3]: - _cprint_literal(f" {err}") failed += 1 except Exception as exc: # noqa: BLE001 _cprint_literal(f" ❌ {product.name:<20} load error: {exc}") @@ -1424,6 +1436,30 @@ def _try_workspace_validate( return 0 +def _report_workspace_product( + product: Any, result: ValidationResult, overlay_warnings: list[str], strict: bool +) -> bool: + """Print one workspace product's line, and whether it passed. + + ``--strict`` fails a product on an overlay patch that lands on another + entry; its other warnings are not reported here. + """ + if not result.is_valid: + cprint(f" ❌ {product.name:<20} {len(result.errors)} error(s)") + for err in result.errors[:3]: + _cprint_literal(f" {err}") + elif overlay_warnings and strict: + cprint( + f" ❌ {product.name:<20} {len(overlay_warnings)} overlay warning(s), " + "treated as errors (--strict)" + ) + else: + cprint(f" ✅ {product.name:<20} valid ({product.fluid_version or '?'})") + for warning in overlay_warnings: + _cprint_literal(f" ⚠️ {warning}") + return result.is_valid and not (overlay_warnings and strict) + + # --------------------------------------------------------------------------- # Bundle (.tgz) validation — stage 2 of the 11-stage pipeline # --------------------------------------------------------------------------- @@ -1624,13 +1660,3 @@ def _bundle_contract_rule_issues( for msg in result.warnings ) return issues - - -def _load_contract_for_workspace( - contract_path: Path, - args: Any, - logger: logging.Logger, -) -> dict: - """Load a contract, applying environment overlay if requested.""" - env = getattr(args, "env", None) - return load_contract_with_overlay(str(contract_path), env, logger) diff --git a/tests/test_overlay_keyed_merge.py b/tests/test_overlay_keyed_merge.py index 05ca7ba98..f77b828c2 100644 --- a/tests/test_overlay_keyed_merge.py +++ b/tests/test_overlay_keyed_merge.py @@ -31,6 +31,7 @@ import re import subprocess import sys +from datetime import datetime from pathlib import Path from typing import Any, Dict @@ -451,9 +452,12 @@ def test_the_retargeted_patch_is_a_warning(self, tmp_path: Path) -> None: " ".join(result.stdout.split()) ) - def test_strict_fails_on_it(self, tmp_path: Path) -> None: + def test_strict_fails_on_it_and_says_why(self, tmp_path: Path) -> None: + # The summary above reads "Valid FLUID contract". _write_product(tmp_path, "0.7.5", PROD_BY_POSITION) - assert _validate(tmp_path, "--strict").returncode == 1 + result = _validate(tmp_path, "--strict") + assert result.returncode == 1 + assert "Validation failed: warnings treated as errors (--strict)" in result.stdout def test_strict_json_lists_it(self, tmp_path: Path) -> None: _write_product(tmp_path, "0.7.5", PROD_BY_POSITION) @@ -475,6 +479,42 @@ def test_an_added_build_does_not_fail_strict(self, tmp_path: Path) -> None: assert result.returncode == 0, result.stdout + result.stderr +def test_workspace_validate_strict_fails_on_it(tmp_path: Path) -> None: + # `fluid validate --env prod --strict` with no contract validates every + # product in the workspace. + (tmp_path / "fluid.workspace.yaml").write_text( + yaml.safe_dump({"products_dir": "products"}), encoding="utf-8" + ) + product = tmp_path / "products" / "svc" + product.mkdir(parents=True) + _write_product(product, "0.7.5", PROD_BY_POSITION) + run = [sys.executable, "-m", "fluid_build.cli", "validate", "--env", "prod", "--offline"] + env = {k: v for k, v in os.environ.items() if not k.endswith("_API_KEY")} + lenient = subprocess.run(run, cwd=tmp_path, capture_output=True, text=True, env=env) + strict = subprocess.run( + run + ["--strict"], cwd=tmp_path, capture_output=True, text=True, env=env + ) + assert lenient.returncode == 0, lenient.stdout + lenient.stderr + assert "overlay_patch_retargeted" in lenient.stdout + assert strict.returncode == 1 + assert "1 overlay warning(s), treated as errors (--strict)" in strict.stdout + + +def test_fluid_test_shows_it_as_a_row() -> None: + # `fluid test --strict` failed on it without printing it. + from fluid_build.cli.contract_validation import ValidationReport + from fluid_build.cli.test import _build_check_rows + + report = ValidationReport("c.yaml", "p", "1", datetime.now(), 0.0) + report.add_issue("warning", "overlay", "overlay_patch_retargeted: in p, the overlay's ...") + rows = {name: (status, detail) for status, name, detail in _build_check_rows(report)} + status, detail = rows["Environment overlay"] + assert status != rows["Schema syntax"][0] + assert detail.startswith("overlay_patch_retargeted: in p") + report.issues.clear() + assert "Environment overlay" not in [name for _, name, _ in _build_check_rows(report)] + + def test_contract_validation_reports_it_as_a_warning(tmp_path: Path) -> None: from fluid_build.cli.contract_validation import ContractValidator From 8d0b730ca4e4e4b1b562141e1e24671698dc6d87 Mon Sep 17 00:00:00 2001 From: Speculator55005 <50082482+fas89@users.noreply.github.com> Date: Wed, 21 Oct 2026 10:33:14 +0200 Subject: [PATCH 06/12] fix(schedule): schedule-sync skips every build no scheduler starts, through one predicate The acquisition schedule-sync used is_continuous_build while synthesis and `fluid generate schedule` use is_unscheduled_build; widening the rule would have missed it. The fluid-apply DAG test now gives the continuous trigger a cron, with a schedule-trigger control, so it can fail. --- fluid_build/cli/_acquisition_stage_ext.py | 4 ++-- tests/test_continuous_trigger.py | 7 +++++-- 2 files changed, 7 insertions(+), 4 deletions(-) diff --git a/fluid_build/cli/_acquisition_stage_ext.py b/fluid_build/cli/_acquisition_stage_ext.py index 81048e038..936ea5343 100644 --- a/fluid_build/cli/_acquisition_stage_ext.py +++ b/fluid_build/cli/_acquisition_stage_ext.py @@ -59,7 +59,7 @@ from fluid_build.build_runners._ids import ( validate_identifier as _validate_identifier, ) -from fluid_build.util.contract import is_continuous_build +from fluid_build.util.contract import is_unscheduled_build LOG = logging.getLogger("fluid.acquire.stage_ext") @@ -484,7 +484,7 @@ def schedule_sync_acquisition( bid = _validate_identifier(build.get("id", ""), kind="build.id") # A continuous connector keeps running on its own: a schedule beside it # is refused by schema 0.7.6, and must not start a second copy here. - if is_continuous_build(build): + if is_unscheduled_build(build): continue execution = build.get("execution", {}) or {} trigger = execution.get("trigger", {}) or {} diff --git a/tests/test_continuous_trigger.py b/tests/test_continuous_trigger.py index 0ab9125c2..c5f43b2a9 100644 --- a/tests/test_continuous_trigger.py +++ b/tests/test_continuous_trigger.py @@ -164,10 +164,13 @@ def test_a_scheduler_gets_no_task_for_a_continuous_connector() -> None: assert [t["taskId"] for t in tasks] == ["daily"] -def test_a_continuous_build_gets_no_fluid_apply_dag() -> None: +@pytest.mark.parametrize("kind, scheduled", [("continuous", False), ("schedule", True)]) +def test_a_continuous_build_gets_no_fluid_apply_dag(kind: str, scheduled: bool) -> None: + # With a cron beside it, as a contract that skipped validation can have. from fluid_build.schedulers.airflow.fluid_apply import has_scheduled_builds - assert has_scheduled_builds({"builds": [_connector({"type": "continuous"})]}) is False + contract = {"builds": [_connector({"type": kind, "cron": "0 2 * * *"})]} + assert has_scheduled_builds(contract) is scheduled def test_schedule_sync_writes_nothing_for_a_continuous_connector(tmp_path: Path) -> None: From ebf15139e89b1ff4b3e8c5520de0e4e00d32a706 Mon Sep 17 00:00:00 2001 From: Speculator55005 <50082482+fas89@users.noreply.github.com> Date: Wed, 21 Oct 2026 10:33:14 +0200 Subject: [PATCH 07/12] docs: continuous builds get no DAG, a not-required rule's description is its error, --strict overlay coverage --- AGENTS.md | 2 +- CHANGELOG.md | 22 +++++++++++++--------- docs/schedule.md | 3 +++ 3 files changed, 17 insertions(+), 10 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index e24e0ca37..66971f58d 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -193,7 +193,7 @@ new extra installed). `uv` is dev-time only — not a runtime dep. - **Adding a provider**: Create `fluid_build/providers//`, implement `BaseProvider` interface - **Adding a template**: Create `fluid_build/templates/.j2` or add to `fluid_build/forge/templates/` - **Adding a policy rule**: Extend validators in `fluid_build/policy/` -- **Modifying the contract schema**: Update `fluid_build/schemas/` and bump `fluidVersion`. Never hardcode the new version as a literal — `SchemaManager.latest_bundled_version()` scans the schemas directory and returns the newest, so fallbacks track future bumps automatically. +- **Modifying the contract schema**: Update `fluid_build/schemas/` and bump `fluidVersion`. Never hardcode the new version as a literal — `SchemaManager.latest_bundled_version()` scans the schemas directory and returns the newest, so fallbacks track future bumps automatically. A `description` on a `not: {required: [field]}` rule is the error message `fluid validate` prints for it (`schema_manager.format_schema_error`), so write it as one: what is refused and what to do instead. - **Data-product type vocabulary** (v0.7.3+): contracts can use medallion (`metadata.layer ∈ {Bronze, Silver, Gold, Platinum}`) or Data Mesh (`metadata.productType ∈ {SDP, ADP, CDP}`) — they're equivalent. Canonical mapping: **Bronze↔SDP** (Source-Aligned Data Product), **Silver↔ADP** (Aggregated), **Gold↔CDP** (Consumption-Aligned). Either-or-both is accepted; when both are set the validator enforces consistency. Cross-check is duplicated inline at three call sites (`fluid_build/schema.py::_check_metadata`, `fluid_build/schema_manager.py::_validate_with_jsonschema`, `fluid_build/cli/contract_validation.py::_validate_metadata`) — keep them in lockstep when changing the mapping. Provider tag/label propagation in `providers/{aws,gcp,snowflake}` and `forge/core/provider_actions.py` emits both as separate cloud labels (`fluid_layer` + `fluid_product_type`). - **Adding a new apply mode**: extend `fluid_build/forge/core/apply_modes.py::ApplyMode` enum + `CANONICAL_CHOICES` list + predicates (`is_destructive`, `needs_build`, etc.) + `check_data_loss_gate` if the mode is destructive. The mode must map to a provider-native DDL flavor in `providers//actions/*.py`. - **Changing what `fluid plan` emits**: `plan.json` must carry both `planDigest` and `bundleDigest` (re-run `inject_digests` after any mutation to the dict). Actions must carry BOTH `op` and `action_type` fields — apply.py's provider dispatcher reads `op`, display/viz reads `action_type`. Dropping either silently breaks the pipeline. diff --git a/CHANGELOG.md b/CHANGELOG.md index fd1b356cd..8a1ee4c5b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,15 +9,19 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Upgrade notes -- `fluid validate --env --strict` and `fluid contract-validation --env - --strict` now fail when an overlay patch lands on an entry that names another id - (`overlay_patch_retargeted`), which a 0.7.5 or older contract, merged by position, can - do. Without `--strict` both commands list it under their warnings, and the loader no - longer also logs it there. Put the overlay's entries in the base's order, or set - `fluidVersion: "0.7.6"` so overlays patch by id. `overlay_entry_added` stays a log line - and does not fail `--strict`. `fluid validate .tgz` does not see the warning: - `fluid bundle --env` merged the overlay before, so validate the contract file with - `--env`. +- `fluid validate --env --strict` (on one contract or a whole workspace), `fluid + contract-validation --env --strict` and `fluid test --env --strict` now fail + when an overlay patch lands on an entry that names another id + (`overlay_patch_retargeted`), which an overlay merged by position can do: a 0.7.5 or + older contract, or a 0.7.6 base whose entries lack unique ids. Without `--strict` these + commands list it among their warnings (`fluid test` as an "Environment overlay" row), + and the loader no longer also logs it there. Put the overlay's entries in the base's + order, or set `fluidVersion: "0.7.6"` so overlays patch by id. `overlay_entry_added` + stays a log line and does not fail `--strict`. `fluid validate .tgz` does not + see the warning: `fluid bundle --env` merged the overlay before, so validate the + contract file with `--env`. +- `fluid validate --strict` now says `Validation failed: warnings treated as errors + (--strict)` when warnings alone fail it, after a summary that reads valid. ### Added diff --git a/docs/schedule.md b/docs/schedule.md index 59a5e6a3c..c1ab5d46e 100644 --- a/docs/schedule.md +++ b/docs/schedule.md @@ -38,6 +38,9 @@ hand-declares `orchestration.tasks` (those keep their declared operators). A build is left out when its own `execution.orchestration.engine` is `none` or not Airflow, or when its trigger is `event`, `dataset`, `schedule_and_dataset` or `timetable`, which a plain cron DAG cannot express. +A build whose trigger is `continuous` (schema 0.7.6), or that runs outside +forge-cli (`engine: flink`, `language: flink_sql`), gets no DAG and no task +at all: its platform keeps the job running. Layout, one directory per product: From 82f53064cca03e7f240d547e35e7808447897ac0 Mon Sep 17 00:00:00 2001 From: Speculator55005 <50082482+fas89@users.noreply.github.com> Date: Wed, 21 Oct 2026 10:59:55 +0200 Subject: [PATCH 08/12] fix(schema): the continuous refusals apply only to a trigger object The rule's `if` had no `type: object`, and `required` and `properties` pass on anything else, so a blank `trigger:` (null), a string or a list drew five refusals of a continuous trigger nobody declared beside the real error. The sibling catalogAuth rules already say `type: object`. --- fluid_build/schemas/fluid-schema-0.7.6.json | 1 + tests/test_continuous_trigger.py | 9 +++++++++ 2 files changed, 10 insertions(+) diff --git a/fluid_build/schemas/fluid-schema-0.7.6.json b/fluid_build/schemas/fluid-schema-0.7.6.json index eda67c969..bb76e772d 100644 --- a/fluid_build/schemas/fluid-schema-0.7.6.json +++ b/fluid_build/schemas/fluid-schema-0.7.6.json @@ -4388,6 +4388,7 @@ "allOf": [ { "if": { + "type": "object", "required": [ "type" ], diff --git a/tests/test_continuous_trigger.py b/tests/test_continuous_trigger.py index c5f43b2a9..2d5e5cc2e 100644 --- a/tests/test_continuous_trigger.py +++ b/tests/test_continuous_trigger.py @@ -123,6 +123,15 @@ def test_another_trigger_beside_it_is_refused( f"time, so its trigger takes no other trigger. Remove {field}, or {remedy}." ] + @pytest.mark.parametrize("trigger", [None, "@daily", ["0 2 * * *"]]) + def test_a_trigger_that_is_not_an_object_gets_one_error(self, trigger: Any) -> None: + # A blank `trigger:` in YAML is null; it used to draw five refusals + # of a continuous trigger nobody declared. + contract = _contract() + contract["builds"][0]["execution"]["trigger"] = trigger + assert [m for m in _errors("0.7.6", contract) if "continuous" in m] == [] + assert len(_errors("0.7.6", contract)) == 1 + @pytest.mark.parametrize( "trigger", [ From d8703c9be8aa3f04e2c5c9f065dd0a7d377ae708 Mon Sep 17 00:00:00 2001 From: Speculator55005 <50082482+fas89@users.noreply.github.com> Date: Wed, 21 Oct 2026 10:59:55 +0200 Subject: [PATCH 09/12] fix(validate): --quiet still names a retargeted overlay patch, and tables escape its ids Under --quiet, which prints errors only, the warning the loader always printed had vanished, and --quiet --strict failed with a count alone. --quiet prints it as a WARNING: line, and under --strict as "ERROR (--strict): ". The warning quotes the contract's ids, which `fluid test`'s "Environment overlay" row and `fluid contract-validation`'s issue table passed to Rich as markup: an id of "[/]" raised MarkupError. Both escape it, the row through test.py's own _detail(). --- fluid_build/cli/contract_validation.py | 8 ++++++- fluid_build/cli/test.py | 2 +- fluid_build/cli/validate.py | 22 +++++++++++++++++-- tests/test_overlay_keyed_merge.py | 29 ++++++++++++++++++++++++++ 4 files changed, 57 insertions(+), 4 deletions(-) diff --git a/fluid_build/cli/contract_validation.py b/fluid_build/cli/contract_validation.py index 56b29c6b0..5bd0deaa7 100644 --- a/fluid_build/cli/contract_validation.py +++ b/fluid_build/cli/contract_validation.py @@ -1633,6 +1633,8 @@ def output_rich_report(report: ValidationReport, output_file: Optional[str] = No # Issues table if report.issues: + from rich.markup import escape + table = Table(title="Validation Issues", show_header=True) table.add_column("Severity", style="bold") table.add_column("Category") @@ -1646,7 +1648,11 @@ def output_rich_report(report: ValidationReport, output_file: Optional[str] = No "info": "[blue]INFO[/blue]", }.get(issue.severity, issue.severity.upper()) - table.add_row(severity_style, issue.category, issue.message, issue.path) + # Messages quote contract text (an overlay warning names the + # contract's ids), which Rich would parse as markup. + table.add_row( + severity_style, issue.category, escape(issue.message), escape(issue.path or "") + ) console.print(table) else: diff --git a/fluid_build/cli/test.py b/fluid_build/cli/test.py index a573b0ffc..80c61bb39 100644 --- a/fluid_build/cli/test.py +++ b/fluid_build/cli/test.py @@ -443,7 +443,7 @@ def _warns(*categories: str) -> List: # entry it does not name): --strict fails on it, so it must show. overlay_issues = [i for i in issues if i.category == "overlay"] if overlay_issues: - rows.append((_WARN, "Environment overlay", "; ".join(e.message for e in overlay_issues))) + rows.append((_WARN, "Environment overlay", _detail(e.message for e in overlay_issues))) # Provider connectivity # The connection probe runs in provider detection, so it happens diff --git a/fluid_build/cli/validate.py b/fluid_build/cli/validate.py index fe0c7f817..66a175c49 100644 --- a/fluid_build/cli/validate.py +++ b/fluid_build/cli/validate.py @@ -363,6 +363,7 @@ def run(args, logger: logging.Logger) -> int: log_metric(logger, "validation_warnings", len(validation_result.warnings), unit="count") # Output results + _print_quiet_overlay_warnings(args, overlay_warnings) exit_code = _output_results(validation_result, args, logger) # Log operation result @@ -1291,13 +1292,30 @@ def _output_text_results(result: ValidationResult, args, logger: logging.Logger) def _explain_strict_failure(result: ValidationResult, args: Any) -> None: - """Say why ``--strict`` exits 1 after a summary that reads valid.""" + """Say why ``--strict`` exits 1 after a summary that reads valid. + + ``--quiet`` prints errors only, so there each warning is printed as the + error ``--strict`` made it: a count alone does not say what to fix. + """ if args.quiet: - _cprint_literal(f"ERROR: {len(result.warnings)} warning(s) treated as errors (--strict)") + for warning in result.warnings: + _cprint_literal(f"ERROR (--strict): {warning}") else: cprint("❌ Validation failed: warnings treated as errors (--strict)") +def _print_quiet_overlay_warnings(args: Any, overlay_warnings: list[str]) -> None: + """Under ``--quiet``, still show an overlay patch that landed on another entry. + + ``--quiet`` prints errors only, but the loader always printed this warning, + whatever the flags. Under ``--strict`` it is printed as an error instead + (:func:`_explain_strict_failure`). + """ + if args.quiet and not args.strict and args.format == "text": + for warning in overlay_warnings: + _cprint_literal(f"WARNING: {warning}") + + # --------------------------------------------------------------------------- # Public helpers for other CLI commands (publish, apply, …) that need to # run FLUID schema validation on an already-loaded contract dict. diff --git a/tests/test_overlay_keyed_merge.py b/tests/test_overlay_keyed_merge.py index f77b828c2..f64ef2f8d 100644 --- a/tests/test_overlay_keyed_merge.py +++ b/tests/test_overlay_keyed_merge.py @@ -459,6 +459,21 @@ def test_strict_fails_on_it_and_says_why(self, tmp_path: Path) -> None: assert result.returncode == 1 assert "Validation failed: warnings treated as errors (--strict)" in result.stdout + def test_quiet_still_shows_it(self, tmp_path: Path) -> None: + # --quiet prints errors only, but the loader always printed this one. + _write_product(tmp_path, "0.7.5", PROD_BY_POSITION) + result = _validate(tmp_path, "--quiet") + assert result.returncode == 0, result.stdout + result.stderr + assert result.stdout.startswith("WARNING: overlay_patch_retargeted: in ") + assert (result.stdout + result.stderr).count("overlay_patch_retargeted") == 1 + + def test_quiet_strict_names_it_as_the_error(self, tmp_path: Path) -> None: + _write_product(tmp_path, "0.7.5", PROD_BY_POSITION) + result = _validate(tmp_path, "--quiet", "--strict") + assert result.returncode == 1 + assert result.stdout.startswith("ERROR (--strict): overlay_patch_retargeted: in ") + assert (result.stdout + result.stderr).count("overlay_patch_retargeted") == 1 + def test_strict_json_lists_it(self, tmp_path: Path) -> None: _write_product(tmp_path, "0.7.5", PROD_BY_POSITION) result = _validate(tmp_path, "--strict", "--format", "json") @@ -515,6 +530,20 @@ def test_fluid_test_shows_it_as_a_row() -> None: assert "Environment overlay" not in [name for _, name, _ in _build_check_rows(report)] +def test_an_id_that_reads_as_markup_does_not_break_either_table(capsys) -> None: + # The message quotes the contract's ids; "[/]" made Rich raise MarkupError. + from fluid_build.cli.contract_validation import ValidationReport, output_rich_report + from fluid_build.cli.test import _output_rich + + message = "overlay_patch_retargeted: in p, the overlay's builds[1] names id [/] but ..." + report = ValidationReport("c.yaml", "p", "1", datetime.now(), 0.0) + report.add_issue("warning", "overlay", message) + output_rich_report(report) + _output_rich(report) + # Each table shows the id as written (the tables wrap the rest of the line). + assert capsys.readouterr().out.count("[/]") == 2 + + def test_contract_validation_reports_it_as_a_warning(tmp_path: Path) -> None: from fluid_build.cli.contract_validation import ContractValidator From a037779da3dc9d0b280b0e8275c316729f1adb6a Mon Sep 17 00:00:00 2001 From: Speculator55005 <50082482+fas89@users.noreply.github.com> Date: Wed, 21 Oct 2026 10:59:55 +0200 Subject: [PATCH 10/12] docs: quiet and soda scope of --strict, stale DAGs and overlays for continuous builds --- CHANGELOG.md | 30 +++++++++++++++++++----------- docs/schedule.md | 2 +- 2 files changed, 20 insertions(+), 12 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 8a1ee4c5b..2ed522049 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,18 +10,26 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Upgrade notes - `fluid validate --env --strict` (on one contract or a whole workspace), `fluid - contract-validation --env --strict` and `fluid test --env --strict` now fail - when an overlay patch lands on an entry that names another id - (`overlay_patch_retargeted`), which an overlay merged by position can do: a 0.7.5 or - older contract, or a 0.7.6 base whose entries lack unique ids. Without `--strict` these - commands list it among their warnings (`fluid test` as an "Environment overlay" row), - and the loader no longer also logs it there. Put the overlay's entries in the base's - order, or set `fluidVersion: "0.7.6"` so overlays patch by id. `overlay_entry_added` - stays a log line and does not fail `--strict`. `fluid validate .tgz` does not - see the warning: `fluid bundle --env` merged the overlay before, so validate the - contract file with `--env`. + contract-validation --env --strict` and `fluid test --env --strict` (the + native engine; `--engine soda` does not read `--strict`) now fail when an overlay patch + lands on an entry that names another id (`overlay_patch_retargeted`), which an overlay + merged by position can do: a 0.7.5 or older contract, or a 0.7.6 base whose entries lack + unique ids. Without `--strict` these commands list it among their warnings (`fluid test` + as an "Environment overlay" row, `fluid validate --quiet` as a `WARNING:` line), and the + loader no longer also logs it there. Put the overlay's entries in the base's order, or + set `fluidVersion: "0.7.6"` with a unique id on every base entry, so overlays patch by + id. `overlay_entry_added` stays a log line and does not fail `--strict`. `fluid validate + .tgz` does not see the warning: `fluid bundle --env` merged the overlay before, + so validate the contract file with `--env`. - `fluid validate --strict` now says `Validation failed: warnings treated as errors - (--strict)` when warnings alone fail it, after a summary that reads valid. + (--strict)` when warnings alone fail it, after a summary that reads valid; with + `--quiet` it prints each such warning as `ERROR (--strict): `. +- 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. +- 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. ### Added diff --git a/docs/schedule.md b/docs/schedule.md index c1ab5d46e..2d5540e10 100644 --- a/docs/schedule.md +++ b/docs/schedule.md @@ -39,7 +39,7 @@ build is left out when its own `execution.orchestration.engine` is `none` or not Airflow, or when its trigger is `event`, `dataset`, `schedule_and_dataset` or `timetable`, which a plain cron DAG cannot express. A build whose trigger is `continuous` (schema 0.7.6), or that runs outside -forge-cli (`engine: flink`, `language: flink_sql`), gets no DAG and no task +forge-cli (`engine: flink`, `properties.language: flink_sql`), gets no DAG and no task at all: its platform keeps the job running. Layout, one directory per product: From 9046a0f2c5db169a532d104e0bec64107aa60091 Mon Sep 17 00:00:00 2001 From: Speculator55005 <50082482+fas89@users.noreply.github.com> Date: Wed, 21 Oct 2026 11:09:55 +0200 Subject: [PATCH 11/12] fix(validate): --quiet --strict names a retargeted overlay patch beside schema errors too The ERROR (--strict) lines printed only when warnings alone failed the run, so a contract that also had a schema error showed less under --strict than under --quiet alone, and the second failure came as a surprise after the fix. --- fluid_build/cli/validate.py | 5 +++-- tests/test_overlay_keyed_merge.py | 11 +++++++++++ 2 files changed, 14 insertions(+), 2 deletions(-) diff --git a/fluid_build/cli/validate.py b/fluid_build/cli/validate.py index 66a175c49..2fc56268c 100644 --- a/fluid_build/cli/validate.py +++ b/fluid_build/cli/validate.py @@ -1284,7 +1284,8 @@ def _output_text_results(result: ValidationResult, args, logger: logging.Logger) treat_warnings_as_errors = args.strict and has_warnings if has_errors or treat_warnings_as_errors: - if not has_errors: + # Under --quiet the warnings were not listed, errors or not. + if treat_warnings_as_errors and (args.quiet or not has_errors): _explain_strict_failure(result, args) return 1 else: @@ -1292,7 +1293,7 @@ def _output_text_results(result: ValidationResult, args, logger: logging.Logger) def _explain_strict_failure(result: ValidationResult, args: Any) -> None: - """Say why ``--strict`` exits 1 after a summary that reads valid. + """Say why ``--strict`` exits 1: after a summary that reads valid, or under ``--quiet``. ``--quiet`` prints errors only, so there each warning is printed as the error ``--strict`` made it: a count alone does not say what to fix. diff --git a/tests/test_overlay_keyed_merge.py b/tests/test_overlay_keyed_merge.py index f64ef2f8d..5c11e0dc2 100644 --- a/tests/test_overlay_keyed_merge.py +++ b/tests/test_overlay_keyed_merge.py @@ -474,6 +474,17 @@ def test_quiet_strict_names_it_as_the_error(self, tmp_path: Path) -> None: assert result.stdout.startswith("ERROR (--strict): overlay_patch_retargeted: in ") assert (result.stdout + result.stderr).count("overlay_patch_retargeted") == 1 + def test_quiet_strict_names_it_beside_other_errors(self, tmp_path: Path) -> None: + # --strict must not show less than --quiet alone. + _write_product(tmp_path, "0.7.5", PROD_BY_POSITION) + contract = yaml.safe_load((tmp_path / "contract.fluid.yaml").read_text("utf-8")) + contract["bogusTopLevel"] = 1 + (tmp_path / "contract.fluid.yaml").write_text(yaml.safe_dump(contract), "utf-8") + result = _validate(tmp_path, "--quiet", "--strict") + assert result.returncode == 1 + assert "ERROR: root: Additional properties" in result.stdout + assert "ERROR (--strict): overlay_patch_retargeted: in " in result.stdout + def test_strict_json_lists_it(self, tmp_path: Path) -> None: _write_product(tmp_path, "0.7.5", PROD_BY_POSITION) result = _validate(tmp_path, "--strict", "--format", "json") From b88d9e8f62e14b655c7466d1207b1a71dd645ecd Mon Sep 17 00:00:00 2001 From: Speculator55005 <50082482+fas89@users.noreply.github.com> Date: Wed, 21 Oct 2026 11:09:55 +0200 Subject: [PATCH 12/12] docs(changelog): where trigger.type continuous belongs, and the two paths that do not know it --- CHANGELOG.md | 3 +++ 1 file changed, 3 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 2ed522049..b5643c62a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -42,6 +42,9 @@ 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. ### Changed