feat: durable dbt state per environment, and --defer onto it (#10975)

* feat: durable dbt state per environment, and `--defer` onto it

`dbt retry` worked off two artifacts and only one was durable: `dbt_run_state`
holds `run_results.json` keyed by principal, and the manifest lived on
worker-local disk under a four-generation cache. That is enough to resume the
last run and nothing else — the next run of a project usually lands on a worker
holding neither artifact — so deferral had nothing to read.

Adds `dbt_environment_state`: one row per (workspace, script path, environment),
holding `manifest.json` and `run_results.json` from the last successful run, with
the blob inline under `DBT_STATE_INLINE_MAX_BYTES` and in the workspace's object
storage above it. Environment is the warehouse, the target, and the database and
schema they resolve to, so a repointed warehouse or a moved schema reads as an
environment nothing has published rather than as state whose relation names no
longer fit.

A run publishes it when its graph becomes what the script owns and it succeeded
— the same condition, and the same reason: an invocation that scoped its own
model set describes where the caller put those relations, not where the
project's models live.

`defer` is a `build` command-block field defaulting to the descriptor's own, and
the state is materialised into the job directory for `--defer --state`. The
retry path already did that materialisation for `dbt retry`; both go through one
`write_state_dir` now.

`--state` is also where `dbt retry` reads the run it resumes, so a retry on
dbt-core 1.x takes `--defer-state` instead, and one on an engine without that
flag is refused before the build rather than rebuilding its nodes with every
unbuilt `ref()` resolving into the schema this run writes into.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Ned2pmRJwB3GpenEcrA9TF

* fix: address the local review of the dbt environment state

The oversized-artifact home moves from the workspace's object storage to the
instance's, where every other internal worker artifact already lives. The
workspace bucket is the one members read and write through `job_helpers/*` with
a caller-supplied key and only `volumes/` is reserved there, so a manifest under
it is one any member could replace — and the next deferring run would hand dbt
an attacker-chosen `defer_relation` for every unbuilt `ref()` while holding the
script's warehouse credentials.

The environment key takes the target dbt actually runs rather than the
descriptor's `profile.target`, which is absent whenever the target is inherited
from the workspace warehouse or the project's own `profiles.yml` — filing every
inherited target under one empty name, while a `target.name` macro decides where
a model is built. `write_profiles` returns a named struct now that it resolves
one more thing.

Publishing takes the row's lock before uploading, so two publishers of one
environment cannot interleave their uploads and leave one run's manifest beside
another's results, and carries the live-dbt-script guard the retry state already
had, so a job finishing after its script was renamed, archived or deleted cannot
recreate state at a path for whatever is created there next.

A rename now clears the environment state instead of moving it: an oversized
artifact's key is derived from the path, so a moved row would keep pointing at a
key a script created at the old path publishes over.

A build recovered by the automatic in-job node retry publishes its manifest
without results — `run_results.json` is then the retry's, naming only the nodes
it redid — and the refusal for an environment with nothing published names the
runs that cannot publish rather than suggesting a run that would not help.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix: serialize dbt state publishers on an advisory lock

The row lock only serializes publishers once a row exists, and the first
publish of an environment — two runs of a newly deployed script — is exactly
when two of them are most likely to race and interleave their uploads.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix: make dbt state publication atomic and bind it to the version that ran

Every publication now writes its own object keys and the row switches to them in
one statement, so an upload never overwrites an artifact the committed row still
names: a run failing between its two uploads, or between them and its row, leaves
the state pointing at the pair it already had. The objects a commit displaces are
dropped afterwards — never before, since a reader that has already read the row
is about to fetch them — and a reader that loses that race re-reads the row once
rather than reporting a state that is there. What a publication uploaded and then
could not commit is dropped on the way out.

The write's guard names the VERSION rather than the path: the live dbt script
there must be the one this job ran, or a later version of it. "Some live dbt
script is here" is also satisfied by a script created at a path this one was
renamed away from, and this job's manifest would then become that project's
deferral state. A preview names no version and so publishes nothing.

A `show` defers too. It compiles the model it previews, so a model whose upstream
this environment built and this run did not is exactly the case a deferral exists
for, and every engine takes the flags on it.

Three comments said "the workspace's object storage" where the code deliberately
uses the instance's, which is the whole security argument; `mib()` labelled MiB
values MB.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix: hold the script row across a dbt state publication, and let a rename move it

The version guard read `script` without a lock, so lifecycle cleanup could find
no environment row to clear, finish, and leave this transaction to commit state
at a path a new script goes on to occupy. It now holds that row (`FOR SHARE`) for
the rest of the publication — taken before the sidecar, the order every other dbt
writer takes — and the artifacts are uploaded before the transaction, so the lock
covers the row work rather than a network round trip.

A commit that reports an error may still have committed: what was lost can be the
acknowledgement. Dropping this run's objects then leaves the committed row naming
objects that are gone, so an orphan is the cheaper side to take.

A failed second upload left the manifest it had already written behind; it is
dropped now.

Per-publication keys retired the reason a rename cleared the environment state
rather than moving it: the path is only a prefix, and the row is what names an
artifact, so a script created at the old path can no longer publish over a moved
row. The rename moves both halves again.

`dbt ls` gets the deferral flags too, without which a `result:` selector — which
reads `run_results.json` out of the state directory, and which `select` passes to
dbt verbatim — fails before the build that would have honoured it.

Also: the migration was the last site describing the workspace's object storage
rather than the instance's, `publication_lock` folded 32 bits where it claimed
64, and `ResolvedProfile` had taken `write_profiles`'s doc block.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix: a deferring dbt run never publishes the state it read

`publishes_ownership` reads the CALLER's overrides, so a descriptor that already
narrows `select` needs none and a run of it with `defer: true` published. A
deferring run built some of the relations its manifest names and resolved the
rest out of the state it read, so recording that manifest claims relations
nothing built — and a model renamed since is recorded under a name only a full
build creates, breaking every later deferral until one repairs it.

Also: `publication_lock` parsed 16 hex digits as `i64`, which overflows for every
digest with the top bit set — half of them — collapsing those environments onto
one advisory key; and a failure to open the transaction returned without dropping
the objects already uploaded.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix: only a deployed dbt run publishes state, and key its objects per execution

A preview carries a caller-supplied `script_hash` into `runnable_id`
(`run_preview_script`), so the version guard alone let anyone who may run a job
publish arbitrary content as a deployed script's deferral state. The job's KIND
is checked beside it now. Verified: a preview submitted with the deployed path
and hash builds and leaves the row untouched.

Object keys carry a per-execution nonce. Zombie recovery re-runs a job under its
own id, so keyed on that alone a second attempt overwrote the objects the first
attempt's committed row still named, then read those same keys back as displaced
and dropped them — leaving the row unreadable. The displaced set is also filtered
against this publication's own keys, so the invariant is stated rather than
re-derived from the key format.

A project-owned `profiles.yml` that templates its schema or database is refused a
deferral: dbt renders those and Windmill does not, so two renderings resolve to
one `relation_root` and would share one environment key. Plainly absent is left
alone — that is the adapter's default, which does not move.

The deferral log line now says the run publishes no state of its own, which was
otherwise invisible.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix: a templated profile location publishes no dbt state either, on every path

A `dbt_profile` resource is one block of the user's own `profiles.yml` copied
through unchanged, and `profile.schema` is written as given, so either can carry
a template dbt renders and this runtime does not — exactly as a project-owned
file can. Only the project-owned path detected it.

And the refusal now covers publication as well as deferral: a published template
would sit under a key a literal profile shares, so de-templating later would make
that stale manifest readable as the new location's.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix: recognise Jinja statement blocks as a rendered dbt profile location

dbt renders a profile through Jinja, so `{% if env_var('ENV') == 'prod' %}…{% endif %}`
moves a schema exactly as an `env_var()` substitution does — and only `{{` was
detected, so such a profile published and deferred under one environment key for
every rendering. One predicate now serves both profile paths, with a test for
each delimiter.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix: a dbt state read outruns successive publications rather than one

The loader re-read once, which answers a single publication overtaking it: a
reader takes no lock and the advisory lock is released before the displaced
objects are dropped, so back-to-back publications could each overtake the same
read and the second was reported as a missing object. It now re-reads for as long
as the row keeps MOVING, bounded, and reports only when an unmoved row's objects
are genuinely gone.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* docs: a dbt state read outruns successive publications, not one

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* docs: name both ways a dbt state read can fail

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix: length-prefix the dbt environment key's components

A dbt target name and a schema are both the user's own strings, so joining them
on `|` let one component spell another tuple's key: `prod|analytics` + `scratch`
and `prod` + `analytics|scratch` were one environment, and a profile moving
between them read as the same one rather than as one nothing has published — the
collision the key exists to prevent. The schema and database are also taken apart
now rather than through `relation_root`'s own join, so neither can absorb the
other's delimiter.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix: name the dbt environment in words where a message shows it

The key is length-prefixed for storage, which is not something to put in front of
a caller: the "nothing published yet" refusal now reads "warehouse `main`, target
`prod`, relations in `dbt_wh_defer.analytics`". The worked example of the encoding
also miscounted a component.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix: delete a script version in the transaction that cleans up after it

`delete_script_by_hash` soft-deleted through the pool, committing before the
cleanup that follows it in `tx`. In that window the path has no live version, so
a concurrent deploy can take it — and `clear_dbt_script_state_if_path_retired`
then finds that new script live, keeps the deleted project's dbt state, and
leaves the replacement able to defer through its manifest. The update moves into
the same transaction, which is what `archive_script_by_hash` beside it already
does.

The retirement guard itself was pinned by nothing: the existing test moved the
only row away before calling the conditional clear, so it could not fail.
`state_goes_only_once_no_live_version_is_left` covers both directions — a second
live version keeps the state, the last one leaving takes it — and fails if the
predicate is inverted.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix: archive a script by path in the transaction that cleans up after it

The last of the four routes still writing outside its own cleanup transaction.
Archived on its own, a cleanup that then fails leaves dbt state at a path no live
version occupies, and whatever is created there next can defer through it. The
by-hash archive and both deletes already take their write in `tx`; this makes the
set uniform.

Two comments beside those clears still called the state the RETRY state alone,
which the rename made false — they cover both halves now — and the merged
verification list had two `11.`, main's #10978 having inserted an item above it.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* feat: refuse a dbt state selector the engines resolve inconsistently

`state:`, `result:` and `source_status:` selectors resolve against the
artifacts in `--state`, which only a deferring run is handed. The engines
disagree about what happens without one, and two of the three disagree
silently: dbt-core 1.x raises, but dbt-sa-cli 2.x and fusion read a missing
state as an empty one and exit 0, so `state:modified` builds nothing and
`state:new` builds the whole project, each reporting success.

Refuse them up front instead, naming `defer`. From the descriptor they are
refused outright, since that selection also decides which nodes the script
owns and the deploy resolves it with no state at all.

`source_status:` is refused under any setting: it compares `sources.json`,
which no run publishes here.

A caller's selection is now allowed to match nothing, which is what
`state:modified+` returns when nothing changed since the published state. It
is stored as that run's own snapshot and never becomes what the script owns,
so the ownership-wipe the refusal guarded against cannot happen. The
descriptor's selection still may not.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix: refuse a dbt result selector the published state cannot answer

Round 18 findings.

Codex P1: `defer` alone was enough to allow a `result:` selector, but a build
recovered by node retry publishes a manifest with no `run_results.json` — the
only file such a selector reads. dbt-core then raises an internal error and the
Rust engines match nothing and exit 0. The deferral now reports whether the
state carries results, and a `result:` selection against one that does not is
refused, naming the run that published it.

Claude P2: a `parse` returns before `defer` is read, so its deferral is always
absent and "turn `defer` on" was advice that led nowhere. The check now
distinguishes a run that could defer from a command that never does, and the
parse path says so.

Codex P2 / Claude P2: the roadmap still listed `state:modified` as out of scope
while the same file documented it as working. Narrowed both that line and the
scope list to the slim-CI work that genuinely remains.

Also pins the invariant the relaxed empty-selection guard rests on: an
overridden selection must not publish ownership, or an empty caller selection
would wipe the script's graph.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix: exempt an empty dbt selection by method, not by who chose it

Round 19 findings.

Codex P1: the empty-selection exemption keyed on whether the caller overrode
the selection, so a misspelled model name resolved to nothing, passed the guard
and reported a build that did its work. Key it on the selector instead: only a
`state:` or `result:` method may match nothing, its empty answer being a real
one. Every other selection matching nothing is refused again, from a run as
from the descriptor, each with the message that applies to it.

Claude P2: the spec still described a node-retry-recovered publication as one
where `result:` selectors merely lose their input, which the previous commit
stopped being true, and the section stating the selector rules recorded neither
the `result:`-without-results refusal nor the `parse` one. Both written down.

Also drops the refusal's claim that the publishing run WAS recovered by node
retry: an unreadable file reaches the same absent-results state, and the remedy
is the same either way.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* docs: record why an exempted empty dbt selection cannot wipe the graph

The safety argument left with the origin-based condition it justified. Under
the method-based one it is a consequence of the descriptor refusal in
check_state_selectors, two hops from this site, so state it here: relaxing that
refusal would let a descriptor-narrowed `state:modified+` reach the exemption
and be ingested as owning nothing.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>

backend/.sqlx/query-3fc12582cbae2ebc77ecfcaeed4bce43c749a44cbc4e7b8f719bf2a598ce57f3.json

@@ -0,0 +1,21 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "INSERT INTO dbt_environment_state (workspace_id, script_path, environment, job_id,\n                                                manifest, manifest_key, run_results,\n                                                run_results_key, updated_at)\n             VALUES ($1, $2, $3, $4, $5, $6, $7, $8, now())\n             ON CONFLICT (workspace_id, script_path, environment) DO UPDATE SET\n               job_id = EXCLUDED.job_id, manifest = EXCLUDED.manifest,\n               manifest_key = EXCLUDED.manifest_key, run_results = EXCLUDED.run_results,\n               run_results_key = EXCLUDED.run_results_key, updated_at = now()",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Varchar",
+        "Text",
+        "Uuid",
+        "Text",
+        "Text",
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "3fc12582cbae2ebc77ecfcaeed4bce43c749a44cbc4e7b8f719bf2a598ce57f3"
+}

backend/.sqlx/query-446909206f289fc3f9201a6ad025cc73eef1e3d01825031ba263881b7cfd5ed4.json

@@ -0,0 +1,14 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "DELETE FROM script WHERE workspace_id = $1 AND hash = 1",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "446909206f289fc3f9201a6ad025cc73eef1e3d01825031ba263881b7cfd5ed4"
+}

backend/.sqlx/query-4ce90ff818e5058a7e31da24abd938eb135d6eb9714ca115359221d7c02861f0.json

@@ -0,0 +1,17 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "INSERT INTO dbt_environment_state (workspace_id, script_path, environment, job_id,\n                                            manifest)\n         SELECT $1::varchar, $2::varchar, 'main||analytics|wh'::text, $3::uuid, '{}'::text\n          WHERE EXISTS (SELECT 1 FROM script\n                         WHERE workspace_id = $1 AND path = $2\n                           AND deleted = false AND archived = false\n                           AND language = 'dbt'\n                           AND (hash = $4 OR $4 = ANY(parent_hashes)))\n         ON CONFLICT (workspace_id, script_path, environment) DO UPDATE SET job_id = EXCLUDED.job_id",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Varchar",
+        "Uuid",
+        "Int8"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "4ce90ff818e5058a7e31da24abd938eb135d6eb9714ca115359221d7c02861f0"
+}

backend/.sqlx/query-58ec340c78af046a40296b15543beeedc63b4a941e5b17ab7bb5d3c259f05147.json

@@ -0,0 +1,24 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT 1 FROM script\n              WHERE workspace_id = $1 AND path = $2\n                AND deleted = false AND archived = false AND language = 'dbt'\n                AND (hash = $3 OR $3 = ANY(parent_hashes))\n              FOR SHARE",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "?column?",
+        "type_info": "Int4"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text",
+        "Int8"
+      ]
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "58ec340c78af046a40296b15543beeedc63b4a941e5b17ab7bb5d3c259f05147"
+}

backend/.sqlx/query-5c7260785ebcece2ddb04dc85d09b6a36c50d70e20c5a10310f75a39930db521.json

@@ -0,0 +1,16 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "INSERT INTO dbt_environment_state (workspace_id, script_path, environment, job_id,\n                                            manifest)\n         VALUES ($1, $2, 'main||analytics|wh', $3, '{}')",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Varchar",
+        "Uuid"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "5c7260785ebcece2ddb04dc85d09b6a36c50d70e20c5a10310f75a39930db521"
+}

backend/.sqlx/query-7f421bdf3dc4d47b36457af703ce69ef1e0784e9f25ada27e1b0f5cc0944e9ba.json

@@ -0,0 +1,16 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE dbt_environment_state SET script_path = $3\n          WHERE workspace_id = $1 AND script_path = $2",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text",
+        "Varchar"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "7f421bdf3dc4d47b36457af703ce69ef1e0784e9f25ada27e1b0f5cc0944e9ba"
+}

backend/.sqlx/query-8165a447f458d62f7bafb9827d87d192f5e926b932c3dd0e1bef4822cf722630.json

@@ -0,0 +1,48 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT job_id, manifest, manifest_key, run_results, run_results_key\n               FROM dbt_environment_state\n              WHERE workspace_id = $1 AND script_path = $2 AND environment = $3",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "job_id",
+        "type_info": "Uuid"
+      },
+      {
+        "ordinal": 1,
+        "name": "manifest",
+        "type_info": "Text"
+      },
+      {
+        "ordinal": 2,
+        "name": "manifest_key",
+        "type_info": "Text"
+      },
+      {
+        "ordinal": 3,
+        "name": "run_results",
+        "type_info": "Text"
+      },
+      {
+        "ordinal": 4,
+        "name": "run_results_key",
+        "type_info": "Text"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": [
+      false,
+      true,
+      true,
+      true,
+      true
+    ]
+  },
+  "hash": "8165a447f458d62f7bafb9827d87d192f5e926b932c3dd0e1bef4822cf722630"
+}

backend/.sqlx/query-a06e1d9f6f95e4c4c2b98310ebddcc9d963cc033582bf2e945e8bf3a301b4247.json

@@ -0,0 +1,22 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT pg_advisory_xact_lock($1)",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "pg_advisory_xact_lock",
+        "type_info": "Void"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Int8"
+      ]
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "a06e1d9f6f95e4c4c2b98310ebddcc9d963cc033582bf2e945e8bf3a301b4247"
+}

