feat(databricks): amend-and-build runs the dbt job now and waits for it

AGENTS.md

@@ -53,6 +53,8 @@ Every contract is delivered through an 11-stage pipeline. Each stage is a hard g
 
 The `replace` rows describe the native engine. On the OpenTofu engine (aws, gcp, snowflake, confluent) `replace` emits the same module as `amend` and adds the data-loss gate; a table is replaced only where `tofu` plans it, and on GCP its row access policies and table grants are recreated with it (`replace_triggered_by`). A BigQuery table is never recreated to replace its rows: the planner's replace and `fluid rollback` run a `WRITE_TRUNCATE_DATA` query job into the existing table (a `CREATE OR REPLACE TABLE` drops its row access policies and policy tags), and the pre-replace backup is a `CREATE SNAPSHOT TABLE … CLONE`. dbt's `table` materialisation still recreates a governed table (known gap; `docs/governance-parity.md`). See `docs/apply.md`.
 
+On Databricks a dbt build with `execution.runtime.platform: databricks` runs as the job `fluid apply` provisions. `amend-and-build` and `replace-and-build` start a run of the job this apply's OpenTofu state holds (found by its `fluid_product` / `fluid_build` tags), wait for it, and fail with the task's error and the end of dbt's output; `replace-and-build` adds `--full-refresh` for that run only. The job's settings, schedule and pause state are never written; `amend` and `create-only` only provision it, and a continuous job is never started. The wait is capped by `FLUID_DATABRICKS_RUN_TIMEOUT_SECONDS` (an hour, or the job's own limits plus 15 minutes); at the cap, on Ctrl-C and on SIGTERM or SIGHUP (unless ignored, as under `nohup`) the run is cancelled, while a run that cannot be read five times in a row is left alone. To build an estate in lineage order, provision every product in parallel with `--mode amend`, then run `--mode amend-and-build` level by level; the second apply plans no change. Code: `build_runners/_databricks_job_run.py`.
+
 ### Plan-Binding (Terraform-Style "Apply Consumes Exact Plan")
 
 Stage 6 `fluid plan` emits two cryptographic fields in `plan.json`:

CHANGELOG.md

@@ -53,6 +53,22 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Changed
 
+- **`fluid apply --mode amend-and-build` runs a Databricks dbt build now.** A dbt build with
+  `execution.runtime.platform: databricks` runs as the serverless job `fluid apply` provisions.
+  `amend-and-build` and `replace-and-build` now start a run of that job after provisioning
+  and wait for it. They print the run page and each state the run enters, with the reason
+  while it is queued. A failed run fails the apply with the task's error and the end of
+  dbt's output. `replace-and-build` adds `--full-refresh` for that run only. Before, the apply
+  only provisioned the job and said how to start it, so a manual job never ran.
+  - `amend` and `create-only` still only provision, and a continuous job is not started.
+  - The job's settings, schedule and pause state are never changed.
+  - `FLUID_DATABRICKS_RUN_TIMEOUT_SECONDS` caps the wait: an hour, or the job's own timeouts
+    and retries plus 15 minutes. The run is cancelled at the cap and on Ctrl-C, SIGTERM or
+    SIGHUP. A run that cannot be read five times in a row is left as it is, and the apply
+    says so. run-now is retried with the same idempotency token, so a retry never starts a
+    second run.
+  - A missing `databricks-sdk` (the `[databricks]` extra) is refused before anything is
+    applied.
 - **Keys for two or more providers stop the run, even with Ollama running.** With
   `OPENAI_API_KEY` and `ANTHROPIC_API_KEY` both set (or one provider's key and `OLLAMA_HOST`) and
   nothing naming a provider, a run without a TTY skipped both keys and used a running local

CLAUDE.md

@@ -434,6 +434,12 @@ hand-rolling the protocol.
 - **The validate hook** (`agent_skills/hooks/validate_contract.py`) reads the edited path from stdin JSON: hook commands get no `${tool_input.*}` substitution, and an `if` rule is one permission rule (`|` is not OR). One file serves the plugin (`python3 "${CLAUDE_PLUGIN_ROOT}/hooks/validate_contract.py"`) and scaffold-ide (`<python> <installed path>/hooks/validate_contract.py`, written to `.claude/settings.local.json`). **Never `python -m` in a hook or an agent-reachable shim:** it puts the working directory (the agent's project) first on `sys.path`, so a planted `json.py` runs (reproduced in the 2026-11-12 security review); the hook calls the `fluid` console script instead.
 - **`tests/agent_evals/`** is the eval harness (arms no-skill / skill / forge, deterministic grading). Its workspaces live in the system temp dir, never inside a repository: Claude Code loads every `CLAUDE.md` on the way up from the working directory, and a first run with workspaces under `tests/agent_evals/runs/` fed forge-cli's own CLAUDE.md (33.8k tokens) to every arm. The live test is marker `agent_eval` (deselected in `addopts`) plus `FLUID_AGENT_EVAL=1`; `test_harness.py` grades every scenario's reference solution in the default lane. A local DuckDB build that reads a sibling product's output needs `fluid.workspace.yaml` at the repo root. Under `--mode amend` the provider path never binds `consumes[]` and reads a `parameters.inputs` entry without a `format` as CSV (`providers/local/planner.py` fills in `csv` before the extension is looked at); `--mode amend-and-build` binds each `consumes[]` entry the SQL reads by its `exposeId` (`build_runners/_embedded_sql_io.py`) and infers an input's format from its extension. An input with a `format` works in both.
 
+## Recent architectural changes worth knowing (2026-11-12, Databricks job runs)
+
+- **`-and-build` runs a Databricks dbt build now.** The seam: `cli/apply.py` passes `applied_state=[]` to `apply_via_opentofu` (only when `_databricks_job_runs` finds a non-continuous Databricks dbt build), the engine fills it with `runner.tofu_state_resources` after `post_apply` binds masks, and `jobs_in_state` (tags `databricks_jobs.PRODUCT_TAG` / `BUILD_TAG`, never the job name) hands `run_builds_from_args(databricks_jobs=...)` the ids. `build_runners/_databricks_job_run.py` does run-now (idempotency token, `queue.enabled`, tried 3 times with the same token), polls `runs/get` (5 failed reads in a row end the wait without a cancel: the SDK does not repeat a 500/502/504), and on failure reads `runs/get-output` per failed task. Ctrl-C, SIGTERM and SIGHUP cancel the run only while it is waited on. An id-less build's job is tagged by its position (`databricks_jobs.job_build_id`). `databricks_jobs=None` keeps the old "runs on its trigger" report. `_gate_databricks_sdk_for_apply` refuses a missing SDK before `tofu`.
+- **Live-verified on Free Edition (2026-11-12):** a queued run's `runs/get` carries only `status.state: QUEUED` and an empty `state.state_message` (no `queue_details`), so the reason comes from `runs/list?active_only=true`; a dbt task's stdout is in `get-output.logs` with ANSI codes; `get-output.dbt_output.artifacts_link` is a pre-signed URL and is never printed; the deprecated run-now `dbt_commands` is still accepted and overrides the commands for that run only.
+- Pins: `tests/build_runners/test_databricks_job_run.py`, `tests/build_runners/test_databricks_job_builds.py`, `tests/cli/test_apply_databricks_job_runs.py`, and the real-`tofu` `tests/iac/test_iac_databricks_jobs_fake_e2e.py` (`_fake_databricks.py` serves runs).
+
 ## Working style
 
 - **Goal-driven execution.** For non-trivial tasks, state success criteria before implementing and loop until verified. Transform imperatives into tests: "add redactor coverage" → "new pattern is masked by a test assertion"; "fix the bug" → "reproducing test passes first, then fix makes it green".