fix(apply): refuse an unusable build id before tofu apply

CHANGELOG.md

@@ -132,6 +132,19 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- **`fluid apply --mode amend-and-build` provisioned a product, then refused its id.** The build
+  phase keeps run records, cursors and watermarks under `.fluid/runs/<contract.id>/<build.id>/`,
+  so its ids must start with a letter or underscore, use only letters, digits, `_`, `.` and `-`,
+  and be at most 128 characters. The contract schema allows more (`1orders`, or a longer id), so
+  `fluid validate` and `fluid plan` passed such an id. Only the build runner checked it, and on the
+  OpenTofu engine (AWS, GCP, Snowflake, Confluent, Databricks and IaC plugin clouds) the runner
+  starts after `tofu apply`: the infrastructure was created, and the run then stopped with
+  "Unexpected error". A build mode (`amend-and-build`, `replace-and-build`) now checks
+  `contract.id` and every `builds[].id` before the apply hooks, `tofu` and the runner, on both
+  engines, and refuses with `ERR_APPLY_BUILD_IDENTIFIER_INVALID`, which names the id, says nothing
+  was applied and suggests a rename. A mode without a build phase, and a contract without builds,
+  are not checked, as before.
+
 - **`fluid diff --baseline` reported nothing for column changes in a 0.7.x contract.** It read
   columns from `exposes[].schema`, but a 0.7.x contract keeps them in `exposes[].contract.schema`
   (as does every `fluid init` template), so a removed column, a type change or a new NOT NULL

fluid_build/_error_catalog.py

@@ -265,6 +265,17 @@ def slug_for(event: str) -> str:
         ["Run 'fluid validate <contract>' first to rule out a contract problem"],
         None,
     ),
+    "apply_build_identifier_invalid": (
+        [
+            "Rename the id to fit, for example orders_v1 instead of 1orders, and apply again",
+            (
+                "A renamed id starts over: its builds get new cursors and watermarks under "
+                ".fluid/runs/, and a renamed contract.id gets a new OpenTofu workdir under "
+                ".fluid/iac/<provider>/"
+            ),
+        ],
+        None,
+    ),
     "output_write_failed": (
         ["Check that the --out path is writable and the parent directory exists"],
         None,

fluid_build/build_runners/_ids.py

@@ -37,6 +37,7 @@
 from __future__ import annotations
 
 import re
+from typing import Any, Mapping, Optional
 
 # A permissive regex, identical to ``validate_ident`` in
 # ``providers/_sql_safety``: alphanumerics + dots + dashes + underscores,
@@ -48,7 +49,17 @@
 
 
 class IdentifierViolation(ValueError):
-    """Raised when a contract identifier fails validation."""
+    """Raised when a contract identifier fails validation.
+
+    ``kind`` and ``value`` name the id that failed when the raiser knows it
+    (:func:`validate_identifier` always does), so a caller can report it
+    without parsing the message.
+    """
+
+    def __init__(self, message: str, *, kind: Optional[str] = None, value: Any = None) -> None:
+        super().__init__(message)
+        self.kind = kind
+        self.value = value
 
 
 def validate_identifier(value: str, *, kind: str) -> str:
@@ -65,6 +76,30 @@ def validate_identifier(value: str, *, kind: str) -> str:
             f"{kind} {value!r} is not a valid identifier "
             "(must match ^[A-Za-z_][A-Za-z0-9_.\\-]{0,127}$). This guard "
             "blocks template-injection / path-traversal attacks via "
-            "attacker-controlled contract metadata."
+            "attacker-controlled contract metadata.",
+            kind=kind,
+            value=value,
         )
     return value
+
+
+def validate_build_ids(contract: Mapping[str, Any]) -> None:
+    """Validate the ids a contract's build phase uses: ``contract.id`` and each ``builds[].id``.
+
+    The build runner (``base.run_builds_from_args``) runs this before any
+    build, and ``fluid apply`` runs it before it applies anything, so the two
+    refuse the same ids. A contract with no builds has no build phase and is
+    not checked. An empty or absent id is skipped: it cannot traverse, and the
+    state store falls back to a safe default. A ``builds`` entry that is not a
+    mapping is left to the schema gate.
+
+    Raises :class:`IdentifierViolation` for the first id that fails.
+    """
+    builds = contract.get("builds") or []
+    if not builds:
+        return
+    if contract.get("id"):
+        validate_identifier(contract["id"], kind="contract.id")
+    for build in builds:
+        if isinstance(build, Mapping) and build.get("id"):
+            validate_identifier(build["id"], kind="build.id")