backend/.sqlx/query-b820bf7b0a93476fb7762e8ff2819b2c2b7df2788c9eb0115abc60ea7b733d64.json

@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "DELETE FROM dbt_environment_state WHERE workspace_id = $1 AND script_path = $2\n           AND NOT EXISTS (SELECT 1 FROM script\n                            WHERE workspace_id = $1 AND path = $2\n                              AND deleted = false AND archived = false)",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "b820bf7b0a93476fb7762e8ff2819b2c2b7df2788c9eb0115abc60ea7b733d64"
+}

backend/.sqlx/query-d2c02f1c7f4555fa849f5cd86169eb8e43d384084b68dc1bbb3b961206ff4b6c.json

@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "DELETE FROM dbt_environment_state WHERE workspace_id = $1 AND script_path = $2",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "d2c02f1c7f4555fa849f5cd86169eb8e43d384084b68dc1bbb3b961206ff4b6c"
+}

backend/.sqlx/query-d8b7241518ce0822288fb123b48205427cc50af031a84d334f65f870d86302ac.json

@@ -0,0 +1,48 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT job_id, manifest, manifest_key, run_results, run_results_key\n           FROM dbt_environment_state\n          WHERE workspace_id = $1 AND script_path = $2 AND environment = $3",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "job_id",
+        "type_info": "Uuid"
+      },
+      {
+        "ordinal": 1,
+        "name": "manifest",
+        "type_info": "Text"
+      },
+      {
+        "ordinal": 2,
+        "name": "manifest_key",
+        "type_info": "Text"
+      },
+      {
+        "ordinal": 3,
+        "name": "run_results",
+        "type_info": "Text"
+      },
+      {
+        "ordinal": 4,
+        "name": "run_results_key",
+        "type_info": "Text"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": [
+      false,
+      true,
+      true,
+      true,
+      true
+    ]
+  },
+  "hash": "d8b7241518ce0822288fb123b48205427cc50af031a84d334f65f870d86302ac"
+}

backend/.sqlx/query-f5e061eed38d483980ee3691aea821dde2c093bb926b77ec3f99a1c0d9283a14.json

@@ -0,0 +1,23 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT count(*) FROM dbt_environment_state WHERE workspace_id = $1 AND script_path = $2",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "count",
+        "type_info": "Int8"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "f5e061eed38d483980ee3691aea821dde2c093bb926b77ec3f99a1c0d9283a14"
+}

backend/.sqlx/query-f963ea31d1744ff0d76ff86140f01cca0bcbc832f01eb3340050504b8b3875a2.json

@@ -0,0 +1,30 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT manifest_key, run_results_key FROM dbt_environment_state\n              WHERE workspace_id = $1 AND script_path = $2 AND environment = $3",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "manifest_key",
+        "type_info": "Text"
+      },
+      {
+        "ordinal": 1,
+        "name": "run_results_key",
+        "type_info": "Text"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": [
+      true,
+      true
+    ]
+  },
+  "hash": "f963ea31d1744ff0d76ff86140f01cca0bcbc832f01eb3340050504b8b3875a2"
+}

backend/.sqlx/query-ffde6e45841090440bc99e22788d313db5945faa0b93c127833e44d38665a392.json

@@ -0,0 +1,30 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT manifest_key, run_results_key FROM dbt_environment_state\n          WHERE workspace_id = $1 AND script_path = $2 AND environment = $3",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "manifest_key",
+        "type_info": "Text"
+      },
+      {
+        "ordinal": 1,
+        "name": "run_results_key",
+        "type_info": "Text"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": [
+      true,
+      true
+    ]
+  },
+  "hash": "ffde6e45841090440bc99e22788d313db5945faa0b93c127833e44d38665a392"
+}

backend/migrations/20260904135713_dbt_environment_state.down.sql

@@ -0,0 +1 @@
+DROP TABLE IF EXISTS dbt_environment_state;

backend/migrations/20260904135713_dbt_environment_state.up.sql

@@ -0,0 +1,52 @@
+-- The dbt state one project last built into one environment: the `manifest.json`
+-- (and the `run_results.json` beside it) that `dbt --defer --state <dir>` resolves
+-- an unbuilt `ref()` through.
+--
+-- Separate from `dbt_run_state`, which answers a different question. That one is
+-- keyed by the executing principal and holds the LAST run whatever its outcome,
+-- so `dbt retry` can resume its failures; this one is keyed by environment and
+-- holds the last SUCCESSFUL run, because a relation a later run defers to has to
+-- exist. Merging them would make a retry resume a run that is not the last one,
+-- or a deferral point at relations a failed run never wrote.
+CREATE TABLE IF NOT EXISTS dbt_environment_state (
+  workspace_id VARCHAR(50) NOT NULL REFERENCES workspace(id) ON DELETE CASCADE ON UPDATE CASCADE,
+  script_path  VARCHAR(255) NOT NULL,
+  -- The workspace warehouse, the dbt target, and the database and schema that
+  -- target resolves to. All four, because deferring is resolving a relation
+  -- NAME: a repointed warehouse or a moved schema makes the stored manifest
+  -- describe relations that are not where this run would look for them, and the
+  -- run has no other way to notice. A move therefore reads as an environment
+  -- with no state yet rather than as state that silently no longer fits.
+  --
+  -- TEXT rather than VARCHAR(255): a project bringing its own `profiles.yml`
+  -- spells its own schema and database, so the length is the project's.
+  environment  TEXT NOT NULL,
+  -- The run that published it, so a deferring run can say what it deferred to.
+  job_id       UUID NOT NULL,
+  -- Exactly one home each. A manifest grows with the project and passes a few
+  -- hundred KB on a handful of models, so a large one goes to the INSTANCE's
+  -- object storage and this row keeps the key; a small one stays here, where it
+  -- costs no round trip and works on an instance that has configured no storage
+  -- at all. The instance's and not the workspace's, because a member can write
+  -- the workspace bucket under a key of their choosing, and a manifest is what a
+  -- later run resolves every unbuilt `ref()` through. `run_results.json` is a
+  -- tenth of the size and takes the same two homes rather than a rule of its own.
+  manifest        TEXT,
+  manifest_key    TEXT,
+  run_results     TEXT,
+  run_results_key TEXT,
+  updated_at   TIMESTAMPTZ NOT NULL DEFAULT now(),
+  PRIMARY KEY (workspace_id, script_path, environment),
+  CONSTRAINT dbt_environment_state_manifest_one_home
+    CHECK (num_nonnulls(manifest, manifest_key) = 1),
+  CONSTRAINT dbt_environment_state_run_results_one_home
+    CHECK (num_nonnulls(run_results, run_results_key) <= 1)
+);
+
+-- No age sweep, unlike the per-run graph rows next door: this table holds one
+-- row per script per environment and replaces it in place, so it does not grow
+-- with runs, and its reader is every later run of that script — a project that
+-- runs monthly must still find last month's state. It goes with the script
+-- instead, alongside `dbt_run_state`.
+GRANT ALL ON dbt_environment_state TO windmill_user;
+GRANT ALL ON dbt_environment_state TO windmill_admin;

backend/parsers/windmill-parser-yaml/src/dbt.rs

@@ -58,6 +58,18 @@ impl DbtEngine {
     pub fn emits_node_events(&self) -> bool {
         matches!(self, DbtEngine::DbtCore1x)
     }
+
+    /// Whether the engine has `--defer-state`, the deferral-only half of
+    /// `--state`.
+    ///
+    /// It matters on one command. `dbt retry` reads the run it resumes from
+    /// `--state`, so an engine with only that flag cannot be told to defer and
+    /// to resume from the job's own results at once: handed the deferral's
+    /// directory, it resumes the all-green run stored there and rebuilds
+    /// nothing. Only dbt-core 1.x separates the two.
+    pub fn has_defer_state_flag(&self) -> bool {
+        matches!(self, DbtEngine::DbtCore1x)
+    }
 }
 
 /// How the warehouse connection is supplied. Both paths are supported
