feat: trigger.type continuous (0.7.6 preview) is never scheduled; --strict fails on a retargeted overlay patch

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/<name>/`, implement `BaseProvider` interface
 - **Adding a template**: Create `fluid_build/templates/<name>.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/<provider>/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.

CHANGELOG.md

@@ -7,6 +7,51 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Upgrade notes
+
+- `fluid validate --env <env> --strict` (on one contract or a whole workspace), `fluid
+  contract-validation --env <env> --strict` and `fluid test --env <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
+  <bundle>.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; with
+  `--quiet` it prints each such warning as `ERROR (--strict): <warning>`.
+- 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
+
+- **`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.
+  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
+
+- 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.
+
 ## [0.21.0] — 2026-10-19
 
 ### Upgrade notes

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`, `properties.language: flink_sql`), gets no DAG and no task
+at all: its platform keeps the job running.
 
 Layout, one directory per product:
 

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_unscheduled_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_unscheduled_build(build):
+            continue
         execution = build.get("execution", {}) or {}
         trigger = execution.get("trigger", {}) or {}
         schedule = trigger.get("schedule")

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()
@@ -1625,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")
@@ -1638,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:

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
 

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", _detail(e.message for e in overlay_issues)))
+
     # Provider connectivity
     # The connection probe runs in provider detection, so it happens
     # under --no-data too.

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: also fail when an overlay patch lands on another entry
+  fluid validate contract.fluid.yaml --env prod --strict
         """,
         formatter_class=argparse.RawDescriptionHelpFormatter,
     )
@@ -302,7 +305,9 @@ def run(args, logger: logging.Logger) -> int:
 
         # Load contract with overlay
         try:
-            contract = load_contract_with_overlay(args.contract, getattr(args, "env", None), logger)
+            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)})
 
@@ -331,6 +336,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)
@@ -355,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
@@ -452,6 +461,23 @@ def run(args, logger: logging.Logger) -> int:
         raise CLIError(1, "cli_unhandled_exception", {"error": str(e)})
 
 
+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.
+    :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
+
+    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]
+
+
 def _handle_list_versions(schema_manager: FluidSchemaManager, args, logger: logging.Logger) -> int:
     """Handle --list-versions flag."""
     try:
@@ -1258,11 +1284,39 @@ 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:
+        # 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:
         return 0
 
 
+def _explain_strict_failure(result: ValidationResult, args: Any) -> None:
+    """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.
+    """
+    if args.quiet:
+        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.
@@ -1377,19 +1431,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}")
@@ -1403,6 +1455,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
 # ---------------------------------------------------------------------------
@@ -1603,13 +1679,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)