fluid_build/build_runners/base.py

@@ -1393,17 +1393,17 @@ def run_builds_from_args(
     # rather than traversing the filesystem. The parallel pipeline
     # (``cli/_acquisition_stage_ext``) already validates the same fields;
     # this closes the one-sided-guard gap on the runtime path.
-    from ._ids import validate_identifier
+    from ._ids import validate_build_ids
 
     # Only non-empty ids are checked: an absent/empty id cannot traverse
     # (``_build_dir`` falls back to a safe "product" default and
     # ``FileStateStore._confine()`` is the backstop), while a non-empty id
     # like ``../../../../tmp/escape`` is rejected here before any path is made.
-    if contract.get("id"):
-        validate_identifier(contract["id"], kind="contract.id")
-    for _b in builds:
-        if _b.get("id"):
-            validate_identifier(_b["id"], kind="build.id")
+    # ``fluid apply`` runs the same check before it applies anything
+    # (``cli/apply.py::_gate_build_ids_for_apply``), so on the OpenTofu engine
+    # an id is not refused here after ``tofu apply``; this one stays as the
+    # backstop for any caller that skips it.
+    validate_build_ids(contract)
 
     # The running ``fluid apply``'s run report, if any: each build is recorded on it.
     from fluid_build.observability.apply_run import current_apply_run

fluid_build/cli/apply.py

@@ -308,6 +308,44 @@ def _gate_contract_for_apply(contract: Dict[str, Any], logger: logging.Logger) -
     _gate_contract_for_plan_or_apply(contract, logger, command="apply")
 
 
+def _gate_build_ids_for_apply(contract: Dict[str, Any]) -> None:
+    """Refuse, before anything is applied, an id the build phase would refuse.
+
+    The build phase turns ``contract.id`` and ``builds[].id`` into directory
+    names, so it takes a narrower grammar than the schema: ``1orders``, or an
+    id of 129 characters, passes ``fluid validate`` and ``fluid plan``. The
+    build runner checks the ids itself, but on the OpenTofu engine it runs
+    after ``tofu apply``, so such an id was refused once the product was
+    provisioned, as an "Unexpected error". This is the runner's own check
+    (``build_runners._ids.validate_build_ids``), run ahead of the hooks,
+    ``tofu`` and the runner on both engines.
+
+    Raises:
+        CLIError: ``apply_build_identifier_invalid``.
+    """
+    from fluid_build.build_runners._ids import IdentifierViolation, validate_build_ids
+
+    try:
+        validate_build_ids(contract)
+    except IdentifierViolation as exc:
+        raise CLIError(
+            1,
+            "apply_build_identifier_invalid",
+            {
+                "field": exc.kind,
+                "value": repr(exc.value),
+                "hint": (
+                    "The build phase keeps each build's run records, cursors and "
+                    "watermarks under .fluid/runs/<contract.id>/<build.id>/, so both ids "
+                    "must start with a letter or underscore, use only letters, digits, "
+                    "'_', '.' and '-', and be at most 128 characters. The contract schema "
+                    "allows more, so fluid validate and fluid plan accept it. Nothing was "
+                    "applied."
+                ),
+            },
+        ) from exc
+
+
 def _log_apply_failure(logger: logging.Logger, start_time: float, error: Exception) -> None:
     """Record a pre-dispatch ``apply_contract`` failure for the structured log."""
     log_operation_failure(logger, "apply_contract", error=error, duration=time.time() - start_time)
@@ -1923,8 +1961,13 @@ def run(args, logger: logging.Logger) -> int:
     # IaC plugin clouds) and both engines' ``needs_build`` returns used to
     # skip it, so a schema-invalid contract reached ``tofu`` or the build
     # runner and could exit 0. One call per run, on the one loaded copy.
+    # A build mode also checks the ids the build phase turns into paths, which
+    # the schema admits more of: on the OpenTofu engine the runner would only
+    # refuse one after ``tofu apply``.
     try:
         _gate_contract_for_apply(source_contract, logger)
+        if needs_build(resolved_mode):
+            _gate_build_ids_for_apply(source_contract)
     except CLIError as exc:
         _log_apply_failure(logger, start_time, exc)
         raise