@@ -137,6 +149,16 @@ pub struct DbtDescriptor {
     pub threads: Option<u32>,
     #[serde(default)]
     pub full_refresh: bool,
+    /// Resolve a `ref()` a run does not build through the state the last
+    /// successful run of this environment published, rather than through the
+    /// schema that run writes into.
+    ///
+    /// Only the default for the `build` block's own `defer`, since the choice is
+    /// per run: the run that publishes an environment's state and the run that
+    /// defers to it are two invocations of ONE script (decision 6), so a project
+    /// that could only defer by descriptor could never populate what it reads.
+    #[serde(default)]
+    pub defer: bool,
     /// Automatic in-job retry of the nodes a build failed on.
     ///
     /// dbt already confines a failure to its own subtree, and `dbt retry`
@@ -258,6 +280,7 @@ pub const RESERVED_ARG_NAMES: &[&str] = &[
     "exclude",
     "vars",
     "full_refresh",
+    "defer",
     "dbt_command",
     "dbt_retry_job",
     "model",
@@ -345,15 +368,26 @@ fn command_variants(d: &DbtDescriptor) -> Vec<(&'static str, Vec<Arg>)> {
             "build",
             selection()
                 .into_iter()
-                .chain([Arg {
-                    name: "full_refresh".to_string(),
-                    otyp: None,
-                    typ: Typ::Bool,
-                    has_default: true,
-                    default: Some(serde_json::json!(d.full_refresh)),
-                    oidx: None,
-                    otyp_inferred: false,
-                }])
+                .chain([
+                    Arg {
+                        name: "full_refresh".to_string(),
+                        otyp: None,
+                        typ: Typ::Bool,
+                        has_default: true,
+                        default: Some(serde_json::json!(d.full_refresh)),
+                        oidx: None,
+                        otyp_inferred: false,
+                    },
+                    Arg {
+                        name: "defer".to_string(),
+                        otyp: None,
+                        typ: Typ::Bool,
+                        has_default: true,
+                        default: Some(serde_json::json!(d.defer)),
+                        oidx: None,
+                        otyp_inferred: false,
+                    },
+                ])
                 .collect(),
         ),
         (
@@ -546,7 +580,9 @@ fn property_of(arg: &Arg) -> serde_json::Value {
             ),
             "select" => Some(
                 "dbt selection syntax, e.g. `tag:nightly`, `stg_orders+`, \
-                 `config.materialized:incremental`. Empty runs the descriptor's own selection.",
+                 `config.materialized:incremental`. `state:modified+` and `result:error+` \
+                 compare against the state a previous run published, so they need `defer` on. \
+                 Empty runs the descriptor's own selection.",
             ),
             "exclude" => Some("Nodes to leave out of the selection above, same syntax."),
             "vars" => Some(
@@ -554,6 +590,11 @@ fn property_of(arg: &Arg) -> serde_json::Value {
                  exist makes this run store its own graph rather than the deployed one.",
             ),
             "full_refresh" => Some("Rebuild incremental models from scratch instead of appending."),
+            "defer" => Some(
+                "Resolve a `ref()` this run does not build to the relation the last successful \
+                 run of this warehouse and target published, instead of to the schema this run \
+                 writes into.",
+            ),
             "model" => Some(
                 "The model to preview, by name — `stg_orders`, or `my_package.stg_orders` when \
                  two packages share a name. Any dbt selector resolving to ONE node works.",
@@ -687,8 +728,15 @@ full_refresh: true
         };
 
         let (build, build_args) = of("build");
-        assert_eq!(build_args, ["exclude", "full_refresh", "select", "vars"]);
+        assert_eq!(
+            build_args,
+            ["defer", "exclude", "full_refresh", "select", "vars"]
+        );
         assert_eq!(build["properties"]["full_refresh"]["type"], "boolean");
+        // `defer` is a per-run toggle rather than a descriptor-only setting: the
+        // run that publishes an environment's state and the run that defers to
+        // it are two invocations of ONE script.
+        assert_eq!(build["properties"]["defer"]["type"], "boolean");
         // Defaults come from the descriptor, so an untouched run reproduces it.
         assert_eq!(
             build["properties"]["select"]["default"],

backend/summarized_schema.txt

@@ -74,6 +74,8 @@ dbt_edge: workspace_id(char), script_path(char), script_hash(bigint), job_id(uui
   FK: (workspace_id) -> workspace(id), (workspace_id, script_hash) -> script(workspace_id, hash)
 dbt_graph_snapshot: workspace_id(char), script_path(char), script_hash(bigint), job_id(uuid), digest(text), relation_root_at_last_ingest(text), ingested_at(ts), permissioned_as(char)
   FK: (workspace_id) -> workspace(id), (workspace_id, script_hash) -> script(workspace_id, hash)
+dbt_environment_state: workspace_id(char), script_path(char), environment(text), job_id(uuid), manifest(text), manifest_key(text), run_results(text), run_results_key(text), updated_at(ts)
+  FK: (workspace_id) -> workspace(id)
 dbt_node: workspace_id(char), script_path(char), script_hash(bigint), job_id(uuid), unique_id(text), resource_type(text), name(text), asset_path(text), materialized(text), materialize_strategy(text), unique_key(text), tags(text[]), description(text), test_kind(text), test_column(text), test_args(jsonb), severity(text), attached_node(text), columns(jsonb), freshness(jsonb), raw_code(text), original_file_path(text), ingested_at(ts)
   FK: (workspace_id) -> workspace(id), (workspace_id, script_hash) -> script(workspace_id, hash)
 dbt_run_progress: workspace_id(char), job_id(uuid), asset_kind(asset_kind), asset_path(char), status(materialization_status), row_count(bigint), error(text), updated_at(ts)

backend/windmill-api-scripts/src/scripts.rs

@@ -2454,27 +2454,28 @@ async fn create_script_internal<'c>(
     // while its own finished runs still render from them. Clearing by path
     // would empty those run pages for good.
     if ns.language != ScriptLang::Dbt {
-        // The saved retry state does go: nothing regenerates it, it is keyed by
-        // path alone, and it carries one user's failed invocation and its
-        // arguments. No dbt version is live at this path any more to resume it.
-        windmill_common::dbt_manifest::clear_dbt_run_state(&mut tx, &w_id, &ns.path).await?;
+        // The saved run and environment state do go: nothing regenerates them,
+        // both are keyed by path alone, and they carry one user's failed
+        // invocation with its arguments and the project's own manifest. No dbt
+        // version is live at this path any more to resume or defer to.
+        windmill_common::dbt_manifest::clear_dbt_script_state(&mut tx, &w_id, &ns.path).await?;
     }
     if let Some(ref old) = p_path_opt {
         if old != &ns.path {
             clear_script_triggers(&mut *tx, &w_id, old, AssetUsageKind::Script).await?;
             clear_static_asset_usage(&mut *tx, &w_id, old, AssetUsageKind::Script).await?;
-            // The saved retry state travels rather than being cleared: nothing
+            // The saved state travels rather than being cleared: nothing
             // regenerates it, so dropping it would throw away a resumable
-            // failure for what is only a rename. Only while the destination is
-            // still dbt — a rename that also converts the language would
-            // otherwise reinstate at the new path the state the branch above
-            // just cleared, leaving one user's arguments and results under a
-            // path no dbt script occupies.
+            // failure and every deferral until the next full run, for what is
+            // only a rename. Only while the destination is still dbt — a rename
+            // that also converts the language would otherwise reinstate at the
+            // new path the state the branch above just cleared, leaving one
+            // user's arguments and results under a path no dbt script occupies.
             if ns.language == ScriptLang::Dbt {
-                windmill_common::dbt_manifest::move_dbt_run_state(&mut tx, &w_id, old, &ns.path)
+                windmill_common::dbt_manifest::move_dbt_script_state(&mut tx, &w_id, old, &ns.path)
                     .await?;
             } else {
-                windmill_common::dbt_manifest::clear_dbt_run_state(&mut tx, &w_id, old).await?;
+                windmill_common::dbt_manifest::clear_dbt_script_state(&mut tx, &w_id, old).await?;
             }
         }
     }
@@ -3712,7 +3713,11 @@ async fn archive_script_by_path(
         path,
         &w_id
     )
-    .fetch_one(&db)
+    // In the SAME transaction as the cleanup below, as the by-hash routes are:
+    // committed on its own, a cleanup that then fails leaves dbt state at a path
+    // no live version occupies, for whatever is created there next to defer
+    // through.
+    .fetch_one(&mut *tx)
     .await
     .map_err(|e| Error::internal_err(format!("archiving script in {w_id}: {e:#}")))?;
 
@@ -3720,9 +3725,10 @@ async fn archive_script_by_path(
     // The graph stays: the pinned read resolves versions through a CTE that
     // already skips archived rows, so it stops answering for current relations
     // either way, while deleting it would empty the Models panel of every
-    // completed run of the project. Retry state does go — nothing may resume a
-    // script that is no longer live.
-    windmill_common::dbt_manifest::clear_dbt_run_state(&mut tx, &w_id, path).await?;
+    // completed run of the project. The saved run and environment state do go —
+    // nothing may resume a script that is no longer live, and nothing may defer
+    // through what it last built.
+    windmill_common::dbt_manifest::clear_dbt_script_state(&mut tx, &w_id, path).await?;
     // Pipeline event hygiene: an archived script must not be triggered by
     // anything. Wipe declared `// on ...` edges (asset-event subscribers
     // look these up).
@@ -3807,7 +3813,7 @@ async fn archive_script_by_hash(
     clear_static_asset_usage_by_script_hash(&mut *tx, &w_id, hash).await?;
     // The version's graph stays: its finished runs still render from it, and
     // the live-version CTE already skips archived rows. Deletion clears it.
-    windmill_common::dbt_manifest::clear_dbt_run_state_if_path_retired(
+    windmill_common::dbt_manifest::clear_dbt_script_state_if_path_retired(
         &mut tx,
         &w_id,
         &script.path,
@@ -3870,7 +3876,12 @@ async fn delete_script_by_hash(
     )
     .bind(&hash.0)
     .bind(&w_id)
-    .fetch_one(&db)
+    // In the SAME transaction as the cleanup below, as `archive_script_by_hash`
+    // already does. Committed on its own, it opens a window where the path has
+    // no live version and a concurrent deploy can take it — and the retirement
+    // guard below then finds that new script live, keeps the old project's dbt
+    // state, and leaves the replacement able to defer through its manifest.
+    .fetch_one(&mut *tx)
     .await
     .map_err(|e| Error::internal_err(format!("deleting script by hash {w_id}: {e:#}")))?;
 
@@ -3883,7 +3894,7 @@ async fn delete_script_by_hash(
     windmill_common::dbt_manifest::clear_dbt_manifest_version(&mut tx, &w_id, &script.path, hash.0)
         .await?;
     clear_static_asset_usage_by_script_hash(&mut *tx, &w_id, hash).await?;
-    windmill_common::dbt_manifest::clear_dbt_run_state_if_path_retired(
+    windmill_common::dbt_manifest::clear_dbt_script_state_if_path_retired(
         &mut tx,
         &w_id,
         &script.path,
@@ -3984,11 +3995,11 @@ async fn delete_script_by_path(
 
     // After the DELETE, never before: every dbt writer locks the `script` row
     // first, so taking a sidecar ahead of it deadlocks one of the pair. The
-    // VERSIONED graph needs no clear at all, cascading off `script`; the retry
-    // state does, being keyed by path alone and so inherited by whatever is
-    // created here next, and so do the editor's own graphs, whose NULL
+    // VERSIONED graph needs no clear at all, cascading off `script`; the saved
+    // run and environment state do, being keyed by path alone and so inherited
+    // by whatever is created here next, and so do the editor's own graphs, whose NULL
     // `script_hash` satisfies that foreign key without riding its cascade.
-    windmill_common::dbt_manifest::clear_dbt_run_state(&mut tx, &w_id, path).await?;
+    windmill_common::dbt_manifest::clear_dbt_script_state(&mut tx, &w_id, path).await?;
     windmill_common::dbt_manifest::clear_dbt_editor_graphs(&mut tx, &w_id, path).await?;
 
     if !trash_scripts.is_empty() {
@@ -4157,7 +4168,7 @@ async fn delete_scripts_bulk(
     // Same reason as the single-path delete, over every requested path rather
     // than the deleted ones: a path that had no script left can still hold state.
     for p in &request.paths {
-        windmill_common::dbt_manifest::clear_dbt_run_state(&mut tx, &w_id, p).await?;
+        windmill_common::dbt_manifest::clear_dbt_script_state(&mut tx, &w_id, p).await?;
         windmill_common::dbt_manifest::clear_dbt_editor_graphs(&mut tx, &w_id, p).await?;
     }
 

backend/windmill-common/src/dbt_manifest.rs

@@ -35,8 +35,8 @@
 //! Every `pub` mutator in this module — the manifest ones
 //! (`replace_dbt_manifest`, `clear_dbt_manifest_version`,
 //! `clear_dbt_editor_graphs`),
-//! the snapshot sweep, and the retry-state ones (`move_dbt_run_state`,
-//! `clear_dbt_run_state`, `clear_dbt_run_state_if_path_retired`) — takes the
+//! the snapshot sweep, and the script-state ones (`move_dbt_script_state`,
+//! `clear_dbt_script_state`, `clear_dbt_script_state_if_path_retired`) — takes the
 //! workspace and the script to act on as plain arguments and enforces nothing:
 //! **the caller must already have verified write access to that script**,
 //! exactly like the sibling `assets::replace_static_asset_usage` each is called
@@ -1120,22 +1120,27 @@ pub async fn clear_dbt_editor_graphs(
     Ok(())
 }
 
-/// Move a dbt script's saved retry state to its new path.
+/// Move a dbt script's saved state to its new path: the run `dbt retry` resumes,
+/// and the state each environment's deferrals resolve through.
 ///
-/// Keyed by path like the sidecar, but unlike the sidecar it is not
-/// regenerated by anything: the deploy re-ingests a manifest, while these are
-/// the results of a run that already happened. Clearing on rename would throw
-/// away a resumable failure for a cosmetic change, so it travels instead.
+/// Keyed by path like the sidecar, but unlike the sidecar neither is regenerated
+/// by anything: the deploy re-ingests a manifest, while these are the results of
+/// runs that already happened. Clearing on rename would throw away a resumable
+/// failure, and every deferral until the next full run, for a cosmetic change —
+/// so they travel instead. An artifact too large for its row is unaffected: its
+/// key is that publication's own, and the moved row is what names it.
 ///
 /// See the mutator contract above: this authorizes nothing.
-pub async fn move_dbt_run_state(
+pub async fn move_dbt_script_state(
     tx: &mut Transaction<'_, Postgres>,
     workspace_id: &str,
     old_path: &str,
     new_path: &str,
 ) -> Result<()> {
     // The destination may already hold state from a script that lived there
-    // before; the incoming row is the newer truth for this project.
+    // before; the incoming row is the newer truth for this project. What the
+    // displaced row named in object storage is left there, as a cleared one's is
+    // — see `clear_dbt_script_state`.
     sqlx::query!(
         "DELETE FROM dbt_run_state WHERE workspace_id = $1 AND script_path = $2",
         workspace_id,
@@ -1151,22 +1156,38 @@ pub async fn move_dbt_run_state(
     )
     .execute(&mut **tx)
     .await?;
+    sqlx::query!(
+        "DELETE FROM dbt_environment_state WHERE workspace_id = $1 AND script_path = $2",
+        workspace_id,
+        new_path
+    )
+    .execute(&mut **tx)
+    .await?;
+    sqlx::query!(
+        "UPDATE dbt_environment_state SET script_path = $3
+          WHERE workspace_id = $1 AND script_path = $2",
+        workspace_id,
+        old_path,
+        new_path
+    )
+    .execute(&mut **tx)
+    .await?;
     Ok(())
 }
 
-/// Drop the saved retry state, but only once NO live version of the path is
-/// left.
+/// Drop the saved state, but only once NO live version of the path is left.
 ///
-/// `dbt_run_state`'s key is the path and the principal — one saved run per script
-/// per identity it executes as, not
-/// one per version — so archiving or deleting a single version must not take it
-/// with them: the live version's `dbt retry` would be refused and the
-/// partial-failure resume lost. It does not need to be version-scoped either,
-/// because `identity` already refuses a resume whose project, warehouse or
-/// engine moved.
+/// Neither table is keyed by version — `dbt_run_state` by path and principal,
+/// `dbt_environment_state` by path and environment — so archiving or deleting a
+/// single version must not take them with it: the live version's `dbt retry`
+/// would be refused, its partial-failure resume lost, and every deferral would
+/// have to wait for another full run to republish. Neither needs to be
+/// version-scoped either: `identity` already refuses a resume whose project,
+/// warehouse or engine moved, and a deferral resolves relation names, which a
+/// new version of the same project spells the same way.
 ///
 /// See the mutator contract above: this authorizes nothing.
-pub async fn clear_dbt_run_state_if_path_retired(
+pub async fn clear_dbt_script_state_if_path_retired(
     tx: &mut Transaction<'_, Postgres>,
     workspace_id: &str,
     script_path: &str,
@@ -1181,17 +1202,34 @@ pub async fn clear_dbt_run_state_if_path_retired(
     )
     .execute(&mut **tx)
     .await?;
+    sqlx::query!(
+        "DELETE FROM dbt_environment_state WHERE workspace_id = $1 AND script_path = $2
+           AND NOT EXISTS (SELECT 1 FROM script
+                            WHERE workspace_id = $1 AND path = $2
+                              AND deleted = false AND archived = false)",
+        workspace_id,
+        script_path
+    )
+    .execute(&mut **tx)
+    .await?;
     Ok(())
 }
 
-/// Drop a dbt script's saved retry state.
+/// Drop a dbt script's saved state, both halves.
 ///
-/// Archive and delete: `run_results` is not small, the invocation arguments it
-/// carries are the user's, and a script later created at the same path would
-/// otherwise inherit a stranger's resumable failure.
+/// Archive and delete: neither is small, the invocation arguments and manifest
+/// they carry are the user's, and a script later created at the same path would
+/// otherwise inherit a stranger's resumable failure and defer to a project it
+/// has nothing to do with.
+///
+/// An artifact too large for its row lives in the instance's object storage, and
+/// this leaves it there — as a deleted script leaves its bundle. Reaching it from
+/// here would mean an object-store client in this crate and a delete that has to
+/// land after the caller's transaction commits, for one object per environment of
+/// a script that is gone.
 ///
 /// See the mutator contract above: this authorizes nothing.
-pub async fn clear_dbt_run_state(
+pub async fn clear_dbt_script_state(
     tx: &mut Transaction<'_, Postgres>,
     workspace_id: &str,
     script_path: &str,
@@ -1203,6 +1241,13 @@ pub async fn clear_dbt_run_state(
     )
     .execute(&mut **tx)
     .await?;
+    sqlx::query!(
+        "DELETE FROM dbt_environment_state WHERE workspace_id = $1 AND script_path = $2",
+        workspace_id,
+        script_path
+    )
+    .execute(&mut **tx)
+    .await?;
     Ok(())
 }
 

backend/windmill-common/tests/dbt_graph_storage.rs

@@ -7,7 +7,8 @@
 
 use sqlx::{Pool, Postgres};
 use windmill_common::dbt_manifest::{
-    clear_dbt_editor_graphs, clear_dbt_manifest_version, prune_dbt_run_graphs,
+    clear_dbt_editor_graphs, clear_dbt_manifest_version, clear_dbt_script_state,
+    clear_dbt_script_state_if_path_retired, move_dbt_script_state, prune_dbt_run_graphs,
     replace_dbt_editor_graph, replace_dbt_manifest, IngestedManifest, IngestedNode,
     DBT_EDITOR_GRAPHS_KEPT, DEPLOYED_GRAPH, DEPLOYED_GRAPH_VERSIONS_KEPT,
 };
@@ -235,7 +236,11 @@ async fn clearing_one_version_leaves_the_others(db: Pool<Postgres>) {
     // this is where two versions coexist: it pins the batched edge insert
     // against a real database as well as the version scoping.
     assert_eq!(edges_for(&db, 1).await, 0, "the cleared version's edges go");
-    assert_eq!(edges_for(&db, 2).await, 1, "the other version keeps its own");
+    assert_eq!(
+        edges_for(&db, 2).await,
+        1,
+        "the other version keeps its own"
+    );
 }
 
 /// The routes that hard-delete a path clear no graph rows: they delete the
@@ -363,7 +368,11 @@ async fn only_the_newest_deploys_keep_their_graph(db: Pool<Postgres>) {
     // The newest is always among them: losing the live version's graph would
     // empty the page of every run of it.
     assert_eq!(nodes_for(&db, over, DEPLOYED_GRAPH).await, 1);
-    assert_eq!(nodes_for(&db, 1, DEPLOYED_GRAPH).await, 0, "the oldest is reclaimed");
+    assert_eq!(
+        nodes_for(&db, 1, DEPLOYED_GRAPH).await,
+        0,
+        "the oldest is reclaimed"
+    );
 }
 
 /// The third provenance: a `parse` of the EDITOR's buffer, which names no
@@ -480,17 +489,27 @@ async fn a_version_clear_spares_editor_graphs_and_a_path_clear_does_not(db: Pool
     replace_dbt_editor_graph(&mut tx, WS, PATH, job, ME, &manifest(&["a"]), "root")
         .await
         .unwrap();
-    clear_dbt_manifest_version(&mut tx, WS, PATH, 1).await.unwrap();
+    clear_dbt_manifest_version(&mut tx, WS, PATH, 1)
+        .await
+        .unwrap();
     tx.commit().await.unwrap();
 
     assert_eq!(nodes_for(&db, 1, DEPLOYED_GRAPH).await, 0);
-    assert_eq!(editor_nodes(&db, job).await, 1, "the buffer's graph survives");
+    assert_eq!(
+        editor_nodes(&db, job).await,
+        1,
+        "the buffer's graph survives"
+    );
 
     let mut tx = db.begin().await.unwrap();
     clear_dbt_editor_graphs(&mut tx, WS, PATH).await.unwrap();
     tx.commit().await.unwrap();
 
-    assert_eq!(editor_nodes(&db, job).await, 0, "retiring the path takes it");
+    assert_eq!(
+        editor_nodes(&db, job).await,
+        0,
+        "retiring the path takes it"
+    );
 }
 
 /// A preview names its own PATH and needs only `jobs:run`, so a bound over the
@@ -552,3 +571,157 @@ async fn editor_markers(db: &Pool<Postgres>) -> i64 {
     .unwrap()
     .unwrap_or(0)
 }
+
+/// A deferral resolves a `ref()` through the manifest of the last successful run
+/// at this path, so that state has to follow the script the way the retry state
+/// does: a rename must not strand it, and a path no live dbt version occupies
+/// must not hand its manifest to whatever is created there next.
+#[sqlx::test(migrations = "../migrations", fixtures("base"))]
+async fn environment_state_follows_the_script(db: Pool<Postgres>) {
+    const MOVED: &str = "f/test/renamed";
+    deploy_script(&db, 1).await;
+    publish_environment_state(&db, PATH).await;
+
+    let mut tx = db.begin().await.unwrap();
+    move_dbt_script_state(&mut tx, WS, PATH, MOVED)
+        .await
+        .unwrap();
+    tx.commit().await.unwrap();
+    assert_eq!(environment_states(&db, PATH).await, 0);
+    assert_eq!(environment_states(&db, MOVED).await, 1);
+
+    let mut tx = db.begin().await.unwrap();
+    clear_dbt_script_state(&mut tx, WS, MOVED).await.unwrap();
+    tx.commit().await.unwrap();
+    assert_eq!(environment_states(&db, MOVED).await, 0);
+}
+
+/// Archiving or deleting ONE version must not take the path's state with it —
+/// the live version's next deferral still needs it — while the last one leaving
+/// must, or a script later created at that path inherits the previous project's
+/// manifest. The condition is a `NOT EXISTS` in raw SQL, so both directions are
+/// pinned against a real database.
+#[sqlx::test(migrations = "../migrations", fixtures("base"))]
+async fn state_goes_only_once_no_live_version_is_left(db: Pool<Postgres>) {
+    deploy_script(&db, 1).await;
+    deploy_script(&db, 2).await;
+    publish_environment_state(&db, PATH).await;
+
+    retire(&db, 1).await;
+    let mut tx = db.begin().await.unwrap();
+    clear_dbt_script_state_if_path_retired(&mut tx, WS, PATH)
+        .await
+        .unwrap();
+    tx.commit().await.unwrap();
+    assert_eq!(
+        environment_states(&db, PATH).await,
+        1,
+        "another version is still live here"
+    );
+
+    retire(&db, 2).await;
+    let mut tx = db.begin().await.unwrap();
+    clear_dbt_script_state_if_path_retired(&mut tx, WS, PATH)
+        .await
+        .unwrap();
+    tx.commit().await.unwrap();
+    assert_eq!(
+        environment_states(&db, PATH).await,
+        0,
+        "the last one leaving takes it"
+    );
+}
+
+async fn retire(db: &Pool<Postgres>, hash: i64) {
+    sqlx::query!(
+        "UPDATE script SET archived = true WHERE workspace_id = $1 AND hash = $2",
+        WS,
+        hash
+    )
+    .execute(db)
+    .await
+    .unwrap();
+}
+
+/// The worker publishes under a guard naming the version that ran, and the whole
+/// point of it is a job that finishes late: its script can be renamed away and an
+/// unrelated one created at the same path while it runs, and that project must
+/// not inherit this one's manifest as its deferral state. Enforced in raw SQL,
+/// where a refactor can drop a predicate with no type error, so it is pinned
+/// against a real database — the same shape `dbt_state::publish` issues.
+#[sqlx::test(migrations = "../migrations", fixtures("base"))]
+async fn a_late_job_cannot_publish_for_a_path_it_no_longer_owns(db: Pool<Postgres>) {
+    deploy_script(&db, 1).await;
+    assert_eq!(guarded_publish(&db, PATH, 1).await, 1, "its own version");
+    assert_eq!(
+        guarded_publish(&db, PATH, 2).await,
+        0,
+        "a version that never lived here"
+    );
+
+    // The script is gone from this path and another one takes it.
+    sqlx::query!(
+        "DELETE FROM script WHERE workspace_id = $1 AND hash = 1",
+        WS
+    )
+    .execute(&db)
+    .await
+    .unwrap();
+    deploy_script(&db, 3).await;
+    assert_eq!(
+        guarded_publish(&db, PATH, 1).await,
+        0,
+        "the late job's version does not own this path any more"
+    );
+}
+
+/// The predicate `dbt_state::publish` locks the script row on, reduced to what it
+/// decides. Keep the two in step — this file cannot call `publish` itself, which
+/// is `pub(crate)` in `windmill-worker`.
+async fn guarded_publish(db: &Pool<Postgres>, path: &str, ran: i64) -> u64 {
+    sqlx::query!(
+        "INSERT INTO dbt_environment_state (workspace_id, script_path, environment, job_id,
+                                            manifest)
+         SELECT $1::varchar, $2::varchar, 'main||analytics|wh'::text, $3::uuid, '{}'::text
+          WHERE EXISTS (SELECT 1 FROM script
+                         WHERE workspace_id = $1 AND path = $2
+                           AND deleted = false AND archived = false
+                           AND language = 'dbt'
+                           AND (hash = $4 OR $4 = ANY(parent_hashes)))
+         ON CONFLICT (workspace_id, script_path, environment) DO UPDATE SET job_id = EXCLUDED.job_id",
+        WS,
+        path,
+        uuid::Uuid::from_u128(9),
+        ran,
+    )
+    .execute(db)
+    .await
+    .unwrap()
+    .rows_affected()
+}
+
+async fn publish_environment_state(db: &Pool<Postgres>, path: &str) {
+    sqlx::query!(
+        "INSERT INTO dbt_environment_state (workspace_id, script_path, environment, job_id,
+                                            manifest)
+         VALUES ($1, $2, 'main||analytics|wh', $3, '{}')",
+        WS,
+        path,
+        uuid::Uuid::from_u128(9),
+    )
+    .execute(db)
+    .await
+    .unwrap();
+}
+
+async fn environment_states(db: &Pool<Postgres>, path: &str) -> i64 {
+    sqlx::query_scalar!(
+        "SELECT count(*) FROM dbt_environment_state WHERE workspace_id = $1 AND script_path = $2",
+        WS,
+        path
+    )
+    .fetch_one(db)
+    .await
+    .unwrap()
+    .unwrap_or(0)
+}

backend/windmill-worker/src/dbt_executor.rs

@@ -19,12 +19,13 @@ use tokio::process::Command;
 use uuid::Uuid;
 use windmill_common::client::AuthedClient;
 use windmill_common::error::{self, Error};
+use windmill_common::jobs::JobKind;
 use windmill_common::materialization::{
     record_materialization, MaterializationStatus, RecordMaterializationRequest,
 };
 use windmill_common::worker::{to_raw_value, write_file, Connection};
 use windmill_parser_yaml::{
-    parse_dbt_descriptor, DbtDescriptor, DbtTestBehavior, DBT_COMMANDS, DBT_COMMAND_ARG,
+    parse_dbt_descriptor, DbtDescriptor, DbtEngine, DbtTestBehavior, DBT_COMMANDS, DBT_COMMAND_ARG,
     DBT_COMMAND_LABEL, DBT_DEFAULT_WAREHOUSE,
 };
 use windmill_queue::{append_logs, CanceledBy, MiniPulledJob};
@@ -37,6 +38,9 @@ use crate::dbt_engine::{provision_engine, ProvisionedEngine, DBT_CACHE_DIR};
 use crate::dbt_profiles::{
     ensure_adapter_licensed, render_dbt_profile, render_profile, DbtAdapter, KnownAdapter,
 };
+use crate::dbt_state::{
+    environment_label, prepare_deferral, write_state_dir, Deferral, StateManifest, STATE_DIR,
+};
 use crate::handle_child::{
     get_mem_peak, handle_child, run_future_with_polling_update_job_poller, JobCtx, JobDeadline,
 };
@@ -121,6 +125,12 @@ pub struct DbtRunResult {
     /// the same project — cannot get them from the job.
     #[serde(skip_serializing_if = "std::collections::HashMap::is_empty")]
     pub invocation_args: std::collections::HashMap<String, Box<RawValue>>,
+    /// The run whose stored state this one resolved its unbuilt `ref()`s
+    /// through, absent when it deferred to none. What a deferring run built
+    /// against is otherwise unrecoverable: the state is replaced by the next
+    /// successful run of that environment.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub deferred_to: Option<Uuid>,
 }
 
 #[derive(Serialize, Debug, Default)]
@@ -189,7 +199,13 @@ pub(crate) async fn handle_dbt_job(
     // result publishes, and both describe an invocation of this script, not one
     // executor's view of it.
     let raw_args = job.args.as_ref().map(|a| a.0.clone()).unwrap_or_default();
-    let inv = Invocation { args: args.clone(), raw_args, envs: envs.clone(), strict: true };
+    let inv = Invocation {
+        args: args.clone(),
+        raw_args,
+        envs: envs.clone(),
+        deferral: None,
+        strict: true,
+    };
     // One wall clock for the whole job. A dbt job is a sequence of
     // subprocesses — provision, deps, parse, ls, build, then the
     // `after_all` tests — and each would otherwise resolve the job's full
@@ -262,6 +278,16 @@ pub(crate) async fn handle_dbt_job(
     // applies — nothing is built, so there is no test phase, no materialization,
     // no retry state and no ownership to publish.
     if command == "parse" {
+        // Checked here rather than at the seam below, which cannot tell a parse
+        // from a run that simply left `defer` off: a parse never reaches the
+        // deferral at all, so it is the one caller for which "turn `defer` on"
+        // would be advice that leads nowhere.
+        check_state_selectors(
+            &effective_select(&descriptor, &inv)?,
+            &effective_exclude(&descriptor, &inv)?,
+            StateAccess::Never(&command),
+            !selection_is_overridden(&descriptor, &inv.args)?,
+        )?;
         return run_parse_only(
             &prepared,
             &descriptor,
@@ -364,6 +390,84 @@ pub(crate) async fn handle_dbt_job(
         inv
     };
 
+    // Read AFTER the retry restore, so a retry defers exactly as the run it
+    // resumes did: a retry's own arguments are the command block alone, and the
+    // relations its unbuilt `ref()`s resolve to must not depend on that.
+    let defer = arg_bool(&inv.args, "defer")?.unwrap_or(descriptor.defer);
+    // Before the state is fetched, not only at the seam where the selection
+    // reaches dbt: a selector that cannot work whatever the state says would
+    // otherwise be masked by the "nothing published yet" refusal, which sends the
+    // caller to publish a state that will not help.
+    check_state_selectors(
+        &effective_select(&descriptor, &inv)?,
+        &effective_exclude(&descriptor, &inv)?,
+        if defer {
+            StateAccess::Given
+        } else {
+            StateAccess::OnRequest
+        },
+        !selection_is_overridden(&descriptor, &inv.args)?,
+    )?;
+    // A `show` defers too, and every engine takes the flags on it: it COMPILES
+    // the model it previews, so a model whose upstream this environment built and
+    // this run did not is exactly the case a deferral exists for.
+    let inv = if defer {
+        // Refused before anything runs. `dbt retry` reads the run it resumes
+        // from `--state`, the flag a deferral needs, so an engine without
+        // `--defer-state` can be given one or the other: told to defer, it
+        // resumes the stored state's own (successful) results and rebuilds
+        // nothing, and left alone it rebuilds the failed nodes with every
+        // `ref()` resolving into the schema THIS run writes — which for the
+        // narrowed run a deferral exists to serve is not where those models go.
+        if command == "retry" && !prepared.engine.engine.has_defer_state_flag() {
+            return Err(Error::BadRequest(format!(
+                "`{}` cannot resume a run that deferred: `dbt retry` takes the run it resumes \
+                 from `--state`, which is also where a deferral reads its manifest, and this \
+                 engine has no `--defer-state` to tell the two apart. Run the script again \
+                 instead of resuming it, or move the project to dbt-core-1x",
+                prepared.engine.engine.as_str()
+            )));
+        }
+        let deferral = prepare_deferral(&prepared, &job.workspace_id, job_dir, conn).await?;
+        // Only answerable once the state is loaded: `defer` is enough for a
+        // `state:` method, which reads the manifest every publication carries,
+        // but a `result:` one reads `run_results.json` — and a build recovered by
+        // node retry publishes without it, since the results it holds describe
+        // only the nodes the retry rebuilt. dbt-core then raises an INTERNAL
+        // error and the Rust engines match nothing and exit 0.
+        if !deferral.has_run_results
+            && selection_names(
+                &effective_select(&descriptor, &inv)?,
+                &effective_exclude(&descriptor, &inv)?,
+                &["result"],
+            )
+        {
+            return Err(Error::BadRequest(format!(
+                "a `result:` selector reads `run_results.json` out of the published state, and \
+                 the state for this environment ({}) carries only the manifest run {} \
+                 published: a build recovered by node retry stores none, its results describing \
+                 the retried nodes rather than the whole build. Run this script once without \
+                 `defer` and without overrides to publish a complete state, or drop the selector",
+                environment_label(&prepared),
+                deferral.published_by
+            )));
+        }
+        append_logs(
+            &job.id,
+            &job.workspace_id,
+            format!(
+                "\nDeferring unbuilt refs to the dbt state published by run {}; this run \
+                 publishes none of its own\n",
+                deferral.published_by
+            ),
+            conn,
+        )
+        .await;
+        Invocation { deferral: Some(deferral), ..inv }
+    } else {
+        inv
+    };
+
     // Ingested BEFORE the build, from a `dbt parse` with this run's vars, so the
     // models shown are the ones about to be built. Rows are keyed by path, version
     // AND job so no two runs collide; the path-keyed `asset` usage belongs to one
@@ -428,9 +532,28 @@ pub(crate) async fn handle_dbt_job(
     // previous attempt's `run_results.json` is still in the job directory. Never on
     // an agent worker, which cannot read `v2_job_queue` — the wait below would be
     // uninterruptible, so a cancelled job would hold its slot and then start dbt.
+    // And never where the engine cannot be told to defer on a `retry`: the
+    // rebuild would resolve this run's unbuilt refs into the schema it writes
+    // into, so the nodes it "recovered" would read from the wrong relations.
+    // Said out loud below rather than silently skipped.
+    let retry_would_lose_the_deferral =
+        inv.deferral.is_some() && !prepared.engine.engine.has_defer_state_flag();
     let node_retry = descriptor
         .retry_failed_nodes
-        .filter(|_| matches!(conn, Connection::Sql(_)));
+        .filter(|_| matches!(conn, Connection::Sql(_)))
+        .filter(|_| !retry_would_lose_the_deferral);
+    if descriptor.retry_failed_nodes.is_some() && retry_would_lose_the_deferral {
+        append_logs(
+            &job.id,
+            &job.workspace_id,
+            format!(
+                "\nSkipping the automatic node retry: `{}` cannot defer on a `dbt retry`\n",
+                prepared.engine.engine.as_str()
+            ),
+            conn,
+        )
+        .await;
+    }
     let mut retries_left = node_retry.map(|p| p.attempts()).unwrap_or(0);
     if let Some(policy) = node_retry.filter(|_| run.is_err()) {
         retry_failed_nodes(
@@ -519,6 +642,56 @@ pub(crate) async fn handle_dbt_job(
     {
         tracing::warn!("dbt: could not save retry state for job {}: {e:#}", job.id);
     }
+    // What a later run defers to, published by the runs whose relations are the
+    // SCRIPT's — the same condition that decides whether a run's graph becomes
+    // what the script owns, and for the same reason: an invocation that scoped
+    // its own models has no standing to say where this project's relations live.
+    // Success is the other half, because a relation a deferral resolves to has
+    // to exist. A `retry` is excluded: its `run_results.json` names only the
+    // nodes it redid, so publishing it would leave the environment claiming a
+    // run of a handful of models.
+    //
+    // And never a run that DEFERRED, whatever narrowed it. A deferring run built
+    // some of the relations its manifest names and resolved the rest out of the
+    // state it read, so publishing that manifest would record relations nothing
+    // built — and a model renamed since would be recorded under a name only a
+    // full build creates, breaking every later deferral until one repairs it.
+    // `publishes_ownership` cannot see this on its own: it reads the caller's
+    // overrides, and a descriptor that already narrows `select` needs none.
+    if run.is_ok()
+        && command == "build"
+        && inv.deferral.is_none()
+        // A run of the DEPLOYED version, by kind. A preview carries a
+        // caller-supplied `script_hash` into `runnable_id`
+        // (`run_preview_script`), so the version guard alone would let anyone who
+        // may run a job publish arbitrary content as a deployed script's state.
+        && job.kind == JobKind::Script
+        && prepared.graph_refresh.publishes_ownership()
+    {
+        // Losing it costs the next deferral, not the run that just finished —
+        // but silently, so the one actionable case (an artifact too large for
+        // the database on an instance with no object storage) says so.
+        if let Err(e) = crate::dbt_state::publish(
+            &prepared,
+            &job.workspace_id,
+            &job.id,
+            job.runnable_id.map(|h| h.0),
+            // An attempt was spent, so `run_results.json` on disk is the one
+            // `dbt retry` left: the nodes it redid, not the build.
+            node_retry.is_some_and(|p| retries_left < p.attempts()),
+            conn,
+        )
+        .await
+        {
+            append_logs(
+                &job.id,
+                &job.workspace_id,
+                format!("\nCould not publish this run as the environment's dbt state: {e}\n"),
+                conn,
+            )
+            .await;
+        }
+    }
     let reconciled = reconcile_materializations(&prepared, &results, job, conn, client).await;
     terminalize_running_relations(job, &reconciled, conn).await;
 
@@ -895,6 +1068,17 @@ pub struct PreparedProject {
     /// The descriptor's `profile.target`, passed as `--target` so it applies to
     /// a project-owned `profiles.yml` as well as a rendered one.
     pub target: Option<String>,
+    /// The target dbt actually runs, which is the above only when the descriptor
+    /// names one: otherwise it is the workspace warehouse's, or the project's own
+    /// `profiles.yml` default. Half of an environment's identity, since a
+    /// `target.name` macro decides where a model is built.
+    pub effective_target: Option<String>,
+    /// Whether the profile templates where its relations go — a project-owned
+    /// `profiles.yml`, a `dbt_profile` resource's block, or `profile.schema`,
+    /// all of which reach dbt as written. Two renderings then share one
+    /// `relation_root` and an environment cannot be told apart, so such a
+    /// project neither publishes state nor defers to any.
+    pub templated_location: bool,
     /// The profile target's database. Nodes that override it qualify their
     /// `dbt://` schema segment so two databases cannot collapse onto one node.
     pub default_database: Option<String>,
@@ -932,7 +1116,7 @@ impl PreparedProject {
     /// Where this run's relations live: the resolved schema and database. Drift
     /// here since the deploy means the stored graph names relations that no
     /// longer exist.
-    fn relation_root(&self) -> String {
+    pub(crate) fn relation_root(&self) -> String {
         format!(
             "{}|{}",
             self.default_schema.as_deref().unwrap_or(""),
@@ -1064,8 +1248,8 @@ pub(crate) async fn prepare_project(
         .chain(invocation_env.iter().map(|(k, v)| (k.clone(), v.clone())))
         .collect();
 
-    let (profiles_dir, warehouse, adapter, default_database, default_schema, profile_digest) =
-        write_profiles(descriptor, &project_dir, job_dir, client, &template_env).await?;
+    let profile = write_profiles(descriptor, &project_dir, job_dir, client, &template_env).await?;
+    let adapter = profile.adapter.clone();
     // The lockfile's version, when it pinned one for this same engine — a
     // descriptor edited to another engine invalidates the pin.
     let pinned_version = locks
@@ -1186,18 +1370,20 @@ pub(crate) async fn prepare_project(
             h.finish()
         },
         sandbox_config,
-        profile_digest,
+        profile_digest: profile.digest,
         project_dir,
-        profiles_dir,
+        profiles_dir: profile.dir,
         engine,
         graph_refresh,
-        warehouse,
+        warehouse: profile.warehouse,
         target: descriptor.profile.target.clone(),
+        effective_target: profile.target,
+        templated_location: profile.templated_location,
         descriptor_content: descriptor_content.to_string(),
         descriptor_env,
 
-        default_database,
-        default_schema,
+        default_database: profile.database,
+        default_schema: profile.schema,
         script_path: script_path.to_string(),
         env,
     };
@@ -1512,6 +1698,24 @@ async fn strip_git_remote(dir: &Path) -> std::io::Result<()> {
     tokio::fs::write(&config, out).await
 }
 
+/// What resolving the run's connection settled, beyond the file itself.
+struct ResolvedProfile {
+    dir: PathBuf,
+    /// The workspace warehouse's NAME, when this project belongs to one.
+    warehouse: Option<String>,
+    adapter: DbtAdapter,
+    database: Option<String>,
+    schema: Option<String>,
+    /// The target dbt actually runs, which is not always the descriptor's: it
+    /// falls back to the workspace warehouse's, and to the project's own
+    /// `profiles.yml` default. Resolved because it is half of an environment's
+    /// identity and a `target.name` macro can move every relation.
+    target: Option<String>,
+    /// Whether a project-owned `profiles.yml` templates where its relations go.
+    templated_location: bool,
+    digest: String,
+}
+
 /// Write `profiles.yml`, either rendered from a Windmill resource or taken from
 /// the project itself. Both paths are supported (decision 8): the workspace
 /// warehouse is the ergonomic one, the project's own file is what makes an
@@ -1522,14 +1726,7 @@ async fn write_profiles(
     job_dir: &str,
     client: &AuthedClient,
     template_env: &HashMap<String, String>,
-) -> error::Result<(
-    PathBuf,
-    Option<String>,
-    DbtAdapter,
-    Option<String>,
-    Option<String>,
-    String,
-)> {
+) -> error::Result<ResolvedProfile> {
     // The workspace's warehouse, always: a descriptor names one by NAME or takes
     // `main`, and cannot name a resource at all. The NAME is what asset identity
     // keys on, so every project on one warehouse shares its nodes while the
@@ -1610,14 +1807,16 @@ async fn write_profiles(
             }
             None => None,
         };
-        return Ok((
+        return Ok(ResolvedProfile {
             dir,
-            identity,
+            warehouse: identity,
             adapter,
-            target.database,
-            target.schema,
-            profile_digest,
-        ));
+            database: target.database,
+            schema: target.schema,
+            target: Some(target.name),
+            templated_location: target.templated_location,
+            digest: profile_digest,
+        });
     }
 
     use windmill_common::workspaces::DBT_PROFILE_RESOURCE_TYPE;
@@ -1712,14 +1911,22 @@ async fn write_profiles(
         rendered.root_certificate_pem.as_deref(),
         &client.token,
     );
-    Ok((
+    Ok(ResolvedProfile {
         dir,
-        Some(warehouse.to_string()),
+        warehouse: Some(warehouse.to_string()),
         adapter,
-        rendered.database,
-        rendered.schema,
-        profile_digest,
-    ))
+        // A `dbt_profile` resource is one block of the user's own
+        // `profiles.yml`, copied through unchanged, and `profile.schema` is
+        // written as given — so either can be a template dbt renders and this
+        // runtime does not, exactly as a project-owned file can.
+        templated_location: [rendered.database.as_deref(), rendered.schema.as_deref()]
+            .iter()
+            .any(|v| v.is_some_and(is_jinja)),
+        database: rendered.database,
+        schema: rendered.schema,
+        target: Some(target.to_string()),
+        digest: profile_digest,
+    })
 }
 
 /// Where a workspace warehouse name points: its resource path and, if the
@@ -1860,13 +2067,45 @@ async fn adapter_from_profiles_yml(
     // identically to one on a workspace warehouse, which is what lets the two
     // meet on the same node when they are on the same relation.
     let (database_key, schema_key) = adapter.target_identity_keys();
-    let read = |k: &str| {
+    let raw = |k: &str| {
         out.get(k)
             .and_then(|v| v.as_str())
-            .map(|v| v.to_string())
-            .filter(|v| !v.is_empty() && !v.contains("{{"))
+            .filter(|v| !v.is_empty())
     };
-    Ok(ProfileTarget { adapter, database: read(database_key), schema: read(schema_key) })
+    let read = |k: &str| raw(k).filter(|v| !v.contains("{{")).map(|v| v.to_string());
+    Ok(ProfileTarget {
+        adapter,
+        database: read(database_key),
+        schema: read(schema_key),
+        // A TEMPLATED location is one dbt renders and this runtime does not, so
+        // two renderings of this file resolve to one `relation_root` and would
+        // share one environment — `{{ }}` because `read` drops it and it reads
+        // as absent, `{% %}` because the raw block is kept and reads the same
+        // for every rendering. Distinguished from plainly absent, which is the
+        // adapter's default and does not move.
+        templated_location: [database_key, schema_key]
+            .iter()
+            .any(|k| raw(k).is_some_and(is_jinja)),
+        // The output actually chosen, which for a templated `target:` is the sole
+        // one rather than the template text no output answers to.
+        name: match (
+            templated_target,
+            outputs.as_mapping().and_then(|m| m.keys().next()),
+        ) {
+            (true, Some(only)) => only.as_str().unwrap_or(target).to_string(),
+            _ => target.to_string(),
+        },
+    })
+}
+
+/// Whether dbt would RENDER this value rather than take it literally.
+///
+/// Both delimiters, because dbt renders a profile through Jinja: `{{ … }}`
+/// substitutes and `{% … %}` branches, and a schema spelled
+/// `{% if env_var('ENV') == 'prod' %}analytics{% else %}dev{% endif %}` moves
+/// every relation exactly as an `env_var()` does.
+fn is_jinja(v: &str) -> bool {
+    v.contains("{{") || v.contains("{%")
 }
 
 /// What a project-owned `profiles.yml` target says, for the two things Windmill
@@ -1877,6 +2116,14 @@ struct ProfileTarget {
     adapter: DbtAdapter,
     database: Option<String>,
     schema: Option<String>,
+    /// The output this resolved to, by name.
+    name: String,
+    /// Whether its database or schema is a template rather than a literal. The
+    /// fields above cannot say: a `{{ }}` value is dropped and reads as absent,
+    /// a `{% %}` block is kept and reads the same for every rendering. So this
+    /// is what separates "the adapter's default, which does not move" from
+    /// "wherever this run's environment renders it to".
+    templated_location: bool,
 }
 
 lazy_static::lazy_static! {
@@ -2192,6 +2439,29 @@ async fn retry_failed_nodes(
     }
 }
 
+/// The flags that point a deferring invocation at its state directory.
+///
+/// `--state` is where a deferred `ref()` resolves through — except on a `retry`,
+/// which reads the run it RESUMES from that same flag: handed the deferral's
+/// directory, dbt resumes the successful run stored there and rebuilds nothing.
+/// dbt-core 1.x has `--defer-state` for exactly this split; the Rust engines do
+/// not, and a run that defers is refused a retry there rather than rebuilt with
+/// its refs resolving into the schema it writes into (`handle_dbt_job`), which
+/// is why the last arm never fires in practice.
+///
+/// The directory is relative because dbt records the invocation's flags into
+/// `run_results.json`: an absolute path would name the job directory of the run
+/// being resumed, gone by the time anything reads it back.
+fn defer_flags(command: &str, engine: DbtEngine) -> &'static [&'static str] {
+    match command {
+        // `--defer` itself is restored with the rest of the resumed
+        // invocation's arguments and cannot be set from here.
+        "retry" if engine.has_defer_state_flag() => &["--defer-state", STATE_DIR],
+        "retry" => &[],
+        _ => &["--defer", "--state", STATE_DIR],
+    }
+}
+
 #[allow(clippy::too_many_arguments)]
 async fn run_dbt(
     p: &PreparedProject,
@@ -2213,6 +2483,10 @@ async fn run_dbt(
         .args(["--log-format-file", "json"])
         .args(["--log-level-file", p.engine.engine.progress_log_level()]);
 
+    if inv.deferral.is_some() {
+        cmd.args(defer_flags(command, p.engine.engine));
+    }
+
     if with_selection && command != "retry" {
         add_selection(&mut cmd, descriptor, inv)?;
     }
@@ -2864,6 +3138,9 @@ async fn run_show(
         )));
     }
     let mut cmd = dbt_command(p, &["show"]);
+    if inv.deferral.is_some() {
+        cmd.args(defer_flags("show", p.engine.engine));
+    }
     add_vars(&mut cmd, descriptor, inv)?;
     // Intersected with `resource_type:model`, because `show` is only read-only
     // for models: dbt dispatches a selected SEED through its seed runner and
@@ -2957,6 +3234,7 @@ fn build_result(
         totals,
         nodes,
         invocation_args: inv.raw_args.clone(),
+        deferred_to: inv.deferral.as_ref().map(|d| d.published_by),
     }
 }
 
@@ -3470,6 +3748,12 @@ async fn resolve_selection(
         return Ok(None);
     }
     let mut cmd = dbt_command(p, &["ls"]);
+    // The same state the build resolves through, or a `result:` selector — which
+    // reads `run_results.json` out of it, and which `select` passes to dbt
+    // verbatim — fails here, before the build that would have honoured it.
+    if inv.deferral.is_some() {
+        cmd.args(defer_flags("ls", p.engine.engine));
+    }
     // A project whose models call `var()` without a default fails to parse
     // without these, so the selection resolver needs them exactly as the run
     // does. Placeholders that only a run can fill are dropped rather than
@@ -3483,6 +3767,8 @@ async fn resolve_selection(
     }
     cmd.args(["--output", "json", "--quiet"]);
     add_selection(&mut cmd, descriptor, inv)?;
+    let select = effective_select(descriptor, inv)?;
+    let exclude = effective_exclude(descriptor, inv)?;
     // Captured directly, not through `handle_child`: its `pipe_stdout` path goes
     // through the job-log writer, which `NO_LOGS_AT_ALL` discards — the selection
     // would resolve to the empty set and the ingest would wipe the script's assets
@@ -3500,15 +3786,37 @@ async fn resolve_selection(
             }
         }
     }
-    if set.is_empty() {
-        // A selection that matches nothing would be ingested as "this script
-        // owns no relations", wiping its graph and cascade edges — the same
-        // outcome a failed capture produces, and indistinguishable from it.
-        // Refuse rather than silently un-wire the script.
+    // Empty is a real answer from a `state:` or `result:` method and from nothing
+    // else: `state:modified+` matches nothing exactly when nothing changed since
+    // the published state, and a run with no work to do is a successful one. Any
+    // other selection matching nothing is a selector that names nothing — a
+    // misspelled model, say — which must not pass as a build that did its job.
+    // Exempting by ORIGIN rather than by method would let every such typo through.
+    //
+    // What makes the exemption safe is that the empty set is never ingested as
+    // ownership, and that now holds through `check_state_selectors`: a `state:`
+    // or `result:` method survives it only from a run's OWN selection, which
+    // makes `add_caller_args` set `per_run_models`, which makes
+    // `publishes_ownership()` false, so the run stores a snapshot of its own.
+    // Relax the descriptor arm there and a descriptor-narrowed `state:modified+`
+    // reaches here on an unchanged project and wipes the graph the `else` below
+    // guards, with nothing failing.
+    if set.is_empty() && !selection_names(&select, &exclude, &["state", "result"]) {
         return Err(Error::ExecutionErr(
-            "the descriptor's `select`/`exclude` matched no dbt nodes; fix the selection rather \
-             than deploying a script that owns nothing"
-                .to_string(),
+            if selection_is_overridden(descriptor, &inv.args)? {
+                "this run's `select`/`exclude` matched no dbt nodes, so it would build nothing; \
+                 check the selector. Only a `state:` or `result:` selector may match nothing, \
+                 its empty answer being a real one"
+                    .to_string()
+            } else {
+                // The descriptor's is also ingested as "this script owns no
+                // relations", wiping its graph and cascade edges — the same
+                // outcome a failed capture produces, and indistinguishable from
+                // it. Refuse rather than silently un-wire the script.
+                "the descriptor's `select`/`exclude` matched no dbt nodes; fix the selection \
+                 rather than deploying a script that owns nothing"
+                    .to_string()
+            },
         ));
     }
     Ok(Some(set))
@@ -3820,7 +4128,7 @@ async fn save_run_state(
     if let Connection::Sql(db) = conn {
         {
             // Only while a live dbt version stays at this path — the test
-            // `clear_dbt_run_state_if_path_retired` retires state by, plus the
+            // `clear_dbt_script_state_if_path_retired` retires state by, plus the
             // language, since a rename leaves the old path archived rather than
             // deleted and a path can come back as another language. A job already
             // running finishes after those move or clear the row: writing then
@@ -3982,6 +4290,12 @@ pub struct Invocation {
     /// what it pointed at must not.
     pub raw_args: HashMap<String, Box<RawValue>>,
     pub envs: HashMap<String, String>,
+    /// The stored dbt state this invocation resolves an unbuilt `ref()` through,
+    /// materialised into the job directory. Carried here rather than passed to
+    /// each phase: the model phase, the `after_all` tests and every in-job node
+    /// retry must all resolve a `ref()` the same way, or the tests assert against
+    /// relations the models never read.
+    pub deferral: Option<Deferral>,
     /// A run must fail on a `{{ }}` placeholder it cannot fill; a deploy, which
     /// has no arguments at all, tolerates them. Declared rather than inferred
     /// from the argument count: a run submitted with `{}` is still a run, and
@@ -4134,11 +4448,12 @@ async fn restore_from_db(
     if !has_retryable_node(&row.run_results) {
         return Err(nothing_to_retry());
     }
-    let target = p.project_dir.join(ARTIFACTS_DIR);
-    tokio::fs::create_dir_all(&target).await.ok();
-    tokio::fs::write(target.join("run_results.json"), &row.run_results)
-        .await
-        .map_err(|e| Error::internal_err(format!("restoring run_results.json: {e}")))?;
+    write_state_dir(
+        &p.project_dir.join(ARTIFACTS_DIR),
+        Some(&row.run_results),
+        StateManifest::None,
+    )
+    .await?;
     // No manifest came with the row, so one has to be re-derived — but not here:
     // these arguments are as SUBMITTED, and a `$var:` in them shapes the graph
     // only once resolved. The caller resolves, then parses.
@@ -4370,20 +4685,16 @@ async fn restore_run_state(
         return Err(different_project());
     }
     let saved_args_digest = saved_args_digest.map(str::to_string);
-    let target = p.project_dir.join(ARTIFACTS_DIR);
-    tokio::fs::create_dir_all(&target).await.ok();
-    // From the bytes already read, not by copying the file again: a burst of saves
-    // can prune this generation mid-restore, and a `dbt retry` whose
+    // The results go from the bytes already read, not by copying the file again: a
+    // burst of saves can prune this generation mid-restore, and a `dbt retry` whose
     // `run_results.json` went missing rebuilds nothing and reports success. The
     // manifest has no such copy, so a failure there falls back to a `dbt parse`.
-    tokio::fs::write(target.join("run_results.json"), &saved_results)
-        .await
-        .map_err(|e| {
-            Error::internal_err(format!("could not restore the previous run's results: {e}"))
-        })?;
-    let needs_parse = tokio::fs::copy(snapshot.join("manifest.json"), target.join("manifest.json"))
-        .await
-        .is_err();
+    let needs_parse = !write_state_dir(
+        &p.project_dir.join(ARTIFACTS_DIR),
+        Some(&saved_results),
+        StateManifest::CopyOf(snapshot.join("manifest.json")),
+    )
+    .await?;
     // The generation was chosen from a row read before the file work above. A run
     // finishing in that window publishes a newer one, and resuming the superseded
     // generation redoes nodes it has already rebuilt — appending to an incremental
@@ -4772,10 +5083,26 @@ fn add_selection(
     descriptor: &DbtDescriptor,
     inv: &Invocation,
 ) -> error::Result<()> {
-    for s in effective_select(descriptor, inv)? {
+    let select = effective_select(descriptor, inv)?;
+    let exclude = effective_exclude(descriptor, inv)?;
+    // The seam itself, which the DEPLOY reaches without going through a run: it
+    // resolves the descriptor's selection to decide what the script owns, and
+    // never computes a `defer`. A run has been checked earlier, where the message
+    // can still come before the state fetch.
+    check_state_selectors(
+        &select,
+        &exclude,
+        if inv.deferral.is_some() {
+            StateAccess::Given
+        } else {
+            StateAccess::OnRequest
+        },
+        !selection_is_overridden(descriptor, &inv.args)?,
+    )?;
+    for s in select {
         cmd.args(["--select", &s]);
     }
-    for s in effective_exclude(descriptor, inv)? {
+    for s in exclude {
         cmd.args(["--exclude", &s]);
     }
     if let Some(sel) = effective_selector(descriptor, inv)? {
@@ -4784,6 +5111,115 @@ fn add_selection(
     Ok(())
 }
 
+/// The method a selection token names, with the graph operators that can
+/// surround a node stripped (`@model`, `+model`, `2+model`, `model+`).
+fn selector_method(token: &str) -> Option<&str> {
+    token
+        .trim_start_matches('@')
+        .trim_start_matches(|c: char| c.is_ascii_digit())
+        .trim_start_matches('+')
+        .split_once(':')
+        .map(|(method, _)| method)
+}
+
+/// Every method a selection names. Each entry is a union of whitespace-separated
+/// tokens, and each of those an intersection of comma-separated ones.
+fn selection_methods<'a>(entries: &'a [String]) -> impl Iterator<Item = &'a str> {
+    entries
+        .iter()
+        .flat_map(|entry| entry.split([' ', '\t', ',']))
+        .filter_map(selector_method)
+}
+
+/// Whether the run being checked has the state directory a `state:` or `result:`
+/// method reads, or could be given one.
+#[derive(Clone, Copy)]
+enum StateAccess<'a> {
+    /// Deferring, so the directory is there.
+    Given,
+    /// Not deferring, and `defer` is what would hand it one.
+    OnRequest,
+    /// This command resolves a selection without ever deferring, so no setting
+    /// gives it a state and "turn `defer` on" would be advice that leads nowhere.
+    Never(&'a str),
+}
+
+/// Whether a selection names any of these methods.
+fn selection_names(select: &[String], exclude: &[String], methods: &[&str]) -> bool {
+    selection_methods(select)
+        .chain(selection_methods(exclude))
+        .any(|method| methods.contains(&method))
+}
+
+/// Refuse a selection dbt cannot resolve, before it silently resolves to the
+/// wrong thing.
+///
+/// `state:` and `result:` compare against the artifacts in `--state`, which only
+/// a deferring run is given. The engines do not agree on what happens without
+/// one: dbt-core 1.x raises, but dbt-sa-cli and fusion read a missing state as an
+/// EMPTY one and exit 0, so `state:modified` builds nothing and `state:new`
+/// builds the whole project, each as a run that reports success.
+///
+/// From the DESCRIPTOR they are refused whether or not the run defers, because
+/// that selection also decides which nodes the script owns, and "whatever changed
+/// last" is not an ownership answer — the deploy resolves it with no state at all.
+/// They describe one run, so they belong in a run's own `select`.
+///
+/// `source_status:` compares `sources.json`, which `dbt source freshness` writes
+/// and no run publishes here, so it has nothing to compare against under any
+/// setting.
+///
+/// Only what `select` and `exclude` spell directly: a method reached through a
+/// `selectors.yml` definition is named nowhere the worker can read, and dbt's
+/// own behaviour is what stands there.
+fn check_state_selectors(
+    select: &[String],
+    exclude: &[String],
+    access: StateAccess<'_>,
+    from_descriptor: bool,
+) -> error::Result<()> {
+    for method in selection_methods(select).chain(selection_methods(exclude)) {
+        match method {
+            "source_status" => {
+                return Err(Error::BadRequest(
+                    "a `source_status:` selector compares the source freshness recorded in \
+                     `sources.json`, which `dbt source freshness` writes and no run stores \
+                     here, so there is nothing for it to compare against. Drop the selector"
+                        .to_string(),
+                ))
+            }
+            "state" | "result" if from_descriptor => {
+                return Err(Error::BadRequest(format!(
+                    "a `{method}:` selector describes what ONE run builds, but the descriptor's \
+                     selection also decides which nodes this script owns, which a deploy \
+                     resolves with no state to compare against. Move it to the `select` of a \
+                     run with `defer` on"
+                )))
+            }
+            "state" | "result" => match access {
+                StateAccess::Given => {}
+                StateAccess::OnRequest => {
+                    return Err(Error::BadRequest(format!(
+                        "a `{method}:` selector compares against the dbt state a previous run \
+                         of this environment published, and only a run with `defer` on is given \
+                         that state. Turn `defer` on, or drop the selector"
+                    )))
+                }
+                StateAccess::Never(command) => {
+                    return Err(Error::BadRequest(format!(
+                        "a `{method}:` selector compares against the dbt state a previous run \
+                         of this environment published, and `{command}` resolves its selection \
+                         without building and never defers, so no setting hands it that state. \
+                         Drop the selector"
+                    )))
+                }
+            },
+            _ => {}
+        }
+    }
+    Ok(())
+}
+
 /// Whether this invocation chose its own `select`/`exclude`.
 ///
 /// DIFFERENT from the descriptor's, not merely present: `parse_dbt_sig` gives
@@ -5797,6 +6233,27 @@ mod tests {
             .unwrap();
         assert!(untouched.publishes_ownership());
         assert_eq!(untouched.snapshot_job(job), None);
+
+        // `resolve_selection` lets a selection match nothing on exactly this
+        // predicate, because a run that scoped its own selection stores a
+        // snapshot instead of publishing ownership. Should the two ever drift
+        // apart, an empty caller selection would wipe the script's graph and
+        // cascade edges, which is the outcome that guard exists to prevent.
+        // One-directional: a `vars` override also withholds ownership without
+        // touching the selection, which is why this is an implication and not an
+        // equivalence.
+        for args in [
+            arg("select", r#"["state:modified+"]"#),
+            arg("exclude", r#"["tag:nightly"]"#),
+        ] {
+            assert!(selection_is_overridden(&descriptor, &args).unwrap());
+            let mut g = GraphRefresh::default();
+            g.add_caller_args(&descriptor, &args).unwrap();
+            assert!(
+                !g.publishes_ownership(),
+                "an overridden selection must not publish ownership"
+            );
+        }
     }
 
     // `dbt retry` restores the previous run's target/ from this directory, so two
@@ -5825,6 +6282,128 @@ mod tests {
         );
     }
 
+    // A profile whose location dbt renders cannot be told apart from another
+    // rendering of itself, so it neither publishes state nor defers. Both
+    // delimiters count: a conditional block moves a schema exactly as an
+    // `env_var()` substitution does.
+    #[test]
+    fn a_rendered_profile_location_is_recognised_by_either_delimiter() {
+        assert!(is_jinja("{{ env_var('DBT_SCHEMA') }}"));
+        assert!(is_jinja(
+            "{% if env_var('ENV') == 'prod' %}analytics{% else %}dev{% endif %}"
+        ));
+        assert!(!is_jinja("analytics"));
+        assert!(!is_jinja(""));
+    }
+
+    // dbt-sa-cli and fusion exit 0 on a state selector with no state, so nothing
+    // downstream would report this: the graph operators have to be stripped for
+    // the method to be seen at all.
+    #[test]
+    fn a_state_selector_is_found_under_any_graph_operator() {
+        // A run's own selection, which is the only place these belong.
+        let refused = |sel: &str, access: StateAccess<'_>| {
+            check_state_selectors(&[sel.to_string()], &[], access, false).is_err()
+        };
+        for sel in [
+            "state:modified",
+            "state:modified+",
+            "+state:new",
+            "@state:modified",
+            "2+state:modified+3",
+            "tag:nightly,state:modified",
+            "stg_orders+ result:error+",
+        ] {
+            assert!(
+                refused(sel, StateAccess::OnRequest),
+                "{sel} should need `defer`"
+            );
+            assert!(
+                !refused(sel, StateAccess::Given),
+                "{sel} should pass while deferring"
+            );
+            // A parse resolves a selection without ever deferring, so it is
+            // refused where a run would have been told to turn `defer` on.
+            assert!(
+                refused(sel, StateAccess::Never("parse")),
+                "{sel} cannot parse"
+            );
+            // The descriptor's selection also decides what the script owns, and
+            // the deploy resolves it with no state, so deferring cannot save it.
+            assert!(
+                check_state_selectors(&[sel.to_string()], &[], StateAccess::Given, true).is_err(),
+                "{sel} should never be a descriptor selection"
+            );
+        }
+        // A node whose name merely starts with a method's letters is not one.
+        for sel in ["stg_orders+", "tag:nightly", "stateful_model+"] {
+            assert!(
+                !refused(sel, StateAccess::OnRequest),
+                "{sel} is not a state selector"
+            );
+        }
+        // No run publishes `sources.json`, so deferring does not help.
+        for access in [
+            StateAccess::Given,
+            StateAccess::OnRequest,
+            StateAccess::Never("parse"),
+        ] {
+            assert!(refused("source_status:fresher+", access));
+        }
+        // `exclude` reaches dbt the same way `select` does.
+        assert!(check_state_selectors(
+            &[],
+            &["state:modified".to_string()],
+            StateAccess::OnRequest,
+            false
+        )
+        .is_err());
+
+        // The same recognition decides which empty selections `resolve_selection`
+        // lets through. Only these two answer "nothing" meaningfully; a selector
+        // naming nothing must not pass as a build that did its work.
+        const STATE_BACKED: &[&str] = &["state", "result"];
+        for sel in ["state:modified+", "result:error+", "tag:x,state:new"] {
+            assert!(
+                selection_names(&[sel.to_string()], &[], STATE_BACKED),
+                "{sel}"
+            );
+        }
+        for sel in [
+            "mispelled_model",
+            "tag:nightly",
+            "stg_orders+",
+            "source_status:fresher+",
+        ] {
+            assert!(
+                !selection_names(&[sel.to_string()], &[], STATE_BACKED),
+                "{sel}"
+            );
+        }
+    }
+
+    // The one flag choice that is silently wrong rather than loudly wrong: a
+    // `retry` handed `--state` resumes the SUCCESSFUL run stored there and
+    // rebuilds nothing, reporting a green retry of a failed run.
+    #[test]
+    fn a_retry_is_never_handed_the_deferral_as_its_state() {
+        assert_eq!(
+            defer_flags("build", DbtEngine::DbtCore1x),
+            ["--defer", "--state", crate::dbt_state::STATE_DIR]
+        );
+        assert_eq!(
+            defer_flags("test", DbtEngine::Fusion),
+            ["--defer", "--state", crate::dbt_state::STATE_DIR]
+        );
+        assert_eq!(
+            defer_flags("retry", DbtEngine::DbtCore1x),
+            ["--defer-state", crate::dbt_state::STATE_DIR]
+        );
+        for engine in [DbtEngine::DbtCore2x, DbtEngine::Fusion] {
+            assert!(defer_flags("retry", engine).is_empty());
+        }
+    }
+
     #[test]
     fn events_without_a_relation_are_not_materializations() {
         // A test node has no relation of its own.

backend/windmill-worker/src/dbt_state.rs

@@ -0,0 +1,748 @@
+//! The dbt state a project last built into one environment, and the state
+//! directory a run reads it back through.
+//!
+//! `dbt --defer --state <dir>` resolves a `ref()` the run does not build to the
+//! relation the manifest in `<dir>` names, instead of to the schema this run
+//! writes into. That makes the state a durable, per-environment artifact rather
+//! than a cache: the next run of a project usually lands on a worker holding
+//! neither the manifest nor the results, so anything worker-local answers for
+//! one machine's history rather than for the environment.
+//!
+//! Two artifacts live in that directory and both are stored: `manifest.json`,
+//! which is what a deferral resolves through, and `run_results.json`, which
+//! `select`'s `result:` selectors read — and `select` reaches dbt verbatim, so a
+//! state directory missing it fails a selection a user may legitimately write.
+
+use std::path::{Path, PathBuf};
+
+use uuid::Uuid;
+use windmill_common::error::{self, Error};
+use windmill_common::worker::Connection;
+
+use crate::dbt_executor::{digest, PreparedProject, ARTIFACTS_DIR};
+
+lazy_static::lazy_static! {
+    /// Above this, an artifact goes to the instance's object storage instead of
+    /// into the row. A manifest passes a few hundred KB on a handful of models
+    /// and grows with the project, so this ceiling is what decides whether a
+    /// large project needs storage configured at all; a small one stays in the
+    /// database, where it costs no round trip and needs nothing configured.
+    static ref DBT_STATE_INLINE_MAX_BYTES: usize = std::env::var("DBT_STATE_INLINE_MAX_BYTES")
+        .ok()
+        .and_then(|v| v.parse().ok())
+        .unwrap_or(8 * 1024 * 1024);
+}
+
+/// The directory `--state` points at. Inside the job directory, so it sits in
+/// the sandbox's one writable bind and goes away with the job, and prefixed like
+/// the artifacts directory beside it so a project carrying a directory of this
+/// name is not overwritten.
+///
+/// Passed to dbt RELATIVE, and that is load-bearing rather than tidiness. dbt
+/// records the invocation's flags into `run_results.json` and a later
+/// `dbt retry` restores them, so an absolute path would name the job directory
+/// of the run being resumed — gone by then, leaving the retry to resolve a
+/// deferred `ref()` against nothing. Relative, it resolves against the project
+/// root, which is whichever job directory the retry landed in.
+pub(crate) const STATE_DIR: &str = "wm_dbt_state";
+
+/// Where this run's relations live, which is the only thing a deferral is about.
+pub(crate) fn environment(p: &PreparedProject) -> String {
+    environment_key(
+        p.warehouse.as_deref(),
+        // The target dbt RUNS, not the descriptor's: it falls back to the
+        // workspace warehouse's and to the project's own default, so reading the
+        // descriptor's would put two inherited targets under one empty name.
+        p.effective_target.as_deref(),
+        // The pair `relation_root` reports to the graph's drift check, taken
+        // apart so neither can absorb the other's delimiter below.
+        p.default_schema.as_deref(),
+        p.default_database.as_deref(),
+    )
+}
+
+/// The warehouse and the target name the environment; the database and schema
+/// they resolve to are in the key because a repointed warehouse resource or a
+/// moved schema keeps both names while putting the relations somewhere else —
+/// and a manifest is a list of relation names, so a deferral has no other way to
+/// notice. A move therefore reads as an environment nothing has published yet.
+///
+/// Length-prefixed rather than joined on a separator. Every component but the
+/// warehouse is spelled by the user — a dbt target name and a schema are both
+/// arbitrary strings a profile may quote — so `prod|analytics` + `scratch` and
+/// `prod` + `analytics|scratch` would otherwise be one key, and a profile moving
+/// between them would read as the same environment rather than as one nothing
+/// has published. Same reasoning as `stable_digest`, and still legible in a row:
+/// `4:main|4:prod|9:analytics|12:dbt_wh_defer`. What a MESSAGE names is
+/// `environment_label`, since this encoding is for storage.
+fn environment_key(
+    warehouse: Option<&str>,
+    target: Option<&str>,
+    schema: Option<&str>,
+    database: Option<&str>,
+) -> String {
+    [warehouse, target, schema, database]
+        .iter()
+        .map(|v| {
+            let v = v.unwrap_or("");
+            format!("{}:{v}", v.len())
+        })
+        .collect::<Vec<_>>()
+        .join("|")
+}
+
+/// The environment as a message names it: the key above is length-prefixed for
+/// storage, which is not something to put in front of a caller.
+pub(crate) fn environment_label(p: &PreparedProject) -> String {
+    format!(
+        "warehouse `{}`, target `{}`, relations in `{}`",
+        p.warehouse.as_deref().unwrap_or("(none)"),
+        p.effective_target
+            .as_deref()
+            .unwrap_or("(the profile's default)"),
+        match (p.default_database.as_deref(), p.default_schema.as_deref()) {
+            (Some(db), Some(schema)) => format!("{db}.{schema}"),
+            (None, Some(schema)) => schema.to_string(),
+            _ => "(the adapter's default)".to_string(),
+        }
+    )
+}
+
+/// The state one environment last published.
+pub(crate) struct StoredState {
+    pub manifest: String,
+    pub run_results: Option<String>,
+    /// The run that published it, so a deferring run can say what it deferred to.
+    pub job_id: Uuid,
+}
+
+/// Publish this run's artifacts as the environment's state.
+///
+/// Called for a run that BUILT what the script's own descriptor selects and
+/// succeeded (see `handle_dbt_job`). Best-effort in the same sense as the retry
+/// state: losing it costs the next deferral, not the run that just finished.
+///
+/// **What the artifacts may carry follows from that condition.** A publishing run
+/// added nothing of its own — no `select` or `vars` override, and a descriptor
+/// interpolating a `{{ }}` placeholder into `vars` never publishes at all — so
+/// dbt's `run_results.json` records the descriptor's own arguments, which are the
+/// script's content. That is why this is keyed by environment where
+/// `dbt_run_state` is keyed by principal: the retry state holds whatever a caller
+/// submitted, this holds what the script says. Widen the publish condition and
+/// that stops being true.
+pub(crate) async fn publish(
+    p: &PreparedProject,
+    w_id: &str,
+    job_id: &Uuid,
+    // The version this job ran. `None` for a preview, which publishes nothing.
+    script_hash: Option<i64>,
+    // A build recovered by the automatic in-job node retry has a
+    // `run_results.json` naming only the nodes that retry redid. The manifest is
+    // unaffected — it is a function of the project, not of what ran — so the
+    // state is published without results rather than with a set describing some
+    // other slice of the build.
+    results_are_partial: bool,
+    conn: &Connection,
+) -> error::Result<()> {
+    let Connection::Sql(db) = conn else {
+        // An agent worker reaches the database only through the API, which does
+        // not expose this table.
+        return Ok(());
+    };
+    if p.script_path.is_empty() {
+        // A preview has no path to key state on, and an empty one would be
+        // shared by every dbt script in the workspace.
+        return Ok(());
+    }
+    if p.templated_location {
+        // Refused on this side too, not only where a deferral reads. A template
+        // renders to one location per environment while the key sees the
+        // template, so publishing would file this run's manifest under a key a
+        // literal profile shares — and de-templating later would make that stale
+        // manifest readable as the new location's.
+        return Ok(());
+    }
+    let artifacts = p.project_dir.join(ARTIFACTS_DIR);
+    // The manifest is what a deferral resolves through, so there is no state
+    // without one. Every engine writes it beside the results of a build, so this
+    // is the invocation that built nothing rather than a case to report.
+    let Ok(manifest) = tokio::fs::read_to_string(artifacts.join("manifest.json")).await else {
+        return Ok(());
+    };
+    let run_results = match results_are_partial {
+        true => None,
+        false => tokio::fs::read_to_string(artifacts.join("run_results.json"))
+            .await
+            .ok(),
+    };
+    let environment = environment(p);
+    // Uploaded BEFORE the transaction, and to this publication's own keys, so two
+    // publishers cannot collide on them and nothing here can overwrite an
+    // artifact a committed row still names. A failure below has only its own
+    // objects to drop.
+    let nonce = Uuid::new_v4();
+    let (manifest, manifest_key) = store(
+        manifest,
+        "manifest.json",
+        &environment,
+        &p.script_path,
+        w_id,
+        job_id,
+        &nonce,
+    )
+    .await?;
+    let (run_results, run_results_key) = match run_results {
+        Some(r) => match store(
+            r,
+            "run_results.json",
+            &environment,
+            &p.script_path,
+            w_id,
+            job_id,
+            &nonce,
+        )
+        .await
+        {
+            Ok(stored) => stored,
+            Err(e) => {
+                forget_objects(&[manifest_key, None]).await;
+                return Err(e);
+            }
+        },
+        None => (None, None),
+    };
+    let mine = [manifest_key.clone(), run_results_key.clone()];
+    // One publisher per environment at a time, so the row and the objects it
+    // displaces are settled by one of them at a time. An advisory lock rather
+    // than the row's, because the first publish of an environment has no row to
+    // lock and is exactly when two runs of a newly deployed script are most
+    // likely to race.
+    let mut tx = match db.begin().await {
+        Ok(tx) => tx,
+        Err(e) => {
+            forget_objects(&mine).await;
+            return Err(e.into());
+        }
+    };
+    let staged = async {
+        sqlx::query_scalar!(
+            "SELECT pg_advisory_xact_lock($1)",
+            publication_lock(w_id, &p.script_path, &environment)
+        )
+        .execute(&mut *tx)
+        .await?;
+        // The script row FIRST, and held, so a rename, archive or delete of this
+        // path either waits for this publication or is seen by it. Reading it
+        // unlocked leaves a window where lifecycle cleanup finds no row to clear,
+        // finishes, and this transaction then commits state at a path a new
+        // script goes on to occupy. Script row before sidecar is also the order
+        // every other dbt writer takes, which is what keeps the two off a
+        // deadlock.
+        //
+        // The version, not just the path: "some live dbt script is here" is also
+        // satisfied by a script created at a path this one was renamed away from.
+        // A preview names no version, so `script_hash` is NULL and nothing
+        // matches — right for a run of content that was never deployed.
+        let owns_path = sqlx::query_scalar!(
+            "SELECT 1 FROM script
+              WHERE workspace_id = $1 AND path = $2
+                AND deleted = false AND archived = false AND language = 'dbt'
+                AND (hash = $3 OR $3 = ANY(parent_hashes))
+              FOR SHARE",
+            w_id,
+            &p.script_path,
+            script_hash,
+        )
+        .fetch_optional(&mut *tx)
+        .await?
+        .is_some();
+        if !owns_path {
+            return error::Result::Ok(None);
+        }
+        // What the row points at NOW, so those objects can go once this one is
+        // committed in their place — never before, since a reader that has
+        // already read the row is about to fetch them.
+        let displaced = sqlx::query!(
+            "SELECT manifest_key, run_results_key FROM dbt_environment_state
+              WHERE workspace_id = $1 AND script_path = $2 AND environment = $3",
+            w_id,
+            &p.script_path,
+            environment
+        )
+        .fetch_optional(&mut *tx)
+        .await?
+        .map(|r| [r.manifest_key, r.run_results_key])
+        .unwrap_or_default()
+        // Never a key this publication is about to commit. The keys carry a
+        // per-execution nonce so the two cannot coincide, and this is what says
+        // so rather than leaving it to be re-derived.
+        .map(|k| k.filter(|k| !mine.iter().flatten().any(|m| m == k)));
+        sqlx::query!(
+            "INSERT INTO dbt_environment_state (workspace_id, script_path, environment, job_id,
+                                                manifest, manifest_key, run_results,
+                                                run_results_key, updated_at)
+             VALUES ($1, $2, $3, $4, $5, $6, $7, $8, now())
+             ON CONFLICT (workspace_id, script_path, environment) DO UPDATE SET
+               job_id = EXCLUDED.job_id, manifest = EXCLUDED.manifest,
+               manifest_key = EXCLUDED.manifest_key, run_results = EXCLUDED.run_results,
+               run_results_key = EXCLUDED.run_results_key, updated_at = now()",
+            w_id,
+            &p.script_path,
+            environment,
+            job_id,
+            manifest,
+            manifest_key,
+            run_results,
+            run_results_key,
+        )
+        .execute(&mut *tx)
+        .await?;
+        error::Result::Ok(Some(displaced))
+    }
+    .await;
+    let displaced = match staged {
+        // Refused by the guard, or the write failed: nothing is committed and
+        // what was uploaded above has no row naming it.
+        Ok(None) | Err(_) => {
+            forget_objects(&mine).await;
+            return staged.map(|_| ());
+        }
+        Ok(Some(displaced)) => displaced,
+    };
+    // A commit that reports an error may still have committed — what was lost can
+    // be the acknowledgement. Dropping this run's objects would then leave the
+    // committed row naming objects that are gone, and every deferral would fail
+    // until the next publication; an orphan costs storage instead.
+    tx.commit().await?;
+    forget_objects(&displaced).await;
+    Ok(())
+}
+
+/// The environment's state, or `None` where nothing has published one.
+pub(crate) async fn load(
+    p: &PreparedProject,
+    w_id: &str,
+    conn: &Connection,
+) -> error::Result<Option<StoredState>> {
+    let Connection::Sql(db) = conn else {
+        return Err(Error::BadRequest(
+            "`defer` resolves a `ref()` through the dbt state stored for this environment, which \
+             an agent worker cannot read: it reaches the database only through the API. Run this \
+             script on a worker of the main group, or without `defer`"
+                .to_string(),
+        ));
+    };
+    let environment = environment(p);
+    // A publication committing between the row and the objects it named drops
+    // those objects, so a miss is re-read rather than reported as a state that is
+    // not there. Re-read for as long as the row keeps MOVING: a reader takes no
+    // lock, so back-to-back publications can each overtake it, and a fixed one
+    // retry would report the second as missing. An unmoved row is the other
+    // answer — nothing republished, so the object really is gone.
+    let mut tried: Option<(Uuid, Option<String>, Option<String>)> = None;
+    for _ in 0..PUBLICATIONS_OUTRUN {
+        let Some(row) = sqlx::query!(
+            "SELECT job_id, manifest, manifest_key, run_results, run_results_key
+               FROM dbt_environment_state
+              WHERE workspace_id = $1 AND script_path = $2 AND environment = $3",
+            w_id,
+            &p.script_path,
+            environment
+        )
+        .fetch_optional(db)
+        .await?
+        else {
+            return Ok(None);
+        };
+        let seen = (
+            row.job_id,
+            row.manifest_key.clone(),
+            row.run_results_key.clone(),
+        );
+        let fetched = async {
+            let manifest = fetch(row.manifest, row.manifest_key).await?;
+            let run_results = fetch(row.run_results, row.run_results_key).await?;
+            error::Result::Ok((manifest, run_results))
+        }
+        .await;
+        match fetched {
+            Ok((Some(manifest), run_results)) => {
+                return Ok(Some(StoredState { manifest, run_results, job_id: seen.0 }))
+            }
+            Ok((None, _)) => return Ok(None),
+            Err(e) => {
+                if tried.as_ref() == Some(&seen) {
+                    return Err(e);
+                }
+                tried = Some(seen);
+            }
+        }
+    }
+    Err(Error::internal_err(
+        "the dbt state for this environment was replaced faster than it could be read; run this \
+         script again"
+            .to_string(),
+    ))
+}
+
+/// How many publications a read may lose to before it gives up. Each one costs a
+/// re-read, and a project publishing this often while another run defers is
+/// already contending for the same relations.
+const PUBLICATIONS_OUTRUN: usize = 5;
+
+/// A `manifest.json` for a state directory, whichever side it comes from.
+///
+/// One enum because the three restores — a deferral's stored state, a retry's
+/// worker-local generation, a retry's database row — differ only in where the
+/// bytes are, and a second copy of the directory layout is a second chance for
+/// one of them to write a directory dbt reads differently.
+pub(crate) enum StateManifest {
+    Bytes(String),
+    /// A file on this worker, copied rather than read into memory: a manifest
+    /// grows with the project.
+    CopyOf(PathBuf),
+    None,
+}
+
+/// Write the artifacts a dbt state directory holds, creating it if needed.
+///
+/// Returns whether a `manifest.json` ended up there — a worker-local generation
+/// can be pruned out from under a restore, and the caller then owes a
+/// `dbt parse` for one.
+pub(crate) async fn write_state_dir(
+    dir: &Path,
+    run_results: Option<&str>,
+    manifest: StateManifest,
+) -> error::Result<bool> {
+    tokio::fs::create_dir_all(dir)
+        .await
+        .map_err(|e| Error::internal_err(format!("preparing the dbt state directory: {e}")))?;
+    if let Some(run_results) = run_results {
+        tokio::fs::write(dir.join("run_results.json"), run_results)
+            .await
+            .map_err(|e| Error::internal_err(format!("writing run_results.json: {e}")))?;
+    }
+    Ok(match manifest {
+        StateManifest::Bytes(m) => {
+            tokio::fs::write(dir.join("manifest.json"), m)
+                .await
+                .map_err(|e| Error::internal_err(format!("writing manifest.json: {e}")))?;
+            true
+        }
+        StateManifest::CopyOf(from) => tokio::fs::copy(from, dir.join("manifest.json"))
+            .await
+            .is_ok(),
+        StateManifest::None => false,
+    })
+}
+
+/// The advisory lock one environment's publishers take, so only one of them
+/// settles the row and the objects it displaces at a time.
+///
+/// Derived from the same three components as the row's key. Two environments
+/// whose digests collide wait for each other, which costs a moment and nothing
+/// else.
+fn publication_lock(w_id: &str, script_path: &str, environment: &str) -> i64 {
+    let d = digest(&format!("{w_id}|{script_path}|{environment}"));
+    // Parsed unsigned and reinterpreted: half of all digests set the top bit,
+    // and read as `i64` those overflow and would collapse onto one key.
+    u64::from_str_radix(&d[..16], 16).unwrap_or_default() as i64
+}
+
+/// The object-storage key an artifact takes.
+///
+/// One key per PUBLICATION, so an upload never overwrites an artifact the
+/// committed row still names: a run that fails between its two uploads, or
+/// between them and its row, leaves the state pointing at the pair it already
+/// had. The row switches to these in one statement and the objects it displaced
+/// are dropped afterwards. The path and environment are only a prefix — the row
+/// is what says where an artifact is, so state that moves with a renamed script
+/// keeps naming objects under the old one. Digested because a Windmill path and a
+/// schema name may both carry characters an object key gives meaning to.
+///
+/// The `nonce` is per EXECUTION rather than per job, because zombie recovery
+/// re-runs a job under its own id: keyed on that alone, the second attempt would
+/// overwrite the objects the first attempt's committed row still names, and then
+/// read those same keys back as displaced and drop them.
+fn object_key(
+    w_id: &str,
+    script_path: &str,
+    environment: &str,
+    job_id: &Uuid,
+    nonce: &Uuid,
+    artifact: &str,
+) -> String {
+    format!(
+        "wmill_dbt_state/{w_id}/{}/{job_id}.{nonce}/{artifact}",
+        digest(&format!("{script_path}|{environment}"))
+    )
+}
+
+/// Put an artifact where its size says it belongs: `(inline, key)`, exactly one
+/// of which is set.
+#[allow(clippy::too_many_arguments)]
+async fn store(
+    value: String,
+    artifact: &str,
+    environment: &str,
+    script_path: &str,
+    w_id: &str,
+    job_id: &Uuid,
+    nonce: &Uuid,
+) -> error::Result<(Option<String>, Option<String>)> {
+    if value.len() <= *DBT_STATE_INLINE_MAX_BYTES {
+        return Ok((Some(value), None));
+    }
+    let key = object_key(w_id, script_path, environment, job_id, nonce, artifact);
+    let size = value.len();
+    if put_object(&key, value).await? {
+        return Ok((None, Some(key)));
+    }
+    Err(Error::BadRequest(format!(
+        "this project's {artifact} is {}, past the {} this instance keeps in the database, and \
+         this instance has no object storage configured to hold it. Configure instance object \
+         storage, or raise DBT_STATE_INLINE_MAX_BYTES",
+        mib(size),
+        mib(*DBT_STATE_INLINE_MAX_BYTES),
+    )))
+}
+
+fn mib(bytes: usize) -> String {
+    format!("{:.1} MiB", bytes as f64 / (1024.0 * 1024.0))
+}
+
+/// Read an artifact back from whichever home the row names.
+async fn fetch(inline: Option<String>, key: Option<String>) -> error::Result<Option<String>> {
+    match (inline, key) {
+        (Some(inline), _) => Ok(Some(inline)),
+        (None, Some(key)) => get_object(&key).await.map(Some),
+        (None, None) => Ok(None),
+    }
+}
+
+/// Drop the objects nothing points at any more. Best-effort: an object left
+/// behind costs storage, and there is nothing useful to do about it in the path
+/// of a run that has already finished.
+async fn forget_objects(keys: &[Option<String>; 2]) {
+    for key in keys.iter().flatten() {
+        delete_object(key).await;
+    }
+}
+
+/// Whether the artifact was stored. `false` means this instance has no object
+/// storage to put it in.
+///
+/// The INSTANCE store, where every other internal worker artifact lives — bun
+/// bundles, python wheels, job logs, the global cache. Not the workspace's:
+/// that bucket is the one workspace members read and write through
+/// `job_helpers/*` and `wmill.write_s3_file`, so a manifest there is one any
+/// member could replace, and the next deferring run would hand dbt an
+/// attacker-chosen `defer_relation` for every unbuilt `ref()` while holding the
+/// script's warehouse credentials. Its compiled SQL would be readable there too,
+/// for a project the reader may have no access to.
+#[cfg(all(feature = "enterprise", feature = "parquet"))]
+async fn put_object(key: &str, value: String) -> error::Result<bool> {
+    use windmill_object_store::object_store_reexports::Path as ObjectPath;
+    let Some(store) = windmill_object_store::get_object_store().await else {
+        return Ok(false);
+    };
+    store
+        .put(&ObjectPath::from(key), bytes::Bytes::from(value).into())
+        .await
+        .map_err(|e| Error::internal_err(format!("storing the dbt state at {key}: {e:#}")))?;
+    Ok(true)
+}
+
+#[cfg(all(feature = "enterprise", feature = "parquet"))]
+async fn get_object(key: &str) -> error::Result<String> {
+    let Some(store) = windmill_object_store::get_object_store().await else {
+        return Err(missing_storage());
+    };
+    let bytes = windmill_object_store::attempt_fetch_bytes(store, key).await?;
+    String::from_utf8(bytes.to_vec())
+        .map_err(|e| Error::internal_err(format!("the stored dbt state is not valid UTF-8: {e}")))
+}
+
+#[cfg(all(feature = "enterprise", feature = "parquet"))]
+async fn delete_object(key: &str) {
+    use windmill_object_store::object_store_reexports::Path as ObjectPath;
+    let Some(store) = windmill_object_store::get_object_store().await else {
+        return;
+    };
+    if let Err(e) = store.delete(&ObjectPath::from(key)).await {
+        tracing::warn!("dbt: could not drop the superseded state object {key}: {e:#}");
+    }
+}
+
+#[cfg(not(all(feature = "enterprise", feature = "parquet")))]
+async fn delete_object(_key: &str) {}
+
+/// A build without the instance store carries no client at all, so an oversized
+/// artifact has nowhere but the row, and a row naming a key was written by a
+/// worker that did have one.
+#[cfg(not(all(feature = "enterprise", feature = "parquet")))]
+async fn put_object(_key: &str, _value: String) -> error::Result<bool> {
+    Ok(false)
+}
+
+#[cfg(not(all(feature = "enterprise", feature = "parquet")))]
+async fn get_object(_key: &str) -> error::Result<String> {
+    Err(missing_storage())
+}
+
+fn missing_storage() -> Error {
+    Error::BadRequest(
+        "the dbt state for this environment is in the instance's object storage, which this \
+         worker cannot reach: it is no longer configured, or this worker was built without \
+         object-storage support"
+            .to_string(),
+    )
+}
+
+/// The stored state this run resolves its unbuilt `ref()`s through, materialised
+/// into the job directory at `STATE_DIR`.
+#[derive(Clone, Debug)]
+pub(crate) struct Deferral {
+    /// The run that published the state, so the job log and the result can say
+    /// what this one deferred to.
+    pub published_by: Uuid,
+    /// Whether the state carries `run_results.json` beside its manifest. A build
+    /// recovered by node retry publishes without one, and that is the only file a
+    /// `result:` selector reads.
+    pub has_run_results: bool,
+}
+
+/// Materialise the environment's state so `--state` has a directory to read.
+///
+/// Refused rather than run without deferral where nothing is published: the run
+/// would build against a `ref()` resolving into the schema it writes, and fail
+/// deep inside dbt with a relation-not-found the caller has no way to connect
+/// back to a missing state.
+pub(crate) async fn prepare_deferral(
+    p: &PreparedProject,
+    w_id: &str,
+    job_dir: &str,
+    conn: &Connection,
+) -> error::Result<Deferral> {
+    if p.script_path.is_empty() {
+        return Err(Error::BadRequest(
+            "`defer` resolves a `ref()` through the state a previous run of this script \
+             published, so it needs a deployed script; a preview run has no environment to have \
+             published one"
+                .to_string(),
+        ));
+    }
+    // An environment is the warehouse, the target and where they RESOLVE to, and
+    // a `profiles.yml` that templates its schema or database resolves somewhere
+    // this runtime does not render. Two renderings would then share one
+    // environment, and a deferral after the value changed would resolve every
+    // unbuilt `ref()` through the previous location's manifest.
+    if p.templated_location {
+        return Err(Error::BadRequest(
+            "this project's profile selects its schema or database with a template, which dbt \
+             renders and Windmill does not — so two environments cannot be told apart and a \
+             deferral could resolve through the wrong one's manifest. Spell the target's schema \
+             and database literally to use `defer`"
+                .to_string(),
+        ));
+    }
+    let Some(state) = load(p, w_id, conn).await? else {
+        return Err(Error::BadRequest(format!(
+            "no dbt state is stored for this environment ({}), so a `ref()` this run does not \
+             build has no relation to resolve to. It is published by a successful run that adds \
+             nothing of its own: one overriding `select` or `vars` does not publish, and neither \
+             does any run of a descriptor that interpolates a `{{{{ }}}}` placeholder into `vars` \
+             or a `$var:` into `env` — those describe a model set the caller's arguments decided. \
+             Run this script once without `defer` and without overrides",
+            environment_label(p)
+        )));
+    };
+    let has_run_results = state.run_results.is_some();
+    write_state_dir(
+        &PathBuf::from(job_dir).join(STATE_DIR),
+        state.run_results.as_deref(),
+        StateManifest::Bytes(state.manifest),
+    )
+    .await?;
+    Ok(Deferral { published_by: state.job_id, has_run_results })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    // Every component earns its place: a deferral resolves relation NAMES, so
+    // state published where those names meant something else has to read as no
+    // state at all rather than as state that silently no longer fits.
+    #[test]
+    fn a_moved_profile_is_another_environment() {
+        let here = environment_key(Some("main"), Some("prod"), Some("analytics"), Some("wh"));
+        assert_eq!(
+            here,
+            environment_key(Some("main"), Some("prod"), Some("analytics"), Some("wh"))
+        );
+        assert_ne!(
+            here,
+            environment_key(Some("other"), Some("prod"), Some("analytics"), Some("wh"))
+        );
+        assert_ne!(
+            here,
+            environment_key(Some("main"), Some("dev"), Some("analytics"), Some("wh"))
+        );
+        assert_ne!(
+            here,
+            environment_key(Some("main"), Some("prod"), Some("marts"), Some("wh"))
+        );
+        assert_ne!(
+            here,
+            environment_key(
+                Some("main"),
+                Some("prod"),
+                Some("analytics"),
+                Some("other_db")
+            )
+        );
+    }
+
+    // A target name and a schema are both the user's own strings, so a component
+    // carrying the separator must not be able to spell another tuple's key: a
+    // profile moving between the two would read as the same environment and
+    // defer through the manifest of relations that are somewhere else.
+    #[test]
+    fn a_component_cannot_spell_another_environments_key() {
+        assert_ne!(
+            environment_key(Some("main"), Some("prod|analytics"), Some("scratch"), None),
+            environment_key(Some("main"), Some("prod"), Some("analytics|scratch"), None)
+        );
+        assert_ne!(
+            environment_key(Some("main"), Some("prod"), Some("a"), Some("b|c")),
+            environment_key(Some("main"), Some("prod"), Some("a|b"), Some("c"))
+        );
+        // A component the profile leaves out is the same environment as one it
+        // spells empty: there is no target named "".
+        assert_eq!(
+            environment_key(Some("main"), None, Some("a"), None),
+            environment_key(Some("main"), Some(""), Some("a"), Some(""))
+        );
+    }
+
+    // Two environments must not queue behind one advisory lock, which is what a
+    // digest folded through a signed parse did for every one whose top bit is
+    // set — half of them.
+    #[test]
+    fn each_environment_gets_its_own_publication_lock() {
+        let mut seen = std::collections::HashSet::new();
+        for i in 0..64 {
+            seen.insert(publication_lock(
+                "ws",
+                "f/a/p",
+                &format!("main|prod|s{i}|db"),
+            ));
+        }
+        assert_eq!(seen.len(), 64);
+        assert_eq!(
+            publication_lock("ws", "f/a/p", "e"),
+            publication_lock("ws", "f/a/p", "e")
+        );
+    }
+}

backend/windmill-worker/src/lib.rs

@@ -36,6 +36,7 @@ mod csharp_executor;
 mod dbt_engine;
 mod dbt_executor;
 mod dbt_profiles;
+mod dbt_state;
 #[cfg(feature = "private")]
 mod dedicated_worker_ee;
 mod dedicated_worker_oss;

docs/dbt-runtime.md

@@ -13,8 +13,8 @@ the dominant way dbt is orchestrated today.
 - **In**: run an unmodified dbt project synced into Windmill, one Windmill job per
   invocation, live per-model observability, dbt models as first-class assets in
   the existing asset graph.
-- **Out**: one Windmill job per dbt model, `state:modified` / slim CI,
-  `dbt docs` hosting, semantic layer, dbt platform integration.
+- **Out**: one Windmill job per dbt model, slim CI orchestration, `dbt docs`
+  hosting, semantic layer, dbt platform integration.
 - **CE**: the runtime, the manifest ingest, the asset graph and every piece of
   UI ship in CE, as do all adapters except two. Only the `mssql` and `oracle`
   adapters are EE, mirroring the native `ScriptLang` boundary (decision 21).
@@ -35,7 +35,7 @@ the dominant way dbt is orchestrated today.
 | 10 | Private repo auth | Not applicable: the project is synced, not fetched |
 | 11 | Asset kind | `dbt://<warehouse>/<schema>/<name>` — keyed on the relation, not on dbt's node id. See below |
 | 12 | Graph refresh | Deploy-time, re-ingested per run only when the descriptor is dynamic, plus an explicit `parse` of the editor's buffer. See below |
-| 13 | Manifest storage | Sidecar table for nodes/edges. Full manifest **not** stored — see below |
+| 13 | Manifest storage | Sidecar table for nodes/edges; the whole manifest is kept once per environment, for deferral — see below |
 | 14 | Metadata depth | Tests, strategy, tags, freshness, column descriptions. Column **lineage** is not in the manifest — see below |
 | 15 | Node rendering | Asset nodes per model plus one runnable node for the script |
 | 16 | Progress | Live, from the JSON event stream |
@@ -48,6 +48,7 @@ the dominant way dbt is orchestrated today.
 | 23 | Descriptor | `wm_dbt.yaml` inside the project, OPTIONAL. See below |
 | 24 | Warehouse | Configured on the workspace by name, `main` by default. See below |
 | 25 | Cascade direction | Into a relation, not out of a run: `// materialize manual dbt://…` declares a write from any language but dbt's own and wakes `# on dbt://…` subscribers; a finished dbt run still does not dispatch. See "No cascade *from* dbt" |
+| 26 | Deferral | A durable state per environment, published by the runs whose relations are the script's; `defer` is a per-run toggle. See below |
 
 ## Decision 1: engine toggle, and why the shipped default is not Fusion yet
 
@@ -1228,13 +1229,343 @@ block, since that is what `dbt_run_state` saves and `invocation_args` publishes.
    without failing. Overriding this would make the same project behave differently
    on Windmill than locally, breaking the core promise.
 
+## Durable state per environment, and what defers to it
+
+`dbt --defer --state <dir>` resolves a `ref()` the run does not build to the
+relation the manifest in `<dir>` names, instead of to the schema this run writes
+into. That is what lets one model be rebuilt into a scratch schema without
+rebuilding everything above it, and it needs a manifest of the environment the
+project actually lives in.
+
+Nothing that already existed could supply one. `dbt_run_state` answers a
+different question — it holds the LAST run whatever its outcome, keyed by the
+principal, so `dbt retry` can resume its failures — and the worker-local
+generations behind it are a cache: the next run of a project usually lands on a
+worker holding neither artifact. So the state is its own table,
+`dbt_environment_state`, one row per (workspace, script path, environment),
+holding `manifest.json` and `run_results.json` from the last SUCCESSFUL run.
+Success is half of the contract: a relation a later run defers to has to exist.
+
+### The environment is the warehouse, the target and where they resolve to
+
+The workspace warehouse's name, the target dbt actually runs, and the database
+and schema that target resolves to — the pair `relation_root` reports to the
+graph's drift check. Each component is length-prefixed rather than joined on a
+separator — `<warehouse>|<target>|<schema>|<database>`, each written `<len>:<value>`,
+so `main`/`prod`/`analytics`/`dbt_wh_defer` is stored as
+`4:main|4:prod|9:analytics|12:dbt_wh_defer`. A target name and a schema are both
+the user's own strings, so `prod|analytics` + `scratch` and `prod` +
+`analytics|scratch` would otherwise be one key, and a profile moving between them
+would read as the same environment rather than as one nothing has published. What
+a message names is spelled out instead, never the encoded key.
+
+The target is the EFFECTIVE one, not the descriptor's `profile.target`: a
+descriptor naming none inherits the workspace warehouse's, or the default in the
+project's own `profiles.yml`, so reading the descriptor's would file every
+inherited target under one empty name — and a `target.name` macro decides where a
+model is built.
+
+The last two are in the key because deferring is resolving a relation NAME. A
+warehouse repointed at another database, or a `profile.schema` moved by a
+redeploy, keeps the first two while putting every relation somewhere else, and a
+manifest is a list of relation names — there is no other way to notice. Keyed on
+the first two alone, such a move would hand the next deferring run the names of
+relations that are no longer there. Keyed on all four, it reads as an
+environment nothing has published yet, which is what it is.
+
+What the key deliberately does NOT carry is the resolved connection. That is the
+`profile_digest` a retry is held to, and it moves when a password is rotated,
+which moves no relation; a warehouse pointing somewhere else entirely is
+decision 11's accepted limitation, spelled the same way here as everywhere else.
+
+Today one script has one environment, because a descriptor fixes both the
+warehouse and the target and a run cannot override either. The key is what makes
+the *later* item — fork and preview environments — an addition rather than a
+migration, and what makes a profile move detectable now.
+
+### Which runs publish it
+
+A successful `build` that did not itself defer, and whose graph becomes what the
+script owns (`GraphRefresh::publishes_ownership`) — the same condition as the
+graph's and the same reason: an invocation that scoped its own model set — a
+`vars` or `select` override, or a descriptor dynamic by construction — describes
+where the CALLER put those relations, not where this project's models live.
+Publishing it would point every later deferral at one caller's scratch schema.
+
+**A run that deferred never publishes, whatever narrowed it**, and that is a
+separate condition rather than a consequence of the first. A deferring run built
+some of the relations its manifest names and resolved the rest out of the state
+it read, so recording that manifest would claim relations nothing built — and a
+model renamed since would be recorded under a name only a full build creates,
+breaking every later deferral until one repairs it. `publishes_ownership` cannot
+see this: it reads the caller's overrides, and a descriptor that already narrows
+`select` needs none.
+
+A `retry` publishes nothing. Its `run_results.json` names only the nodes it
+redid, so the environment would come to claim a run of a handful of models. The
+environment's state is therefore the last full successful build, exactly as dbt
+Cloud's "last successful run" is, and a run recovered by a retry leaves it at
+the previous one.
+
+The AUTOMATIC in-job node retry is the same artifact under a different name: a
+build it recovers is a successful build, but the `run_results.json` on disk is
+the retry's. Such a run publishes the manifest **without** results, rather than
+with a set describing some other slice of the build — the manifest is a function
+of the project rather than of what ran, so deferral is unaffected. A `result:`
+selector is the one thing left with nothing to read, and it is refused by name
+against such a publication rather than passed to dbt (see "Selectors that read
+the state" below).
+
+Under `test_behavior: after_all` the stored `run_results.json` is the test
+phase's, because that is what the second invocation leaves in the target
+directory — the same artifact a local `dbt run && dbt test` leaves behind.
+
+**What that condition means for what the artifacts may carry**, and why this
+table is keyed by environment where `dbt_run_state` is keyed by principal. dbt
+records the invocation's flags into `run_results.json`, and Windmill resolves
+`$var:` / `$res:` references before dbt sees them — which is exactly why the
+retry state is per-principal, so one caller's resolved `select` and `vars` are
+not restorable by the next. Here they cannot be one caller's: a publishing run
+added nothing of its own, and a descriptor that interpolates a `{{ }}`
+placeholder into `vars` never publishes at all, so what is recorded is the
+descriptor's own arguments — the script's content, which anyone entitled to run
+it may already read. Widen the publish condition and that stops being true.
+
+### Where the blob goes
+
+`run_results.json` is small; `manifest.json` is not, and grows with the project
+(535 KB on a two-model fixture). Each takes the same two homes: inline in the row
+under `DBT_STATE_INLINE_MAX_BYTES` (8 MiB), and the INSTANCE's object storage
+above it, with the row keeping the key. Inline is what makes the feature work on
+an instance that has configured no storage at all; the ceiling is what stops one
+project's manifest from becoming a multi-megabyte row rewritten by every run. A
+project past the ceiling with no storage configured is told so, in the job log,
+naming the setting and the variable — the run itself still succeeds, since
+losing the state costs the next deferral rather than the build that just ran.
+
+**The instance store, not the workspace's**, which is where every other internal
+worker artifact already lives (bun bundles, python wheels, job logs, the global
+cache). The workspace bucket is the one members read and write through
+`job_helpers/*` and `wmill.write_s3_file` with a caller-supplied key, and only
+`volumes/` is reserved there — so a manifest under it is one any member could
+replace, and the next deferring run would hand dbt an attacker-chosen
+`defer_relation` for every unbuilt `ref()` while holding the script's warehouse
+credentials. Its compiled SQL would be readable there too, for a project the
+reader may have no access to. The consequence to know: a project past the ceiling
+needs the instance store configured, which is an EE feature, so on CE the ceiling
+is the limit and `DBT_STATE_INLINE_MAX_BYTES` is how it moves.
+
+Each publication writes its OWN keys
+(`wmill_dbt_state/<workspace>/<digest of path and environment>/<job>.<nonce>/<artifact>`)
+and the row switches to them in one statement, so an upload never overwrites an
+artifact the committed row still names: a run that fails between its two uploads,
+or between them and its row, leaves the state pointing at the pair it already
+had. The objects the commit displaced are dropped afterwards, never before, since
+a reader that has already read the row is about to fetch them; a reader that
+loses that race re-reads for as long as the row keeps MOVING, rather than
+reporting a state that is there. A reader takes no lock, so successive
+publications can each overtake one; an unmoved row whose objects are gone is the
+error that means what it says, and a bound on the re-reads is the other, for a
+project republishing faster than a run can read. What a publication uploaded and then could not commit is dropped on the
+way out — except after a commit that REPORTED an error, where what was lost may
+be only the acknowledgement: dropping then would leave a committed row naming
+objects that are gone, so an orphan is the cheaper side to take.
+
+The path and the environment are only a prefix of that key. The row is what says
+where an artifact is, which is why state can travel with a renamed script and go
+on naming objects under the old path's digest. The rest of the key is the job and
+a per-EXECUTION nonce — zombie recovery re-runs a job under its own id, so keyed
+on that alone a second attempt would overwrite the objects the first attempt's
+committed row still names, then read those keys back as displaced and drop them.
+
+Publishers of one environment serialize on `pg_advisory_xact_lock`, so only one
+of them settles the row and the objects it displaces at a time — an advisory lock
+rather than the row's, because the first publish of an environment has no row to
+lock and is exactly when two runs of a newly deployed script are most likely to
+race.
+
+### Retention
+
+None, deliberately, and this is where it differs from the graph tables next
+door. Those are pruned by age by the dbt runs themselves because their reader is
+a transient run page. This one holds a single row per script per environment,
+replaced in place, so it does not grow with runs — and its reader is every later
+run of that script, so a project that runs monthly must still find last month's
+state. It goes with the script instead: a path no live dbt version occupies any
+more clears it, alongside `dbt_run_state` (`clear_dbt_script_state`,
+`clear_dbt_script_state_if_path_retired`).
+
+The write carries a guard of its own, and it names the VERSION rather than the
+path: the live dbt script there must be the one this job ran, or a later version
+of it (`hash = $n OR $n = ANY(parent_hashes)`). "Some live dbt script is here" —
+which is what the retry state settles for — is also satisfied by a script created
+at a path this one was renamed away from, and this job's manifest would then
+become that project's deferral state. A preview names no version and so publishes
+nothing, which is right for a run of content that was never deployed.
+
+The job's KIND is checked beside it, because a preview carries a caller-supplied
+`script_hash` into `runnable_id` (`run_preview_script`): the version alone would
+let anyone who may run a job publish arbitrary content as a deployed script's
+state. A flow or app step naming a deployed dbt script by path is an ordinary
+`script` job carrying that script's own hash, so it publishes like any other run;
+only INLINE flow code is a `FlowScript`, and that has no deployed version to
+publish for.
+
+That guard HOLDS the script row (`FOR SHARE`) for the rest of the publication, so
+a rename, archive or delete of the path either waits for it or is seen by it.
+Read unlocked, it leaves a window where the lifecycle clear finds no row to take,
+finishes, and the publication then commits state at a path a new script goes on
+to occupy. The script row is taken before the sidecar, which is the order every
+other dbt writer takes and what keeps the two off a deadlock.
+
+An artifact too large for its row is left in the store when the row is cleared,
+as a deleted script leaves its bundle: reaching it from the delete would mean an
+object-store client in `windmill-common` and a delete that has to land after the
+caller's transaction commits, for one object per environment of a script that is
+gone.
+
+### Asking for it
+
+`defer` is a field on the `build` command block, defaulting to the descriptor's
+own `defer:`. A per-run toggle rather than a descriptor-only setting, because the
+run that publishes an environment's state and the run that defers to it are two
+invocations of ONE script (decision 6: N scripts means N projects): a project
+that could only defer by descriptor could never populate the state it reads.
+
+A project whose profile selects its schema or database with a TEMPLATE — either
+delimiter, since dbt renders `{% … %}` blocks as well as `{{ … }}` — is refused a
+deferral outright, and publishes no state either: dbt renders those and Windmill
+does not, so two renderings resolve to one `relation_root`, and a
+deferral after the value changed would resolve every unbuilt `ref()` through the
+previous location's manifest. Both sides, because a published template would sit
+under a key a literal profile shares, and de-templating later would make that
+stale manifest readable as the new location's. It covers a project-owned
+`profiles.yml`, a `dbt_profile` resource — one block of the user's own file,
+copied through unchanged — and a `profile.schema` written as given. Plainly
+absent is different: that is the adapter's default, which does not move.
+
+A run that asks to defer with nothing published is refused, naming the
+environment and the runs that cannot publish one. The alternative — running
+without deferral — fails deep inside dbt with a relation-not-found the caller has
+no way to connect back to a missing state. An agent worker is refused the same
+way and for a reason it can act on: it reaches the database only through the API,
+which does not expose this table.
+
+A `show` defers too, and every engine takes the flags on it. It compiles the
+model it previews, so a model whose upstream this environment built and this run
+did not is exactly the case a deferral exists for. So does the `dbt ls` that
+resolves what a run's selection owns, without which a `result:` selector — which
+reads `run_results.json` out of the state directory, and which `select` passes to
+dbt verbatim — would fail before the build that would have honoured it.
+
+The result carries `deferred_to`, the run whose state was used. Without it what
+a deferring run built against is unrecoverable, since the next successful run of
+that environment replaces the state.
+
+### Selectors that read the state, and why they are refused rather than passed
+
+`--state` also feeds dbt's own selector methods, so publishing the state is what
+makes `state:modified+`, `state:new` and `result:error+` resolve at all. Only a
+deferring run is handed the directory, so a `state:` or `result:` method in
+`select` or `exclude` without `defer` is refused before dbt starts.
+
+Refused, rather than left to dbt, because the engines disagree about it and two
+of the three disagree silently. Given a state selector and no `--state`,
+dbt-core 1.x raises (`Got a state selector method, but no comparison manifest`,
+exit 2), but dbt-sa-cli 2.x and fusion read a MISSING state as an EMPTY one and
+exit 0: `state:modified` then selects nothing and the run reports success having
+built nothing, while `state:new` selects everything, because against an empty
+state every node is new. A scheduled run that quietly stops doing work, or
+quietly rebuilds the project, is the failure this state exists to prevent.
+
+From the DESCRIPTOR they are refused whether or not the run defers, and the
+message says so. That selection is also what decides which nodes the script owns,
+and the deploy resolves it before any run exists, with no state to compare
+against. "Whatever changed last" is not an ownership answer. They describe one
+run, so they belong in a run's own `select`.
+
+`source_status:` is refused under any setting: it compares `sources.json`, which
+`dbt source freshness` writes and no run publishes here, so there is nothing to
+compare against even while deferring.
+
+Two more refusals follow from the same argument, that a selector with nothing to
+read must say so rather than resolve to a silent answer:
+
+- A `result:` method while deferring to a state that carries **no**
+  `run_results.json`. Publishing that is deliberate — a build recovered by
+  automatic node retry stores the manifest alone, its results describing the
+  retried nodes rather than the build ("Which runs publish it") — so `defer`
+  being on is not enough to know the file is there. Answerable only once the
+  state is loaded, so it is checked right after, naming the run that published.
+- Any of them on a `parse`. A parse resolves a selection to store the graph and
+  never defers, so `defer` would not hand it a state at any setting, and the
+  remedy the other refusal offers would lead nowhere. It says that instead.
+
+Matching nothing is then an ordinary outcome for these methods, and for no
+others. `state:modified+` selects the empty set exactly when nothing changed
+since the published state, which is the answer a CI run wants, so a selection
+naming a `state:` or `result:` method may resolve to no nodes. Such a run scoped
+its own selection, so what it stores is a snapshot of its own and never what the
+script owns, and nothing is un-wired by the empty set.
+
+The exemption is by METHOD, not by who chose the selection. Exempting every
+caller-chosen one would take a misspelled model name, which resolves to nothing
+just as surely, and report it as a build that did its work. An ordinary selection
+matching nothing stays refused, from a run as from the descriptor — from the
+descriptor because that one also decides ownership.
+
+Only what `select` and `exclude` spell directly. A method reached through a
+`selectors.yml` definition is named nowhere the worker reads, and dbt's own
+behaviour — including the silent one — is what stands there.
+
+### `--state` is also a retry's own argument, and that is a trap
+
+`dbt retry` reads the run it RESUMES from `--state`. Handed the deferral's
+directory it resumes the successful run stored there, finds nothing failed, and
+reports a green retry having rebuilt nothing — silently, on dbt-core 1.x, which
+warns and exits 0.
+
+dbt-core 1.x has `--defer-state`, the deferral-only half of the pair, so a retry
+there passes that and leaves `--state` alone. The Rust engines do not have it,
+and a run that deferred is refused a retry on them, before the build: the
+alternative is rebuilding the failed nodes with every `ref()` resolving into the
+schema this run writes into, which for the narrowed run a deferral exists to
+serve means writing them somewhere they do not belong. The automatic in-job node
+retry is dropped for the same reason and says so in the log.
+
+The state directory is passed RELATIVE (`wm_dbt_state`, beside `wm_target` in the
+job directory). dbt records the invocation's flags into `run_results.json` and a
+later `dbt retry` restores them, so an absolute path would name the job directory
+of the run being resumed, which is gone by then. Relative, it resolves against
+the project root — whichever job directory the retry landed in.
+
+Three engine facts found while wiring this up, all worth knowing before filing a
+bug against the feature. `dbt retry` on dbt-core 2.x restores **neither** the
+resumed invocation's `--vars` nor its deferral: it re-parses with the current
+(empty) ones, so a retry of a run that overrode `vars` rebuilds into the
+descriptor's schema rather than the run's. That is independent of deferral and
+predates it; the refusal above stops the deferring case from being the way it is
+discovered. `dbt show` on either Rust engine prints a bare JSON array where
+dbt-core frames it as `{"node": …, "show": […]}`, which `run_show` is written
+against — so a preview there fails to parse whether or not it defers, and the
+deferral itself resolves correctly under it. And neither Rust engine reached
+dbt's own service-backed State (`--manage-state`) on any run measured here, so no
+flag is passed to disable it.
+
+Because `select` reaches dbt verbatim, a deferring run also has a `--state`
+directory for `result:` selectors, which is why `run_results.json` is stored
+beside the manifest rather than the manifest alone.
+
 ## Two decisions the implementation narrowed
 
-**Decision 13 — no S3 copy of the manifest.** The sidecar holds every field the
-graph renders; nothing reads a stored `manifest.json`, so writing one to S3
-would be an unread copy of data that is already reproducible by redeploying (or,
-for a dynamic descriptor, by the next run). Worth adding the day something needs the
-parts the sidecar drops — compiled SQL, macro definitions — and not before.
+**Decision 13 — the manifest is stored once per environment, not per version.**
+The sidecar holds every field the graph renders, so a copy of `manifest.json`
+bought the graph nothing: it is reproducible by redeploying, or for a dynamic
+descriptor by the next run. Deferral is the reader that changed that — it
+resolves an unbuilt `ref()` through a manifest, and one on worker-local disk
+answers for a machine's history rather than for the environment. So exactly one
+manifest is kept per (script, environment), replaced by each successful run,
+rather than one per version (see "Durable state per environment" above).
 
 **Decision 14 — column lineage is not available.** The decision assumed
 `manifest.json` carries column-to-column edges; it does not, in either core
@@ -1280,7 +1611,10 @@ render through the existing `RunnableNode.svelte` / `AssetNode.svelte` /
 on the canvas mid-run. `record_materialization` per model. Profile and select
 pickers in the editor. Per-model failure triage in the run view.
 
-**Phase 4 (not in this PR).** `--defer` and `state:modified`. Partition and
+**Phase 4 (not in this PR).** Slim CI: the fork and preview environments a
+deferral would name instead of its own. The selectors themselves are here, since
+`state:` and `result:` read the published state like any deferral does; what is
+missing is a per-branch environment to compare a CI run against. Partition and
 backfill integration so `BackfillRangeDialog.svelte` works on dbt models.
 `wmill dbt import <dag.py>` reading `DbtDag(...)` kwargs.
 
@@ -1317,6 +1651,16 @@ Against a real dbt project (jaffle_shop shape) and the local Postgres:
    own `profiles.yml` with env-var injection.
 11. **Caching**: a second run reuses the cached `dbt_packages/` with no network
     fetch.
+12. **Deferral**: a full run publishes the environment's state; a second run
+    that builds one downstream model into another schema resolves its unbuilt
+    `ref()` to the relation the state names, where the same run without `defer`
+    fails with relation-not-found.
+13. **State selectors**: with a state published, `state:modified+` selects
+    nothing while the project is unchanged and exactly the changed model and its
+    children after one is edited. Without `defer` it is refused rather than
+    passed, and a `result:` selector against a state published by a
+    node-retry-recovered build is refused too, that one carrying no
+    `run_results.json`.
 
 Keep only tests that pin behavior a future change could break. Per AGENTS.md,
 delete development scaffolding before marking the PR ready.

frontend/src/lib/script_helpers.ts

@@ -1370,6 +1370,11 @@ test_behavior: build
 vars: {}
 threads: 4
 full_refresh: false
+# Resolve a ref() this run does not build through the state the last successful
+# run of this environment published, instead of through the schema it writes
+# into. The default for the run form's toggle: the run that publishes the state
+# and the run that defers to it are two invocations of this one script.
+defer: false
 # Rebuild the nodes a failed build left failed or skipped, in this same job,
 # before reporting failure. dbt confines a failure to its own subtree, so a
 # transient warehouse error costs those nodes rather than the whole project.