From b33b9897b567bcace2497acbfcb648342e32bb80 Mon Sep 17 00:00:00 2001
From: Zach Daniel <zach@zachdaniel.dev>
Date: Sun, 4 Oct 2026 10:52:24 +0200
Subject: [PATCH] fix: split temporal periods without FOR PORTION OF, on
 PostgreSQL 18

PostgreSQL reverted UPDATE/DELETE FOR PORTION OF from 19 (a4b26b8f7cd),
because concurrent writes under READ COMMITTED could lose updates. Temporal
writes now split versions themselves in one statement, using PG18's
RETURNING old, and lock the versions they split first: if a concurrent
write changed one, the statement writes nothing and reruns. Upserts that
race to insert the same record skip on the WITHOUT OVERLAPS primary key and
retry that key. After 25 conflicts, AshPostgres.Temporal.WriteConflict.

- migrations emit temporal DDL unconditionally; VerifyTemporal requires the
  repo's min_pg_version/0 to be at least 18
- temporal upsert errors go through handle_raised_error
- test suite: temporal resources on AshPostgres.TemporalTestRepo, only
  migrated on PG18+; test repo port is PG_PORT (CI broke on 5433)
- docs: temporal resources topic
---
 config/config.exs                             |  27 +-
 .../topics/advanced/temporal-resources.md     |  85 +++
 lib/data_layer.ex                             |  22 +-
 lib/migration_generator/operation.ex          |  47 +-
 lib/temporal.ex                               | 560 +++++++++++++-----
 lib/temporal/write_conflict.ex                |  22 +
 .../validate_temporal_references.ex           |   6 +-
 lib/verifiers/verify_temporal.ex              |  91 ++-
 mix.exs                                       |   4 +-
 test/support/repo_case.ex                     |  13 +-
 test/support/temporal/domain.ex               |   1 +
 test/support/temporal/event.ex                |   2 +-
 test/support/temporal/member.ex               |   2 +-
 test/support/temporal/subscription.ex         |   4 +-
 test/support/temporal/tier.ex                 |   2 +-
 test/support/temporal_test_repo.ex            |  31 +
 test/temporal_concurrency_test.exs            | 185 ++++++
 test/temporal_identity_test.exs               |  20 +-
 test/temporal_migration_generator_test.exs    |  26 +-
 test/temporal_test.exs                        | 153 +++--
 test/test_helper.exs                          |   5 +
 .../event/20261004073135.json}                |   4 +-
 .../event/20261004073135.json.license}        |   0
 .../temporal_test_repo/extensions.json        |   8 +
 .../extensions.json.license}                  |   0
 .../member/20261004073136.json}               |   4 +-
 .../member/20261004073136.json.license}       |   0
 .../subscription/20261004073137.json}         |   4 +-
 .../subscription/20261004073137.json.license} |   0
 .../tier/20261004073138.json}                 |   4 +-
 .../tier/20261004073138.json.license}         |   0
 .../test_repo/event/20260824043711.json       |  57 --
 .../subscription/20260627032021.json          | 119 ----
 .../subscription/20260824043713.json          | 121 ----
 .../subscription/20260927024130.json.license  |   3 -
 .../test_repo/tier/20260627032020.json        |  60 --
 .../tier/20260627032020.json.license          |   3 -
 .../tier/20260824043712.json.license          |   3 -
 ...4073133_migrate_resources_extensions_1.exs | 136 +++++
 .../20261004073134_migrate_resources1.exs}    |  54 +-
 .../20260926192338_migrate_resources76.exs    |  48 --
 .../20260927024129_migrate_resources77.exs    |  28 -
 42 files changed, 1204 insertions(+), 760 deletions(-)
 create mode 100644 documentation/topics/advanced/temporal-resources.md
 create mode 100644 lib/temporal/write_conflict.ex
 create mode 100644 test/support/temporal_test_repo.ex
 create mode 100644 test/temporal_concurrency_test.exs
 rename test_priv/resource_snapshots/{test_repo/event/20260627032019.json => temporal_test_repo/event/20261004073135.json} (89%)
 rename test_priv/resource_snapshots/{test_repo/event/20260627032019.json.license => temporal_test_repo/event/20261004073135.json.license} (100%)
 create mode 100644 test_priv/resource_snapshots/temporal_test_repo/extensions.json
 rename test_priv/resource_snapshots/{test_repo/event/20260824043711.json.license => temporal_test_repo/extensions.json.license} (100%)
 rename test_priv/resource_snapshots/{test_repo/member/20260926192339.json => temporal_test_repo/member/20261004073136.json} (92%)
 rename test_priv/resource_snapshots/{test_repo/member/20260926192339.json.license => temporal_test_repo/member/20261004073136.json.license} (100%)
 rename test_priv/resource_snapshots/{test_repo/subscription/20260927024130.json => temporal_test_repo/subscription/20261004073137.json} (95%)
 rename test_priv/resource_snapshots/{test_repo/subscription/20260627032021.json.license => temporal_test_repo/subscription/20261004073137.json.license} (100%)
 rename test_priv/resource_snapshots/{test_repo/tier/20260824043712.json => temporal_test_repo/tier/20261004073138.json} (90%)
 rename test_priv/resource_snapshots/{test_repo/subscription/20260824043713.json.license => temporal_test_repo/tier/20261004073138.json.license} (100%)
 delete mode 100644 test_priv/resource_snapshots/test_repo/event/20260824043711.json
 delete mode 100644 test_priv/resource_snapshots/test_repo/subscription/20260627032021.json
 delete mode 100644 test_priv/resource_snapshots/test_repo/subscription/20260824043713.json
 delete mode 100644 test_priv/resource_snapshots/test_repo/subscription/20260927024130.json.license
 delete mode 100644 test_priv/resource_snapshots/test_repo/tier/20260627032020.json
 delete mode 100644 test_priv/resource_snapshots/test_repo/tier/20260627032020.json.license
 delete mode 100644 test_priv/resource_snapshots/test_repo/tier/20260824043712.json.license
 create mode 100644 test_priv/temporal_test_repo/migrations/20261004073133_migrate_resources_extensions_1.exs
 rename test_priv/{test_repo/migrations/20260824043710_migrate_resources78.exs => temporal_test_repo/migrations/20261004073134_migrate_resources1.exs} (51%)
 delete mode 100644 test_priv/test_repo/migrations/20260926192338_migrate_resources76.exs
 delete mode 100644 test_priv/test_repo/migrations/20260927024129_migrate_resources77.exs

diff --git a/config/config.exs b/config/config.exs
index ef2a0fbc..08863ea2 100644
--- a/config/config.exs
+++ b/config/config.exs
@@ -27,6 +27,7 @@ if Mix.env() == :test do
   config :elixir, :time_zone_database, Tz.TimeZoneDatabase
   config :ash_postgres, AshPostgres.TestRepo, log: false
   config :ash_postgres, AshPostgres.TestNoSandboxRepo, log: false
+  config :ash_postgres, AshPostgres.TemporalTestRepo, log: false
 
   config :ash, :validate_domain_resource_inclusion?, false
   config :ash, :validate_domain_config_inclusion?, false
@@ -47,11 +48,13 @@ if Mix.env() == :test do
     AshPostgres.Test.OperatorExpression.IntListType
   ]
 
+  pg_port = String.to_integer(System.get_env("PG_PORT", "5432"))
+
   config :ash_postgres, AshPostgres.TestRepo,
     username: "postgres",
     database: "ash_postgres_test",
     hostname: "localhost",
-    port: 5433,
+    port: pg_port,
     pool: Ecto.Adapters.SQL.Sandbox,
     types: AshPostgres.Test.PostgrexTypes
 
@@ -60,7 +63,7 @@ if Mix.env() == :test do
     password: "postgres",
     database: "ash_postgres_dev_test",
     hostname: "localhost",
-    port: 5433,
+    port: pg_port,
     migration_primary_key: [name: :id, type: :binary_id],
     pool: Ecto.Adapters.SQL.Sandbox
 
@@ -73,7 +76,7 @@ if Mix.env() == :test do
     username: "postgres",
     database: "ash_postgres_test",
     hostname: "localhost",
-    port: 5433
+    port: pg_port
 
   # sobelow_skip ["Config.Secrets"]
   config :ash_postgres, AshPostgres.TestNoSandboxRepo, password: "postgres"
@@ -81,8 +84,24 @@ if Mix.env() == :test do
   config :ash_postgres, AshPostgres.TestNoSandboxRepo,
     migration_primary_key: [name: :id, type: :binary_id]
 
+  config :ash_postgres, AshPostgres.TemporalTestRepo,
+    username: "postgres",
+    password: "postgres",
+    database: "ash_postgres_test",
+    hostname: "localhost",
+    port: pg_port,
+    pool: Ecto.Adapters.SQL.Sandbox,
+    types: AshPostgres.Test.PostgrexTypes
+
+  # Temporal resources need PostgreSQL 18+, so their repo is only migrated when testing
+  # against it. An unset `PG_VERSION` means the newest (see `AshPostgres.TestRepo`).
+  {pg_version, _} = Integer.parse(System.get_env("PG_VERSION", "19"))
+  temporal_repos = if pg_version >= 18, do: [AshPostgres.TemporalTestRepo], else: []
+
   config :ash_postgres,
-    ecto_repos: [AshPostgres.TestRepo, AshPostgres.DevTestRepo, AshPostgres.TestNoSandboxRepo],
+    ecto_repos:
+      [AshPostgres.TestRepo, AshPostgres.DevTestRepo, AshPostgres.TestNoSandboxRepo] ++
+        temporal_repos,
     ash_domains: [
       AshPostgres.Test.Domain,
       AshPostgres.MultitenancyTest.Domain,
diff --git a/documentation/topics/advanced/temporal-resources.md b/documentation/topics/advanced/temporal-resources.md
new file mode 100644
index 00000000..2bc50e69
--- /dev/null
+++ b/documentation/topics/advanced/temporal-resources.md
@@ -0,0 +1,85 @@
+<!--
+SPDX-FileCopyrightText: 2020 Zach Daniel
+
+SPDX-License-Identifier: MIT
+-->
+
+# Temporal Resources
+
+Temporal resources keep every version of every record, each valid for a period of time.
+How to define and use them is covered in Ash's
+[temporal resources guide](https://hexdocs.pm/ash/temporal-resources.html). This page covers
+what AshPostgres needs and does for them.
+
+## Requirements
+
+Temporal resources need **PostgreSQL 18 or later**, and the repo has to say so:
+
+```elixir
+def min_pg_version do
+  %Version{major: 18, minor: 0, patch: 0}
+end
+```
+
+They also need the `btree_gist` extension, because the temporal primary key is backed by a
+GiST exclusion constraint:
+
+```elixir
+def installed_extensions do
+  ["ash-functions", "btree_gist"]
+end
+```
+
+A temporal resource whose repo doesn't declare both fails to compile.
+
+## What the migration generator creates
+
+For a temporal resource, `mix ash_postgres.generate_migrations` creates:
+
+- the period column, a `tstzrange`
+- `PRIMARY KEY (id, valid_at WITHOUT OVERLAPS)`, so a record can have many versions, but never
+  two valid at the same time
+- `PERIOD` foreign keys for relationships between temporal resources
+- identities as exclusion constraints, so they're unique at every instant rather than across
+  all of time
+
+## How writes split versions
+
+A write at an instant changes only the part of each version from that instant onward, and
+keeps the rest as it was. SQL:2011 calls this `UPDATE/DELETE ... FOR PORTION OF`.
+
+That was planned for PostgreSQL 19, but it was reverted before release, because concurrent
+writes under `READ COMMITTED` could silently lose part of an update. So AshPostgres splits
+versions itself, in a single statement: it writes the slice the write covers, and puts back
+the rest of the version as new rows. Once PostgreSQL ships `FOR PORTION OF` natively,
+AshPostgres will use it.
+
+## Concurrent writes
+
+Under `READ COMMITTED`, each temporal write locks the versions it's about to split, and checks
+that they're the versions it expected. If another transaction changed one of them in the
+meantime, the write does nothing and runs again, holding the locks it already took. Two upserts
+that race to create the same record are retried the same way.
+
+After 25 conflicts in a row, which takes many transactions writing the same slice of the same
+record at once, the action fails with `AshPostgres.Temporal.WriteConflict`. Nothing was
+written, so retrying it is safe.
+
+Under `REPEATABLE READ` or `SERIALIZABLE`, PostgreSQL raises a serialization failure for these
+cases itself, and retrying is up to you.
+
+## Upgrading from earlier versions
+
+Migrations generated by earlier versions of AshPostgres only created the temporal primary key,
+temporal identities and `PERIOD` foreign keys on PostgreSQL 19, and fell back to plain keys
+otherwise. If you ran them on PostgreSQL 18, your tables have the plain keys. Check with:
+
+```sql
+SELECT conname, pg_get_constraintdef(oid)
+FROM pg_constraint
+WHERE conrelid = 'your_table'::regclass;
+```
+
+If the primary key doesn't include `WITHOUT OVERLAPS`, write a migration that drops it and
+adds `PRIMARY KEY (id, valid_at WITHOUT OVERLAPS)`, and do the same for identities and
+`PERIOD` foreign keys.
diff --git a/lib/data_layer.ex b/lib/data_layer.ex
index aa097cbc..a3138770 100644
--- a/lib/data_layer.ex
+++ b/lib/data_layer.ex
@@ -2018,7 +2018,7 @@ defmodule AshPostgres.DataLayer do
 
                 if changeset.context[:data_layer][:use_atomic_update_data?] &&
                      Enum.count_until(results, 2) == 1 do
-                  # For temporal resources `FOR PORTION OF` truncates `valid_at` to
+                  # For temporal resources the split truncates `valid_at` to
                   # the affected slice; include it so the returned record reflects
                   # the slice rather than the pre-split range.
                   temporal_attrs =
@@ -2416,13 +2416,25 @@ defmodule AshPostgres.DataLayer do
   @impl true
   def bulk_create(resource, stream, options) do
     if options[:upsert?] && Ash.Resource.Info.temporal?(resource) do
-      # Temporal upsert (single or bulk): one atomic `FOR PORTION OF`-update-or-insert CTE
+      # Temporal upsert (single or bulk): one atomic split-or-insert CTE
       # per `as_of` (see `AshPostgres.Temporal.upsert_all/5`).
       changesets = Enum.to_list(stream)
       repo = AshSql.dynamic_repo(resource, AshPostgres.SqlImplementation, Enum.at(changesets, 0))
       keys = options[:upsert_keys] || Ash.Resource.Info.primary_key(resource)
 
-      AshPostgres.Temporal.upsert_all(repo, resource, changesets, keys, options[:upsert_fields])
+      try do
+        AshPostgres.Temporal.upsert_all(repo, resource, changesets, keys, options[:upsert_fields])
+      rescue
+        e ->
+          changeset = Ash.Changeset.new(resource)
+
+          handle_raised_error(
+            e,
+            __STACKTRACE__,
+            {:bulk_create, ecto_changeset(changeset.data, changeset, :create, repo, false)},
+            resource
+          )
+      end
     else
       do_bulk_create(resource, stream, options)
     end
@@ -2921,7 +2933,7 @@ defmodule AshPostgres.DataLayer do
     end)
   end
 
-  # The `FOR PORTION OF` portion for a temporal mutation. A range on the changeset
+  # The portion a temporal mutation splits out. A range on the changeset
   # names it outright and is the only place a period survives; otherwise the query's
   # threaded `as_of` (set for both single and bulk operations), the changeset's
   # `as_of` (single), or the wall clock each give `[as_of, ∞)`.
@@ -4394,7 +4406,7 @@ defmodule AshPostgres.DataLayer do
       |> then(fn query ->
         # A temporal update targets the period valid at `as_of` (the pkey filter
         # alone matches every period of the id), and `update_query` splits it via
-        # FOR PORTION OF. Scope to that period with `valid_at @> as_of`.
+        # splitting it at as_of. Scope to that period with `valid_at @> as_of`.
         if Ash.Resource.Info.temporal?(resource) do
           {:ok, query} =
             set_as_of(
diff --git a/lib/migration_generator/operation.ex b/lib/migration_generator/operation.ex
index 19761331..5f217a2b 100644
--- a/lib/migration_generator/operation.ex
+++ b/lib/migration_generator/operation.ex
@@ -1140,11 +1140,9 @@ defmodule AshPostgres.MigrationGenerator.Operation do
 
         period ->
           # Period-aware uniqueness: the same key values may recur across non-overlapping
-          # periods (history), but never within a single instant. PG19+ enforces this with
-          # a `WITHOUT OVERLAPS`-style GiST exclusion (`key WITH =`, `period WITH &&`,
-          # needs `btree_gist`). Version-guarded so the migration is portable — on servers
-          # older than PG19 (where temporal isn't supported) it degrades to a plain unique
-          # index rather than failing.
+          # periods (history), but never within a single instant. Enforced with a
+          # `WITHOUT OVERLAPS`-style GiST exclusion (`key WITH =`, `period WITH &&`,
+          # needs `btree_gist`).
           predicate =
             cond do
               base_filter && where -> "(#{where}) AND (#{base_filter})"
@@ -1168,13 +1166,7 @@ defmodule AshPostgres.MigrationGenerator.Operation do
           body =
             "ADD CONSTRAINT \\\"#{index_name}\\\" EXCLUDE USING gist (#{exclude_cols})#{where_sql}"
 
-          """
-          if repo().query!("SHOW server_version_num").rows |> hd() |> hd() |> String.to_integer() >= 190_000 do
-            #{alter_table(table, schema, multitenancy, body)}
-          else
-            #{plain}
-          end
-          """
+          alter_table(table, schema, multitenancy, body)
       end
     end
 
@@ -1205,13 +1197,7 @@ defmodule AshPostgres.MigrationGenerator.Operation do
         _period ->
           body = "DROP CONSTRAINT IF EXISTS \\\"#{index_name}\\\""
 
-          """
-          if repo().query!("SHOW server_version_num").rows |> hd() |> hd() |> String.to_integer() >= 190_000 do
-            #{alter_table(table, schema, multitenancy, body)}
-          else
-            #{plain}
-          end
-          """
+          alter_table(table, schema, multitenancy, body)
       end
     end
 
@@ -1435,18 +1421,9 @@ defmodule AshPostgres.MigrationGenerator.Operation do
         # Temporal primary key: the period column is checked for non-overlap
         # rather than equality (SQL:2011 / PG18+, `WITHOUT OVERLAPS`, backed by a
         # GiST exclusion constraint requiring `btree_gist` — enforced via
-        # `installed_extensions`, see VerifyTemporal). Guarded at migration time so
-        # the migration is portable: on servers older than PG19 it degrades to a
-        # plain primary key rather than failing.
-        temporal_keys = plain_keys <> ", #{op.temporal.attribute} WITHOUT OVERLAPS"
-
-        """
-        if repo().query!("SHOW server_version_num").rows |> hd() |> hd() |> String.to_integer() >= 190_000 do
-          #{alter_for.(temporal_keys)}
-        else
-          #{alter_for.(plain_keys)}
-        end
-        """
+        # `installed_extensions`, see VerifyTemporal). Needs PG18+, which VerifyTemporal
+        # requires of the repo's `min_pg_version/0`.
+        alter_for.(plain_keys <> ", #{op.temporal.attribute} WITHOUT OVERLAPS") <> "\n"
       else
         alter_for.(plain_keys) <> "\n"
       end
@@ -1545,12 +1522,10 @@ defmodule AshPostgres.MigrationGenerator.Operation do
           "\\\"#{op.destination_table}\\\""
         end
 
-      # PERIOD foreign keys need PG18+; guarded at migration time so the migration
-      # is portable (skipped entirely on servers older than PG19).
+      # PERIOD foreign keys need PG18+, which VerifyTemporal requires of the repo's
+      # `min_pg_version/0`.
       """
-      if repo().query!("SHOW server_version_num").rows |> hd() |> hd() |> String.to_integer() >= 190_000 do
-        execute("ALTER TABLE \\\"#{op.table}\\\" ADD CONSTRAINT #{name} FOREIGN KEY (#{op.column}, PERIOD #{op.source_period}) REFERENCES #{destination} (#{op.destination_attribute}, PERIOD #{op.destination_period})")
-      end
+      execute("ALTER TABLE \\\"#{op.table}\\\" ADD CONSTRAINT #{name} FOREIGN KEY (#{op.column}, PERIOD #{op.source_period}) REFERENCES #{destination} (#{op.destination_attribute}, PERIOD #{op.destination_period})")
       """
     end
 
diff --git a/lib/temporal.ex b/lib/temporal.ex
index 29987d23..416f9ae7 100644
--- a/lib/temporal.ex
+++ b/lib/temporal.ex
@@ -4,21 +4,60 @@
 
 defmodule AshPostgres.Temporal do
   @moduledoc false
-  # Emits `UPDATE/DELETE ... FOR PORTION OF <valid_at> FROM $1 TO NULL ...`
-  # for temporal resources. Postgres applies the change only to the slice of each
-  # matched row from `as_of` onward, truncating the prior row at `as_of` (its
-  # before-portion is preserved as a new row). A temporal write is always
-  # `[as_of, ∞)`, so `TO` is always `NULL` (the open/unbounded end) and `FROM` is
-  # the changeset's `as_of`. `TO NULL` (not `TO 'infinity'`) is essential: a current
-  # row's stored upper is unbounded (`NULL`), and `'infinity'` is a real value that
-  # an unbounded range *contains* — splitting `TO 'infinity'` would leave a literal
-  # `'infinity'` upper (which Postgrex can't decode) plus a junk `[infinity, )` row.
+  # Splits periods for temporal writes. A write changes only the slice of each matched
+  # version that falls inside its *portion*: `[as_of, ∞)` for an instant, or the period
+  # itself when `as_of` names one. The rest of the version is kept as it was, as one or
+  # two *leftover* rows.
   #
-  # Ecto has no `FOR PORTION OF`, and PG requires the clause BETWEEN the table and
-  # the `AS` alias (`UPDATE t AS a FOR PORTION OF ...` is a syntax error). So we
-  # render the statement with Ecto (`to_sql`, with the bound as `$1` via a counter
-  # offset), splice the clause in after the table, and run it via `repo.query!`.
-  # Mirrors the approach in `AshPostgres.Merge`.
+  # That's SQL:2011's `UPDATE/DELETE ... FOR PORTION OF`, which Postgres doesn't have (it
+  # was reverted from 19 before release, over the same READ COMMITTED lost updates handled
+  # below). Once Postgres ships it natively, we'll use it. Until then we do it by hand, in
+  # one statement, using PG18's `RETURNING old`:
+  #
+  #     WITH "__fpo" AS (
+  #       UPDATE t AS s0 SET ..., valid_at = s0.valid_at * <portion>
+  #       WHERE (...) AND (s0.valid_at && <portion>)
+  #       RETURNING ..., old AS "__old"
+  #     ), "__leftovers" AS (
+  #       INSERT INTO t (<cols>, valid_at)
+  #       SELECT (f."__old").<col>, ..., l.r
+  #       FROM "__fpo" f, unnest(multirange((f."__old").valid_at) - multirange(<portion>)) l(r)
+  #     )
+  #     SELECT ... FROM "__fpo"
+  #
+  # A delete is the same, except it removes the whole version and the leftovers put back
+  # what was outside the portion.
+  #
+  # The portion's upper is `NULL` (unbounded), never `'infinity'`. A current version's
+  # stored upper is unbounded, and `'infinity'` is a real value that an unbounded range
+  # *contains*, so splitting at it would leave a literal `'infinity'` upper (which
+  # Postgrex can't decode) plus a junk `[infinity, )` row.
+  #
+  # Concurrency: under READ COMMITTED, two writes to one record can lose an update. The
+  # second blocks on the version the first is splitting and rechecks it once the first
+  # commits, but by then the version may no longer match (so it's skipped), and the
+  # leftovers the first inserted are newer than the second's snapshot (so they're never
+  # seen). This is what got `FOR PORTION OF` reverted.
+  #
+  # So each statement first locks the versions it's about to split, and compares them
+  # with the versions its snapshot matched (see `gate_ctes/2`). Locking waits out any
+  # concurrent write and then sees the version as that write left it, so if the two
+  # differ, a concurrent write got in the way. The writes are gated on that, so the
+  # statement writes nothing and we just run it again, with a snapshot that sees what the
+  # other write committed. No savepoint needed: a savepoint per write would cost two round
+  # trips and a subtransaction, and enough subtransactions in one transaction slow down
+  # every snapshot on the server.
+  #
+  # When there's nothing to lock (an upsert of a record that doesn't exist yet, or into a
+  # gap), the `WITHOUT OVERLAPS` primary key catches a concurrent insert instead: our insert
+  # waits for it and then conflicts. The insert skips that key (`ON CONFLICT ON CONSTRAINT
+  # <pkey> DO NOTHING`) and we retry just the skipped keys.
+  #
+  # Under REPEATABLE READ and SERIALIZABLE, Postgres raises a serialization failure for
+  # these cases itself.
+
+  # How many times a write runs before giving up on getting past concurrent writes.
+  @attempts 25
 
   @doc """
   Atomic temporal upsert for one or more changesets.
@@ -29,25 +68,29 @@ defmodule AshPostgres.Temporal do
 
       WITH src(<cols>) AS (VALUES (...), (...)),
         upd AS (
-          UPDATE t FOR PORTION OF <valid_at> FROM $as_of TO NULL
-          SET <col> = src.<col>, ... FROM src
+          UPDATE t SET <col> = src.<col>, ..., <valid_at> = t.<valid_at> * [as_of, ∞)
+          FROM src
           WHERE t.<keys> = src.<keys> AND t.<valid_at> @> $as_of   -- period AT as_of
-          RETURNING t.*
+          AND NOT <conflict>                                       -- see `gate_ctes/2`
+          RETURNING new AS "__new", old AS "__old"
         ),
+        "__leftovers" AS (INSERT INTO t ... the part of each matched period before as_of),
         ins AS (
           INSERT INTO t (<cols>, <valid_at>)
           SELECT src.<cols>, <bounded range> FROM src
-          WHERE NOT EXISTS (SELECT 1 FROM upd WHERE upd.<keys> = src.<keys>)
-          RETURNING *
+          WHERE NOT <conflict>
+            AND NOT EXISTS (SELECT 1 FROM upd WHERE (upd."__new").<keys> = src.<keys>)
+          ON CONFLICT ON CONSTRAINT <pkey> DO NOTHING         -- lost a race: retried
+          RETURNING ...
         )
-      SELECT * FROM upd UNION ALL SELECT * FROM ins
+      SELECT (upd."__new").* FROM upd UNION ALL SELECT * FROM ins
 
-  Per changeset: match the period valid AT `as_of` and split it via `FOR PORTION OF`
-  (future periods, which don't contain `as_of`, are untouched); if none, insert a new
-  period bounded to the next one — or unbounded above when there is no later period.
+  Per changeset: match the period valid AT `as_of` and split it there (future periods,
+  which don't contain `as_of`, are untouched); if none, insert a new period bounded to
+  the next one — or unbounded above when there is no later period.
 
-  All changesets sharing an `as_of` run in one statement (the `FOR PORTION OF` bound is a
-  single parameter — it can't be a per-row column). Changesets with differing `as_of`s are
+  All changesets sharing an `as_of` run in one statement (the split point is a single
+  parameter — it can't be a per-row column). Changesets with differing `as_of`s are
   grouped and run as one statement each, inside a transaction. Returns `{:ok, records}`.
   """
   def upsert_all(repo, resource, changesets, upsert_keys, upsert_fields \\ nil) do
@@ -62,30 +105,43 @@ defmodule AshPostgres.Temporal do
         end
       end)
 
-    run = fn ->
-      Enum.flat_map(groups, fn {as_of, group} ->
-        upsert_group(repo, resource, group, upsert_keys, upsert_fields, as_of)
-      end)
-    end
+    prefix = get_in(hd(changesets).context, [:data_layer, :schema])
+    qtable = quote_table(prefix, AshPostgres.DataLayer.Info.table(resource))
+    col_types = column_types(repo, qtable)
 
     records =
-      if map_size(groups) > 1 do
-        {:ok, records} = repo.transaction(run)
-        records
-      else
-        run.()
-      end
+      transaction(repo, fn ->
+        Enum.flat_map(groups, fn {as_of, group} ->
+          upsert_group(
+            repo,
+            resource,
+            qtable,
+            col_types,
+            group,
+            upsert_keys,
+            upsert_fields,
+            as_of
+          )
+        end)
+      end)
 
     {:ok, records}
   end
 
-  defp upsert_group(repo, resource, changesets, upsert_keys, upsert_fields, as_of) do
+  defp upsert_group(
+         repo,
+         resource,
+         qtable,
+         col_types,
+         changesets,
+         upsert_keys,
+         upsert_fields,
+         as_of,
+         attempt \\ 1
+       ) do
     attribute = Ash.Resource.Info.temporal_attribute(resource)
-    prefix = get_in(hd(changesets).context, [:data_layer, :schema])
-    qtable = quote_table(prefix, AshPostgres.DataLayer.Info.table(resource))
     tname = quote_name(AshPostgres.DataLayer.Info.table(resource))
-    qattr = quote_name(attribute)
-    adapter = repo.__adapter__()
+    qattr = quote_name(source_col(resource, attribute))
 
     # The columns to write — every changeset attribute in the group except the temporal
     # range, which we build from `as_of`.
@@ -107,68 +163,43 @@ defmodule AshPostgres.Temporal do
       |> Kernel.--(upsert_keys)
       |> Kernel.--([attribute])
 
-    col_types = column_types(repo, qtable)
     qsrc = fn field -> quote_name(source_col(resource, field)) end
-    col_count = length(insert_cols)
 
-    # `VALUES` rows + their params. `$1` is `as_of`; row i, column j is `$(2 + i*c + j)`.
-    # The first row carries `::type` casts so Postgres knows each `src` column's type.
-    {rows_sql, row_params} =
-      changesets
-      |> Enum.with_index()
-      |> Enum.map(fn {changeset, ri} ->
-        insert_cols
-        |> Enum.with_index()
-        |> Enum.map(fn {field, ci} ->
-          n = 2 + ri * col_count + ci
+    # `$1` is `as_of`, the `VALUES` params follow it.
+    {src_clause, row_params} =
+      values_source(repo, resource, col_types, changesets, insert_cols, 2)
 
-          {:ok, dumped} =
-            Ecto.Type.adapter_dump(
-              adapter,
-              resource.__schema__(:type, field),
-              Map.get(changeset.attributes, field)
-            )
-
-          placeholder =
-            if ri == 0 do
-              "$#{n}::#{Map.fetch!(col_types, to_string(source_col(resource, field)))}"
-            else
-              "$#{n}"
-            end
-
-          {placeholder, dumped}
-        end)
-        |> Enum.unzip()
-        |> then(fn {phs, vals} -> {"(" <> Enum.join(phs, ", ") <> ")", vals} end)
-      end)
-      |> Enum.unzip()
-
-    params = [as_of | List.flatten(row_params)]
-
-    src_clause =
-      "src(#{Enum.map_join(insert_cols, ", ", qsrc)}) AS (VALUES #{Enum.join(rows_sql, ", ")})"
+    params = [as_of | row_params]
 
     key_match = fn left ->
       Enum.map_join(upsert_keys, " AND ", fn k -> "#{left}.#{qsrc.(k)} = src.#{qsrc.(k)}" end)
     end
 
     has_update? = set_fields != []
+    range = "tstzrange($1::timestamptz, NULL)"
+    pkey = quote_name("#{AshPostgres.DataLayer.Info.table(resource)}_pkey")
+    columns = Enum.map(stored_columns(resource), &quote_name/1)
+    upsert_cols = Enum.map(upsert_keys, qsrc)
 
-    upd_cte =
+    {gate_ctes, attempted} =
       if has_update? do
         set_sql = Enum.map_join(set_fields, ", ", fn f -> "#{qsrc.(f)} = src.#{qsrc.(f)}" end)
 
-        "upd AS (UPDATE #{qtable} FOR PORTION OF #{qattr} FROM $1::timestamptz TO NULL " <>
-          "SET #{set_sql} FROM src " <>
-          "WHERE #{key_match.(tname)} AND #{tname}.#{qattr} @> $1::timestamptz RETURNING #{tname}.*)"
-      end
-
-    not_exists =
-      if has_update? do
-        "NOT EXISTS (SELECT 1 FROM upd WHERE #{key_match.("upd")})"
+        matched =
+          "FROM #{qtable} t JOIN src ON #{key_match.("t")} AND t.#{qattr} @> $1::timestamptz"
+
+        {gate_ctes(matched, "t") <>
+           "upd AS (UPDATE #{qtable} SET #{set_sql}, #{qattr} = #{tname}.#{qattr} * #{range} " <>
+           "FROM src " <>
+           "WHERE #{key_match.(tname)} AND #{tname}.#{qattr} @> $1::timestamptz " <>
+           ~s|AND NOT (SELECT conflict FROM "__conflict") | <>
+           ~s|RETURNING new AS "__new", old AS "__old"), | <>
+           leftovers_cte(resource, qtable, "upd", range) <> ", ",
+         ~s|NOT EXISTS (SELECT 1 FROM upd WHERE #{key_match.(~s|(upd."__new")|)})|}
       else
-        "NOT EXISTS (SELECT 1 FROM #{qtable} ex WHERE #{key_match.("ex")} " <>
-          "AND ex.#{qattr} @> $1::timestamptz)"
+        {~s|"__conflict" AS (SELECT false AS conflict), |,
+         "NOT EXISTS (SELECT 1 FROM #{qtable} ex WHERE #{key_match.("ex")} " <>
+           "AND ex.#{qattr} @> $1::timestamptz)"}
       end
 
     # Bound the inserted range to this key's next period (gap-fill), else unbounded above
@@ -182,24 +213,102 @@ defmodule AshPostgres.Temporal do
     insert_select_sql =
       Enum.map_join(insert_cols, ", ", fn f -> "src.#{qsrc.(f)}" end) <> ", " <> range_sql
 
+    # A key that loses an insert race to a concurrent write of the same record hits the
+    # `WITHOUT OVERLAPS` primary key. It's skipped rather than raised, and retried below.
+    # Only the primary key is an arbiter, so an identity violation still raises.
     ins_cte =
-      "ins AS (INSERT INTO #{qtable} (#{insert_cols_sql}) " <>
-        "SELECT #{insert_select_sql} FROM src WHERE #{not_exists} RETURNING *)"
+      ~s|"__attempted" AS (SELECT * FROM src WHERE NOT (SELECT conflict FROM "__conflict") | <>
+        "AND #{attempted}), " <>
+        "ins AS (INSERT INTO #{qtable} (#{insert_cols_sql}) " <>
+        ~s|SELECT #{insert_select_sql} FROM "__attempted" src | <>
+        "ON CONFLICT ON CONSTRAINT #{pkey} DO NOTHING " <>
+        "RETURNING #{Enum.join(columns, ", ")})"
+
+    written =
+      if has_update? do
+        ~s|SELECT #{Enum.map_join(columns, ", ", &~s|(upd."__new").#{&1}|)}, 'ok' AS "__status" | <>
+          "FROM upd UNION ALL SELECT #{Enum.join(columns, ", ")}, 'ok' FROM ins"
+      else
+        ~s|SELECT #{Enum.join(columns, ", ")}, 'ok' AS "__status" FROM ins|
+      end
+
+    skipped_columns =
+      Enum.map_join(columns, ", ", fn column ->
+        if column in upsert_cols, do: "src.#{column}", else: "NULL"
+      end)
 
     sql =
-      if has_update? do
-        "WITH #{src_clause}, #{upd_cte}, #{ins_cte} SELECT * FROM upd UNION ALL SELECT * FROM ins"
+      "WITH #{src_clause}, #{gate_ctes}#{ins_cte} " <>
+        written <>
+        ~s| UNION ALL SELECT #{skipped_columns}, 'skipped' FROM "__attempted" src | <>
+        "WHERE NOT EXISTS (SELECT 1 FROM ins WHERE #{key_match.("ins")}) " <>
+        "UNION ALL SELECT #{Enum.map_join(columns, ", ", fn _ -> "NULL" end)}, 'conflict' " <>
+        ~s|FROM "__conflict" WHERE conflict|
+
+    {columns, %{"ok" => written, "skipped" => skipped, "conflict" => conflict}} =
+      by_status(repo.query!(sql, params))
+
+    retry = fn retried ->
+      if attempt < @attempts do
+        upsert_group(
+          repo,
+          resource,
+          qtable,
+          col_types,
+          retried,
+          upsert_keys,
+          upsert_fields,
+          as_of,
+          attempt + 1
+        )
       else
-        "WITH #{src_clause}, #{ins_cte} SELECT * FROM ins"
+        raise AshPostgres.Temporal.WriteConflict, resource: resource, attempts: @attempts
       end
+    end
+
+    cond do
+      # Nothing was written, so run the whole group again.
+      conflict != [] ->
+        retry.(changesets)
+
+      # Every other key was written, so run just the skipped ones again.
+      skipped != [] ->
+        skipped_keys =
+          repo
+          |> load_rows(resource, columns, skipped)
+          |> MapSet.new(fn record -> Enum.map(upsert_keys, &Map.get(record, &1)) end)
+
+        retried =
+          Enum.filter(changesets, fn changeset ->
+            Enum.map(upsert_keys, &Map.get(changeset.attributes, &1)) in skipped_keys
+          end)
+
+        records(repo, resource, columns, written, changesets, upsert_keys) ++
+          retry.(retried)
 
-    result = repo.query!(sql, params)
+      true ->
+        records(repo, resource, columns, written, changesets, upsert_keys)
+    end
+  end
 
+  defp records(repo, resource, columns, rows, changesets, upsert_keys) do
     repo
-    |> load_rows(resource, result.columns, result.rows || [])
+    |> load_rows(resource, columns, rows)
     |> tag_bulk_refs(changesets, upsert_keys)
   end
 
+  # Splits a result's rows by their `"__status"` column, which is dropped.
+  defp by_status(%{columns: columns, rows: rows}) do
+    index = Enum.find_index(columns, &(&1 == "__status"))
+
+    groups =
+      rows
+      |> Enum.group_by(&Enum.at(&1, index), &List.delete_at(&1, index))
+      |> then(&Map.merge(%{"ok" => [], "skipped" => [], "conflict" => []}, &1))
+
+    {List.delete_at(columns, index), groups}
+  end
+
   # Ash's bulk action correlates each returned record back to its changeset via
   # `__metadata__.bulk_action_ref`. Our `UNION` returns rows out of input order, so we
   # re-attach the ref by matching each record's upsert-key values to its changeset.
@@ -227,6 +336,47 @@ defmodule AshPostgres.Temporal do
     end
   end
 
+  # `src(<cols>) AS (VALUES ...)` over the changesets, with its params numbered from
+  # `first`. The first row carries `::type` casts so Postgres knows each column's type.
+  defp values_source(repo, resource, col_types, changesets, cols, first) do
+    adapter = repo.__adapter__()
+    col_count = length(cols)
+
+    {rows_sql, row_params} =
+      changesets
+      |> Enum.with_index()
+      |> Enum.map(fn {changeset, ri} ->
+        cols
+        |> Enum.with_index()
+        |> Enum.map(fn {field, ci} ->
+          n = first + ri * col_count + ci
+
+          {:ok, dumped} =
+            Ecto.Type.adapter_dump(
+              adapter,
+              resource.__schema__(:type, field),
+              Map.get(changeset.attributes, field)
+            )
+
+          placeholder =
+            if ri == 0 do
+              "$#{n}::#{Map.fetch!(col_types, to_string(source_col(resource, field)))}"
+            else
+              "$#{n}"
+            end
+
+          {placeholder, dumped}
+        end)
+        |> Enum.unzip()
+        |> then(fn {phs, vals} -> {"(" <> Enum.join(phs, ", ") <> ")", vals} end)
+      end)
+      |> Enum.unzip()
+
+    columns = Enum.map_join(cols, ", ", &quote_name(source_col(resource, &1)))
+
+    {"src(#{columns}) AS (VALUES #{Enum.join(rows_sql, ", ")})", List.flatten(row_params)}
+  end
+
   # The Postgres type of each column (e.g. `"integer"`, `"tstzrange"`) so `VALUES` params
   # can be cast — otherwise they default to `text` and join/compare against typed columns.
   defp column_types(repo, qtable) do
@@ -244,91 +394,219 @@ defmodule AshPostgres.Temporal do
   defp portion(%Ash.Range{} = period), do: period
   defp portion(instant), do: %Ash.Range{lower: instant, upper: nil, bounds: :"[)"}
 
-  @doc "Run an UPDATE ... FOR PORTION OF. Returns `{count, loaded_records | nil}`."
+  @doc "Run a temporal UPDATE over `as_of`'s portion. Returns `{count, loaded_records | nil}`."
   def update_all(repo, query, resource, as_of) do
     run(repo, :update_all, query, resource, as_of)
   end
 
-  @doc "Run a DELETE ... FOR PORTION OF. Returns `{count, loaded_records | nil}`."
+  @doc "Run a temporal DELETE over `as_of`'s portion. Returns `{count, loaded_records | nil}`."
   def delete_all(repo, query, resource, as_of) do
     run(repo, :delete_all, query, resource, as_of)
   end
 
   defp run(repo, kind, query, resource, as_of) do
-    attribute = Ash.Resource.Info.temporal_attribute(resource)
+    query = Map.delete(query, :__ash_bindings__)
     returning? = not is_nil(query.select)
     %Ash.Range{lower: lower, upper: upper} = portion(as_of)
 
     # Render with the portion's bounds reserved as $1 (and $2 when it has an upper).
     counter = if upper, do: 2, else: 1
+    {sql, params} = Ecto.Adapters.SQL.to_sql(kind, repo, query, counter: counter)
 
-    {sql, params} =
-      Ecto.Adapters.SQL.to_sql(kind, repo, Map.delete(query, :__ash_bindings__), counter: counter)
-
-    sql = splice_for_portion_of(sql, kind, attribute, not is_nil(upper))
-    sql = if returning?, do: first_versions(sql, resource), else: sql
+    range = "tstzrange($1::timestamptz, #{if upper, do: "$2::timestamptz", else: "NULL"})"
+    sql = split_portion(sql, kind, resource, range, returning?)
     bounds = if upper, do: [lower, upper], else: [lower]
-    result = repo.query!(sql, bounds ++ params)
+    result = write(repo, resource, sql, bounds ++ params)
 
     if returning? do
-      rows = result.rows || []
-      {length(rows), load_rows(repo, resource, result.columns, rows)}
+      {length(result.rows), load_rows(repo, resource, result.columns, result.rows)}
     else
-      {result.num_rows, nil}
+      {result.rows |> hd() |> hd(), nil}
     end
   end
 
-  # A range can carve several versions of one record; the write returns the first of them.
-  defp first_versions(sql, resource) do
-    keys =
-      resource
-      |> Ash.Resource.Info.primary_key()
-      |> Enum.map_join(", ", &quote_name(source_col(resource, &1)))
+  # Rewrites Ecto's `UPDATE`/`DELETE` into the split described in the moduledoc.
+  defp split_portion(sql, kind, resource, range, returning?) do
+    [_, qtable, alias_token] =
+      Regex.run(~r/^(?:UPDATE|DELETE FROM) ((?:"[^"]+"\.)?"[^"]+") AS (\w+) /, sql)
 
-    period = quote_name(source_col(resource, Ash.Resource.Info.temporal_attribute(resource)))
+    period_col = quote_name(source_col(resource, temporal_attribute(resource)))
+    period = alias_token <> "." <> period_col
 
-    "WITH first_versions AS (#{sql}) SELECT DISTINCT ON (#{keys}) * FROM first_versions " <>
-      "ORDER BY #{keys}, lower(#{period})"
-  end
+    {dml, returned} =
+      case find_top_level(sql, " RETURNING ") do
+        nil -> {sql, nil}
+        pos -> {String.slice(sql, 0, pos), String.slice(sql, (pos + 11)..-1//1)}
+      end
 
-  # Insert ` FOR PORTION OF "<attr>" FROM $1::timestamptz TO …` immediately after the
-  # table name and before the ` AS <alias>` that Ecto emits. The upper is `$2` when
-  # `as_of` names a period, else `NULL` — see the moduledoc for why not `'infinity'`.
-  defp splice_for_portion_of(sql, kind, attribute, upper?) do
-    {head, rest} = split_on_set_or_where(sql, kind)
-    alias_token = head |> String.trim() |> String.split() |> List.last()
-    table_part = String.replace_suffix(head, " AS #{alias_token}", "")
+    dml =
+      case kind do
+        :update_all ->
+          set_end =
+            [find_top_level(dml, " FROM "), find_top_level(dml, " WHERE ")]
+            |> Enum.reject(&is_nil/1)
+            |> Enum.min(fn -> String.length(dml) end)
 
-    to = if upper?, do: "$2::timestamptz", else: "NULL"
+          String.slice(dml, 0, set_end) <>
+            ", #{period_col} = #{period} * #{range}" <> String.slice(dml, set_end..-1//1)
 
-    clause =
-      " FOR PORTION OF " <>
-        quote_name(attribute) <> " FROM $1::timestamptz TO " <> to
+        :delete_all ->
+          dml
+      end
+
+    overlaps = "(#{period} && #{range})"
+
+    dml =
+      case find_top_level(dml, " WHERE ") do
+        nil ->
+          dml <> " WHERE " <> overlaps
+
+        pos ->
+          String.slice(dml, 0, pos) <>
+            " WHERE (" <> String.slice(dml, (pos + 7)..-1//1) <> ") AND " <> overlaps
+      end
+
+    # The versions the snapshot matched: the same table, joins and filter, as a SELECT.
+    {from_keyword, from_pos} =
+      case kind do
+        :update_all -> {" FROM ", find_top_level(dml, " FROM ")}
+        :delete_all -> {" USING ", find_top_level(dml, " USING ")}
+      end
+
+    where_pos = find_top_level(dml, " WHERE ")
+
+    joined =
+      if from_pos && from_pos < where_pos do
+        ", " <> String.slice(dml, (from_pos + String.length(from_keyword))..(where_pos - 1)//1)
+      else
+        ""
+      end
+
+    matched = "FROM #{qtable} AS #{alias_token}#{joined}" <> String.slice(dml, where_pos..-1//1)
+
+    returning = Enum.join(List.wrap(returned) ++ [~s|old AS "__old"|], ", ")
+
+    select =
+      if returning? do
+        # A range can carve several versions of one record; the write returns the first.
+        columns = returned_columns(returned, alias_token)
+
+        keys =
+          resource
+          |> Ash.Resource.Info.primary_key()
+          |> Enum.map_join(", ", &"f.#{quote_name(source_col(resource, &1))}")
+
+        ~s|(SELECT DISTINCT ON (#{keys}) #{Enum.map_join(columns, ", ", &"f.#{&1}")}, | <>
+          ~s|'ok' AS "__status" FROM "__fpo" f ORDER BY #{keys}, lower(f.#{period_col})) | <>
+          "UNION ALL SELECT #{Enum.map_join(columns, ", ", &"n.#{&1}")}, 'conflict' " <>
+          ~s|FROM (SELECT (NULL::#{qtable}).*) n, "__conflict" WHERE conflict|
+      else
+        ~s|SELECT count(*), | <>
+          ~s|(SELECT CASE WHEN conflict THEN 'conflict' ELSE 'ok' END FROM "__conflict") | <>
+          ~s|AS "__status" FROM "__fpo"|
+      end
 
-    table_part <> clause <> " AS " <> alias_token <> rest
+    "WITH " <>
+      gate_ctes(matched, alias_token) <>
+      ~s|"__fpo" AS (#{dml} AND NOT (SELECT conflict FROM "__conflict") RETURNING #{returning}), | <>
+      leftovers_cte(resource, qtable, ~s|"__fpo"|, range) <> " " <> select
   end
 
-  # `head` is everything up to (and including the keyword we split on); `rest`
-  # is the keyword + remainder. We keep the keyword on `rest`.
-  defp split_on_set_or_where(sql, :update_all), do: split_keeping(sql, " SET ")
+  # `"__cand"` is the versions `matched` (a `FROM ... WHERE ...`) finds in the snapshot, and
+  # `"__locked"` is the same versions, locked. Locking waits out any concurrent write to a
+  # version and then sees it as that write left it: changed (a newer row) or gone. So the
+  # two differ exactly when a concurrent write got in the way, and `"__conflict"` says so.
+  # The writes are gated on it, so a conflicting statement writes nothing.
+  defp gate_ctes(matched, alias_token) do
+    ~s|"__cand" AS (SELECT #{alias_token}.ctid #{matched}), | <>
+      ~s|"__locked" AS (SELECT #{alias_token}.ctid #{matched} FOR UPDATE OF #{alias_token}), | <>
+      ~s|"__conflict" AS (SELECT (| <>
+      ~s|EXISTS (SELECT 1 FROM "__cand" c WHERE c.ctid NOT IN (SELECT ctid FROM "__locked")) OR | <>
+      ~s|EXISTS (SELECT 1 FROM "__locked" l WHERE l.ctid NOT IN (SELECT ctid FROM "__cand"))| <>
+      ~s|) AS conflict), |
+  end
+
+  # Runs a split, and runs it again while it reports a conflict (see the moduledoc). A
+  # conflicting statement wrote nothing, so there's nothing to undo first.
+  defp write(repo, resource, sql, params, attempt \\ 1) do
+    case by_status(repo.query!(sql, params)) do
+      # Retried in a transaction, so the versions this attempt locked stay locked and can't
+      # move again under the next one.
+      {_, %{"conflict" => [_ | _]}} when attempt < @attempts ->
+        transaction(repo, fn -> write(repo, resource, sql, params, attempt + 1) end)
+
+      {_, %{"conflict" => [_ | _]}} ->
+        raise AshPostgres.Temporal.WriteConflict, resource: resource, attempts: @attempts
 
-  defp split_on_set_or_where(sql, :delete_all) do
-    case find_top_level(sql, " WHERE ") do
-      nil -> split_keeping(sql, " RETURNING ")
-      _ -> split_keeping(sql, " WHERE ")
+      {columns, %{"ok" => rows}} ->
+        %{columns: columns, rows: rows}
     end
   end
 
-  defp split_keeping(sql, keyword) do
-    case find_top_level(sql, keyword) do
+  # Ecto returns plain columns (`s0."id", s0."name"`); select them by name.
+  defp returned_columns(returned, alias_token) do
+    returned
+    |> split_top_level(", ")
+    |> Enum.map(fn item ->
+      case Regex.run(~r/^#{alias_token}\.("[^"]+")$/, item) do
+        [_, column] ->
+          column
+
+        nil ->
+          raise ArgumentError,
+                "Expected a plain column in temporal RETURNING, got: #{inspect(item)}"
+      end
+    end)
+  end
+
+  defp split_top_level(sql, separator) do
+    case find_top_level(sql, separator) do
       nil ->
-        raise ArgumentError, "Expected #{keyword} in rendered temporal SQL: #{inspect(sql)}"
+        [sql]
 
       pos ->
-        {String.slice(sql, 0, pos), String.slice(sql, pos..-1//1)}
+        [
+          String.slice(sql, 0, pos)
+          | split_top_level(String.slice(sql, (pos + String.length(separator))..-1//1), separator)
+        ]
+    end
+  end
+
+  # Puts back what each version in `from` had outside the portion: every column copied
+  # from `"__old"`, over `old - portion` (zero, one or two ranges).
+  defp leftovers_cte(resource, qtable, from, range) do
+    attribute = temporal_attribute(resource)
+    period = quote_name(source_col(resource, attribute))
+
+    columns =
+      resource
+      |> stored_columns()
+      |> Enum.reject(&(&1 == source_col(resource, attribute)))
+      |> Enum.map(&quote_name/1)
+
+    ~s|"__leftovers" AS (INSERT INTO #{qtable} (#{Enum.join(columns ++ [period], ", ")}) | <>
+      "OVERRIDING SYSTEM VALUE " <>
+      ~s|SELECT #{Enum.map_join(columns, ", ", &~s|(f."__old").#{&1}|)}, l.r | <>
+      ~s|FROM #{from} f, unnest(multirange((f."__old").#{period}) - multirange(#{range})) AS l(r))|
+  end
+
+  defp stored_columns(resource) do
+    resource.__schema__(:fields)
+    |> Enum.map(&resource.__schema__(:field_source, &1))
+  end
+
+  # A partly retried upsert must commit or roll back as a whole.
+  defp transaction(repo, fun) do
+    if repo.in_transaction?() do
+      fun.()
+    else
+      {:ok, result} = repo.transaction(fun)
+      result
     end
   end
 
+  defp temporal_attribute(resource), do: Ash.Resource.Info.temporal_attribute(resource)
+
   defp load_rows(repo, resource, columns, rows) do
     source = AshPostgres.DataLayer.Info.table(resource)
     keys = Enum.map(columns, &String.to_existing_atom/1)
diff --git a/lib/temporal/write_conflict.ex b/lib/temporal/write_conflict.ex
new file mode 100644
index 00000000..d0ef431c
--- /dev/null
+++ b/lib/temporal/write_conflict.ex
@@ -0,0 +1,22 @@
+# SPDX-FileCopyrightText: 2024 ash_postgres contributors <https://github.com/ash-project/ash_postgres/graphs/contributors>
+#
+# SPDX-License-Identifier: MIT
+
+defmodule AshPostgres.Temporal.WriteConflict do
+  @moduledoc """
+  Used when a temporal write keeps losing to concurrent writes to the same records.
+
+  A temporal write that finds the versions it's splitting were changed by another
+  transaction in the meantime runs again. This is raised when it has run `attempts` times
+  and still conflicted, which takes many transactions writing the same slice of the same
+  records at once. Retrying the action is safe: nothing was written.
+  """
+
+  use Splode.Error, fields: [:resource, :attempts], class: :framework
+
+  def message(error) do
+    "Temporal write to #{inspect(error.resource)} conflicted with concurrent writes to the " <>
+      "same records #{error.attempts} times in a row, so nothing was written. " <>
+      "Retrying the action is safe."
+  end
+end
diff --git a/lib/transformers/validate_temporal_references.ex b/lib/transformers/validate_temporal_references.ex
index dda1ee96..7f3331ec 100644
--- a/lib/transformers/validate_temporal_references.ex
+++ b/lib/transformers/validate_temporal_references.ex
@@ -6,9 +6,9 @@ defmodule AshPostgres.Transformers.ValidateTemporalReferences do
   @moduledoc false
   # A temporal relationship (`temporal_keys` with both a source and destination period
   # attribute) is backed by a Postgres `PERIOD` foreign key. PostgreSQL only supports
-  # `NO ACTION` for those — "PostgreSQL supports temporal foreign keys with action
-  # NO ACTION, but not RESTRICT, CASCADE, SET NULL, or SET DEFAULT" (PG19 docs, 5.7
-  # Temporal Tables). So reject an `on_delete`/`on_update` referential action on such a
+  # `NO ACTION` for those, not RESTRICT, CASCADE, SET NULL, or SET DEFAULT ("unsupported
+  # ON DELETE action for foreign key constraint using PERIOD"). So reject an
+  # `on_delete`/`on_update` referential action on such a
   # relationship's `references` — it can't be expressed at the database level and must be
   # handled in the application (e.g. an `Ash.Resource.Change.CascadeDestroy`).
   #
diff --git a/lib/verifiers/verify_temporal.ex b/lib/verifiers/verify_temporal.ex
index e7811bda..a719e312 100644
--- a/lib/verifiers/verify_temporal.ex
+++ b/lib/verifiers/verify_temporal.ex
@@ -4,41 +4,76 @@
 
 defmodule AshPostgres.Verifiers.VerifyTemporal do
   @moduledoc false
-  # A temporal resource's `WITHOUT OVERLAPS` primary key is backed by a GiST
-  # exclusion constraint, which requires the `btree_gist` extension. Ensure the
-  # repo declares it in `installed_extensions/0` so migrations install it.
+  # A temporal resource needs:
+  #
+  # - PostgreSQL 18+, for `WITHOUT OVERLAPS` primary keys, `PERIOD` foreign keys and
+  #   `RETURNING old`. Migrations emit that DDL unconditionally, so the repo has to
+  #   declare it with `min_pg_version/0`.
+  # - The `btree_gist` extension, since the `WITHOUT OVERLAPS` primary key is backed by
+  #   a GiST exclusion constraint. The repo has to declare it in `installed_extensions/0`
+  #   so migrations install it.
   use Spark.Dsl.Verifier
   alias Spark.Dsl.Verifier
 
   def verify(dsl) do
-    if is_nil(Ash.Resource.Info.temporal_strategy(dsl)) do
+    repo = AshPostgres.DataLayer.Info.repo(dsl, :mutate)
+
+    if is_nil(Ash.Resource.Info.temporal_strategy(dsl)) or is_nil(repo) or
+         not Code.ensure_loaded?(repo) do
       :ok
     else
-      repo = AshPostgres.DataLayer.Info.repo(dsl, :mutate)
-
-      with true <- not is_nil(repo) and Code.ensure_loaded?(repo),
-           true <- function_exported?(repo, :installed_extensions, 0),
-           false <- "btree_gist" in repo.installed_extensions() do
-        resource = Verifier.get_persisted(dsl, :module)
-
-        raise Spark.Error.DslError,
-          module: resource,
-          message: """
-          Temporal resource #{inspect(resource)} requires the `btree_gist` extension.
-
-          Its `WITHOUT OVERLAPS` primary key is backed by a GiST exclusion constraint,
-          which needs `btree_gist`. Add it to your repo's `installed_extensions/0`:
-
-          ```elixir
-          def installed_extensions do
-            ["btree_gist"]
-          end
-          ```
-          """,
-          path: [:temporal]
-      else
-        _ -> :ok
+      resource = Verifier.get_persisted(dsl, :module)
+
+      with :ok <- verify_min_pg_version(resource, repo) do
+        verify_btree_gist(resource, repo)
       end
     end
   end
+
+  defp verify_min_pg_version(resource, repo) do
+    with true <- function_exported?(repo, :min_pg_version, 0),
+         %Version{major: major} = version when major < 18 <- repo.min_pg_version() do
+      raise Spark.Error.DslError,
+        module: resource,
+        message: """
+        Temporal resource #{inspect(resource)} requires PostgreSQL 18 or later.
+
+        `#{inspect(repo)}.min_pg_version/0` is #{version}. Temporal resources rely on
+        `WITHOUT OVERLAPS` primary keys, `PERIOD` foreign keys and `RETURNING old`,
+        which arrived in PostgreSQL 18. If your database is on 18 or later, say so:
+
+        ```elixir
+        def min_pg_version do
+          %Version{major: 18, minor: 0, patch: 0}
+        end
+        ```
+        """,
+        path: [:temporal]
+    else
+      _ -> :ok
+    end
+  end
+
+  defp verify_btree_gist(resource, repo) do
+    with true <- function_exported?(repo, :installed_extensions, 0),
+         false <- "btree_gist" in repo.installed_extensions() do
+      raise Spark.Error.DslError,
+        module: resource,
+        message: """
+        Temporal resource #{inspect(resource)} requires the `btree_gist` extension.
+
+        Its `WITHOUT OVERLAPS` primary key is backed by a GiST exclusion constraint,
+        which needs `btree_gist`. Add it to your repo's `installed_extensions/0`:
+
+        ```elixir
+        def installed_extensions do
+          ["btree_gist"]
+        end
+        ```
+        """,
+        path: [:temporal]
+    else
+      _ -> :ok
+    end
+  end
 end
diff --git a/mix.exs b/mix.exs
index d792f91b..cbb8342f 100644
--- a/mix.exs
+++ b/mix.exs
@@ -120,6 +120,7 @@ defmodule AshPostgres.MixProject do
         "documentation/topics/advanced/schema-based-multitenancy.md",
         "documentation/topics/advanced/using-multiple-repos.md",
         "documentation/topics/advanced/upserts.md",
+        "documentation/topics/advanced/temporal-resources.md",
         {"documentation/dsls/DSL-AshPostgres.DataLayer.md",
          search_data: Spark.Docs.search_data_for(AshPostgres.DataLayer)},
         "CHANGELOG.md"
@@ -189,8 +190,7 @@ defmodule AshPostgres.MixProject do
     [
       {:ash, ash_version("~> 3.34")},
       {:spark, "~> 2.3 and >= 2.3.4"},
-      {:ash_sql,
-       ash_sql_version("~> 0.8")},
+      {:ash_sql, ash_sql_version("~> 0.8")},
       {:igniter, "~> 0.6 and >= 0.6.29", optional: true},
       {:ecto_sql, "~> 3.13"},
       {:ecto, "~> 3.13"},
diff --git a/test/support/repo_case.ex b/test/support/repo_case.ex
index 8a971ea5..f7ced59c 100644
--- a/test/support/repo_case.ex
+++ b/test/support/repo_case.ex
@@ -21,10 +21,17 @@ defmodule AshPostgres.RepoCase do
   end
 
   setup tags do
-    :ok = Sandbox.checkout(AshPostgres.TestRepo)
+    # `AshPostgres.TemporalTestRepo` is only started on PostgreSQL 18+.
+    repos =
+      [AshPostgres.TestRepo, AshPostgres.TemporalTestRepo]
+      |> Enum.filter(&Process.whereis/1)
 
-    if !tags[:async] do
-      Sandbox.mode(AshPostgres.TestRepo, {:shared, self()})
+    for repo <- repos do
+      :ok = Sandbox.checkout(repo)
+
+      if !tags[:async] do
+        Sandbox.mode(repo, {:shared, self()})
+      end
     end
 
     :ok
diff --git a/test/support/temporal/domain.ex b/test/support/temporal/domain.ex
index 23b30c02..81189fc3 100644
--- a/test/support/temporal/domain.ex
+++ b/test/support/temporal/domain.ex
@@ -10,6 +10,7 @@ defmodule AshPostgres.Test.Temporal.Domain do
     resource AshPostgres.Test.Temporal.Subscription do
       define(:add_seat_if_basic, action: :add_seat_if_basic, get_by: [:id])
     end
+
     resource(AshPostgres.Test.Temporal.Tier)
     resource(AshPostgres.Test.Temporal.Event)
     resource(AshPostgres.Test.Temporal.Member)
diff --git a/test/support/temporal/event.ex b/test/support/temporal/event.ex
index a76a7e96..2285cc7c 100644
--- a/test/support/temporal/event.ex
+++ b/test/support/temporal/event.ex
@@ -10,7 +10,7 @@ defmodule AshPostgres.Test.Temporal.Event do
 
   postgres do
     table("event")
-    repo(AshPostgres.TestRepo)
+    repo(AshPostgres.TemporalTestRepo)
   end
 
   attributes do
diff --git a/test/support/temporal/member.ex b/test/support/temporal/member.ex
index c7f0afc0..b0e29c90 100644
--- a/test/support/temporal/member.ex
+++ b/test/support/temporal/member.ex
@@ -10,7 +10,7 @@ defmodule AshPostgres.Test.Temporal.Member do
 
   postgres do
     table("member")
-    repo(AshPostgres.TestRepo)
+    repo(AshPostgres.TemporalTestRepo)
   end
 
   temporal do
diff --git a/test/support/temporal/subscription.ex b/test/support/temporal/subscription.ex
index 17d17446..67f55911 100644
--- a/test/support/temporal/subscription.ex
+++ b/test/support/temporal/subscription.ex
@@ -10,7 +10,7 @@ defmodule AshPostgres.Test.Temporal.Subscription do
 
   postgres do
     table("subscription")
-    repo(AshPostgres.TestRepo)
+    repo(AshPostgres.TemporalTestRepo)
   end
 
   temporal do
@@ -87,7 +87,7 @@ defmodule AshPostgres.Test.Temporal.Subscription do
       accept([:tier])
     end
 
-    # Atomic arithmetic update — exercised over FOR PORTION OF.
+    # Atomic arithmetic update — exercised over a period split.
     update :add_seat do
       require_atomic?(true)
       change(atomic_update(:seats, expr(seats + 1)))
diff --git a/test/support/temporal/tier.ex b/test/support/temporal/tier.ex
index 4695103f..beebc7cf 100644
--- a/test/support/temporal/tier.ex
+++ b/test/support/temporal/tier.ex
@@ -10,7 +10,7 @@ defmodule AshPostgres.Test.Temporal.Tier do
 
   postgres do
     table("tier")
-    repo(AshPostgres.TestRepo)
+    repo(AshPostgres.TemporalTestRepo)
   end
 
   temporal do
diff --git a/test/support/temporal_test_repo.ex b/test/support/temporal_test_repo.ex
new file mode 100644
index 00000000..fc1c003a
--- /dev/null
+++ b/test/support/temporal_test_repo.ex
@@ -0,0 +1,31 @@
+# SPDX-FileCopyrightText: 2019 ash_postgres contributors <https://github.com/ash-project/ash_postgres/graphs/contributors>
+#
+# SPDX-License-Identifier: MIT
+
+defmodule AshPostgres.TemporalTestRepo do
+  @moduledoc false
+  use AshPostgres.Repo,
+    otp_app: :ash_postgres
+
+  # Temporal resources need PostgreSQL 18+ (see `AshPostgres.Verifiers.VerifyTemporal`),
+  # while the rest of the suite also runs on older servers. So they get a repo of their
+  # own, in the same database, that's only migrated and started when testing against 18+
+  # (see `config/config.exs` and `test/test_helper.exs`). On older servers the resources
+  # still compile, and their tests are excluded by the `:postgres_18` tag.
+  #
+  # See `AshPostgres.TestPaths` for why the migrations live outside `priv/`.
+  def init(type, config) do
+    {:ok, config} = super(type, config)
+    {:ok, AshPostgres.TestPaths.put_paths(config, "temporal_test_repo")}
+  end
+
+  def min_pg_version, do: %Version{major: 18, minor: 0, patch: 0}
+
+  def installed_extensions, do: ["ash-functions", "citext", "btree_gist"]
+
+  def prefer_transaction?, do: false
+
+  def prefer_transaction_for_atomic_updates?, do: false
+
+  def all_tenants, do: []
+end
diff --git a/test/temporal_concurrency_test.exs b/test/temporal_concurrency_test.exs
new file mode 100644
index 00000000..d5b8b82f
--- /dev/null
+++ b/test/temporal_concurrency_test.exs
@@ -0,0 +1,185 @@
+# SPDX-FileCopyrightText: 2019 ash_postgres contributors <https://github.com/ash-project/ash_postgres/graphs/contributors>
+#
+# SPDX-License-Identifier: MIT
+
+defmodule AshPostgres.TemporalConcurrencyTest do
+  @moduledoc """
+  Concurrent temporal writes under READ COMMITTED. A write that blocks on a version
+  another transaction is splitting must still apply once that transaction commits, even
+  though the version it found has moved and the leftovers are newer than its snapshot.
+
+  These need real, committed transactions on separate connections, so they run outside
+  the sandbox and clean up after themselves.
+  """
+  use ExUnit.Case, async: false
+
+  @moduletag :temporal
+  @moduletag :postgres_18
+
+  require Ash.Query
+  alias AshPostgres.Test.Temporal.Subscription
+  alias AshPostgres.TemporalTestRepo
+  alias Ecto.Adapters.SQL.Sandbox
+
+  @jan1 ~U[2026-01-01 00:00:00.000000Z]
+  @jan15 ~U[2026-01-15 00:00:00.000000Z]
+  @feb1 ~U[2026-02-01 00:00:00.000000Z]
+
+  setup do
+    :ok = Sandbox.checkout(TemporalTestRepo, sandbox: false)
+    cleanup()
+
+    TemporalTestRepo.query!(
+      "INSERT INTO tier (id, name, valid_at) VALUES (100, 'basic', tstzrange('2026-01-01', NULL))"
+    )
+
+    TemporalTestRepo.query!("""
+    INSERT INTO subscription (id, tier, tier_id, seats, valid_at) VALUES
+      (101, 'orig', 100, 1, tstzrange('2026-01-01', NULL)),
+      (102, 'orig', 100, 1, tstzrange('2026-01-01', NULL))
+    """)
+
+    on_exit(fn ->
+      :ok = Sandbox.checkout(TemporalTestRepo, sandbox: false)
+      cleanup()
+    end)
+  end
+
+  defp cleanup do
+    TemporalTestRepo.query!("DELETE FROM subscription WHERE id IN (101, 102, 103)")
+    TemporalTestRepo.query!("DELETE FROM tier WHERE id = 100")
+  end
+
+  # Runs `fun` in a transaction on its own connection and holds it open until released.
+  defp hold_open(fun) do
+    parent = self()
+
+    task =
+      Task.async(fn ->
+        :ok = Sandbox.checkout(TemporalTestRepo, sandbox: false)
+
+        TemporalTestRepo.transaction(fn ->
+          fun.()
+          send(parent, :written)
+
+          receive do
+            :commit -> :ok
+          end
+        end)
+      end)
+
+    assert_receive :written, 5_000
+    task
+  end
+
+  # Starts `fun` on its own connection, lets it block on the held transaction, then
+  # commits that transaction so `fun` resumes.
+  defp race(held, fun) do
+    task =
+      Task.async(fn ->
+        :ok = Sandbox.checkout(TemporalTestRepo, sandbox: false)
+        fun.()
+      end)
+
+    Process.sleep(300)
+    send(held.pid, :commit)
+    Task.await(held)
+    Task.await(task)
+  end
+
+  defp change_tier(id, tier, as_of) do
+    Subscription
+    |> Ash.Query.filter(id == ^id)
+    |> Ash.Query.as_of(as_of)
+    |> Ash.read_one!()
+    |> Ash.Changeset.for_update(:change_tier, %{tier: tier})
+    |> Ash.Changeset.as_of(as_of)
+    |> Ash.update!()
+  end
+
+  defp add_seat_to_all(as_of) do
+    Subscription
+    |> Ash.Query.filter(id in [101, 102])
+    |> Ash.Query.as_of(as_of)
+    |> Ash.bulk_update!(:add_seat, %{}, strategy: :atomic)
+  end
+
+  defp upsert(id, tier, as_of) do
+    Subscription
+    |> Ash.Changeset.for_create(:create, %{id: id, tier: tier, tier_id: 100, seats: 1})
+    |> Ash.Changeset.as_of(as_of)
+    |> Ash.create!(upsert?: true)
+  end
+
+  defp timeline(id, column) do
+    TemporalTestRepo.query!(
+      "SELECT lower(valid_at), upper(valid_at), #{column} FROM subscription " <>
+        "WHERE id = $1 ORDER BY lower(valid_at)",
+      [id]
+    ).rows
+  end
+
+  test "an update isn't lost when the version it found is split under it" do
+    held = hold_open(fn -> change_tier(101, "a", @feb1) end)
+    race(held, fn -> change_tier(101, "b", @jan15) end)
+
+    assert [[@jan1, @jan15, "orig"], [@jan15, @feb1, "b"], [@feb1, nil, "a"]] =
+             timeline(101, "tier")
+  end
+
+  test "an update isn't lost when the version it found is ended under it" do
+    held =
+      hold_open(fn ->
+        Subscription
+        |> Ash.Query.filter(id == 101)
+        |> Ash.Query.as_of(@feb1)
+        |> Ash.bulk_destroy!(:expire, %{}, strategy: :atomic)
+      end)
+
+    race(held, fn -> change_tier(101, "b", @jan15) end)
+
+    assert [[@jan1, @jan15, "orig"], [@jan15, @feb1, "b"]] = timeline(101, "tier")
+  end
+
+  test "a bulk update isn't lost on any record whose version is split under it" do
+    held = hold_open(fn -> add_seat_to_all(@feb1) end)
+    race(held, fn -> add_seat_to_all(@jan15) end)
+
+    for id <- [101, 102] do
+      assert [[@jan1, @jan15, 1], [@jan15, @feb1, 2], [@feb1, nil, 2]] = timeline(id, "seats")
+    end
+  end
+
+  test "two upserts inserting the same new record both apply" do
+    held = hold_open(fn -> upsert(103, "a", @jan1) end)
+    race(held, fn -> upsert(103, "b", @feb1) end)
+
+    assert [[@jan1, @feb1, "a"], [@feb1, nil, "b"]] = timeline(103, "tier")
+  end
+
+  test "a bulk upsert retries just the record that lost an insert race" do
+    held = hold_open(fn -> upsert(103, "a", @jan1) end)
+
+    result =
+      race(held, fn ->
+        Ash.bulk_create!(
+          [
+            %{id: 101, tier: "b", tier_id: 100, seats: 1},
+            %{id: 103, tier: "b", tier_id: 100, seats: 1}
+          ],
+          Subscription,
+          :create,
+          upsert?: true,
+          upsert_fields: [:tier],
+          as_of: @feb1,
+          return_records?: true,
+          return_errors?: true
+        )
+      end)
+
+    assert %Ash.BulkResult{status: :success, records: records} = result
+    assert records |> Enum.map(& &1.id) |> Enum.sort() == [101, 103]
+    assert [[@jan1, @feb1, "orig"], [@feb1, nil, "b"]] = timeline(101, "tier")
+    assert [[@jan1, @feb1, "a"], [@feb1, nil, "b"]] = timeline(103, "tier")
+  end
+end
diff --git a/test/temporal_identity_test.exs b/test/temporal_identity_test.exs
index b3693a9c..2956d86e 100644
--- a/test/temporal_identity_test.exs
+++ b/test/temporal_identity_test.exs
@@ -3,12 +3,12 @@
 # SPDX-License-Identifier: MIT
 
 defmodule AshPostgres.TemporalIdentityTest do
-  @moduledoc "Identities on temporal resources: period-aware uniqueness + as_of-anchored eager checks (PG19)."
+  @moduledoc "Identities on temporal resources: period-aware uniqueness + as_of-anchored eager checks (PG18+)."
   use AshPostgres.RepoCase, async: false
   @moduletag :temporal
-  @moduletag :postgres_19
+  @moduletag :postgres_18
 
-  alias AshPostgres.TestRepo
+  alias AshPostgres.TemporalTestRepo
 
   defmodule Domain do
     @moduledoc false
@@ -25,7 +25,7 @@ defmodule AshPostgres.TemporalIdentityTest do
 
     postgres do
       table("temporal_id_thing")
-      repo(AshPostgres.TestRepo)
+      repo(AshPostgres.TemporalTestRepo)
     end
 
     temporal do
@@ -70,7 +70,7 @@ defmodule AshPostgres.TemporalIdentityTest do
   setup do
     # Mirror what the migration generator now emits for an identity on a temporal
     # resource: a period-aware GiST exclusion rather than a plain unique index.
-    TestRepo.query!("""
+    TemporalTestRepo.query!("""
     CREATE TABLE temporal_id_thing (
       id integer NOT NULL,
       name text,
@@ -95,7 +95,7 @@ defmodule AshPostgres.TemporalIdentityTest do
 
   test "an identity allows the same value across non-overlapping periods (history)" do
     # bounded historical "x" in [jan, mar) (the prior period of entity 1)
-    TestRepo.query!(
+    TemporalTestRepo.query!(
       "INSERT INTO temporal_id_thing (id, name, valid_at) VALUES (1, 'x', tstzrange('2026-01-01','2026-03-01','[)'))"
     )
 
@@ -104,7 +104,9 @@ defmodule AshPostgres.TemporalIdentityTest do
     assert {:ok, _} = create(%{id: 1, name: "x"}, @mar)
 
     rows =
-      TestRepo.query!("SELECT name FROM temporal_id_thing WHERE id = 1 ORDER BY lower(valid_at)").rows
+      TemporalTestRepo.query!(
+        "SELECT name FROM temporal_id_thing WHERE id = 1 ORDER BY lower(valid_at)"
+      ).rows
 
     assert rows == [["x"], ["x"]]
   end
@@ -118,7 +120,7 @@ defmodule AshPostgres.TemporalIdentityTest do
     near_future = DateTime.add(now, 365, :day)
     far_future = DateTime.add(now, 730, :day)
 
-    TestRepo.query!(
+    TemporalTestRepo.query!(
       "INSERT INTO temporal_id_thing (id, name, valid_at) VALUES (9, 'z', tstzrange($1, $2, '[)'))",
       [past, near_future]
     )
@@ -131,7 +133,7 @@ defmodule AshPostgres.TemporalIdentityTest do
 
   test "eager_check is anchored at as_of: a create conflicting at that instant is rejected" do
     # "z" valid across [jan, jun)
-    TestRepo.query!(
+    TemporalTestRepo.query!(
       "INSERT INTO temporal_id_thing (id, name, valid_at) VALUES (9, 'z', tstzrange('2026-01-01','2026-06-01','[)'))"
     )
 
diff --git a/test/temporal_migration_generator_test.exs b/test/temporal_migration_generator_test.exs
index 6b5e91db..73ec74cd 100644
--- a/test/temporal_migration_generator_test.exs
+++ b/test/temporal_migration_generator_test.exs
@@ -3,10 +3,10 @@
 # SPDX-License-Identifier: MIT
 
 defmodule AshPostgres.TemporalMigrationGeneratorTest do
-  @moduledoc "Asserts the migration generator emits temporal DDL (PG19)."
+  @moduledoc "Asserts the migration generator emits temporal DDL (PG18+)."
   use AshPostgres.RepoCase, async: false
   @moduletag :temporal
-  @moduletag :postgres_19
+  @moduletag :postgres_18
   @moduletag :tmp_dir
 
   defmodule GenTier do
@@ -15,7 +15,7 @@ defmodule AshPostgres.TemporalMigrationGeneratorTest do
 
     postgres do
       table("gen_tier")
-      repo(AshPostgres.TestRepo)
+      repo(AshPostgres.TemporalTestRepo)
     end
 
     temporal do
@@ -54,7 +54,7 @@ defmodule AshPostgres.TemporalMigrationGeneratorTest do
 
     postgres do
       table("gen_sub")
-      repo(AshPostgres.TestRepo)
+      repo(AshPostgres.TemporalTestRepo)
     end
 
     temporal do
@@ -96,7 +96,7 @@ defmodule AshPostgres.TemporalMigrationGeneratorTest do
 
     postgres do
       table("gen_a_badge")
-      repo(AshPostgres.TestRepo)
+      repo(AshPostgres.TemporalTestRepo)
     end
 
     temporal do
@@ -146,7 +146,7 @@ defmodule AshPostgres.TemporalMigrationGeneratorTest do
     }
   end
 
-  test "emits range column, WITHOUT OVERLAPS PK, PERIOD FK and btree_gist, version-guarded", %{
+  test "emits range column, WITHOUT OVERLAPS PK, PERIOD FK and btree_gist", %{
     snapshot_path: snapshot_path,
     migration_path: migration_path
   } do
@@ -168,20 +168,20 @@ defmodule AshPostgres.TemporalMigrationGeneratorTest do
     # range period column
     assert migration =~ ":tstzrange"
 
-    # temporal WITHOUT OVERLAPS primary key, version-guarded with a plain-PK fallback
+    # temporal WITHOUT OVERLAPS primary key, with no fallback: the repo's
+    # `min_pg_version/0` guarantees PostgreSQL 18+, so nothing is checked at migration time
     assert migration =~ "ADD PRIMARY KEY (id, valid_at WITHOUT OVERLAPS)"
-    assert migration =~ "ADD PRIMARY KEY (id)"
-    assert migration =~ "server_version_num"
-    assert migration =~ "190_000"
+    refute migration =~ "ADD PRIMARY KEY (id)"
+    refute migration =~ "server_version_num"
 
-    # temporal PERIOD foreign key (also version-guarded)
+    # temporal PERIOD foreign key
     assert migration =~ "FOREIGN KEY (tier_id, PERIOD valid_at)"
     assert migration =~ "PERIOD valid_at)"
 
     # identity on a temporal resource -> period-aware exclusion (WITHOUT OVERLAPS),
-    # version-guarded with a plain unique-index fallback on pre-PG19 servers.
+    # never a plain unique index
     assert migration =~ "EXCLUDE USING gist (name WITH =, valid_at WITH &&)"
-    assert migration =~ ~r/create unique_index\(:gen_tier, \[:name\]/
+    refute migration =~ ~r/create unique_index\(:gen_tier, \[:name\]/
 
     # GiST has no citext operator class, so citext keys compare as citext does
     assert migration =~ "EXCLUDE USING gist ((lower(email::text)) WITH =, valid_at WITH &&)"
diff --git a/test/temporal_test.exs b/test/temporal_test.exs
index 2dd2e0bb..6428bc6f 100644
--- a/test/temporal_test.exs
+++ b/test/temporal_test.exs
@@ -4,10 +4,10 @@
 
 defmodule AshPostgres.TemporalTest do
   @moduledoc """
-  End-to-end temporal (bitemporal) tests against Postgres 19.
+  End-to-end temporal (bitemporal) tests against PostgreSQL 18+.
 
   Exercises `as_of` time-travel reads, `now()` anchoring (filters/calcs/atomics),
-  the `Ash.Type.Range` <-> `Postgrex.Range` bridge, `FOR PORTION OF` mutations,
+  the `Ash.Type.Range` <-> `Postgrex.Range` bridge, period-splitting mutations,
   `range_overlaps/2`, and `temporal_keys` relationships.
 
   Tagged `:temporal`; run with:
@@ -17,12 +17,12 @@ defmodule AshPostgres.TemporalTest do
   use AshPostgres.RepoCase, async: false
 
   @moduletag :temporal
-  @moduletag :postgres_19
+  @moduletag :postgres_18
 
   require Ash.Query
   require Ash.Expr
   alias AshPostgres.Test.Temporal.{Event, Member, Subscription, Tier}
-  alias AshPostgres.TestRepo
+  alias AshPostgres.TemporalTestRepo
 
   @jan15 ~U[2026-01-15 00:00:00.000000Z]
   @mar1 ~U[2026-03-01 00:00:00.000000Z]
@@ -30,17 +30,17 @@ defmodule AshPostgres.TemporalTest do
   setup do
     # tier first: the temporal PERIOD foreign key requires the referenced tier
     # period to exist (and cover the subscription's period) before insert.
-    TestRepo.query!(
+    TemporalTestRepo.query!(
       "INSERT INTO tier (id, name, valid_at) VALUES (10, 'basic', tstzrange('2026-01-01','2026-06-01','[)'))"
     )
 
-    TestRepo.query!("""
+    TemporalTestRepo.query!("""
     INSERT INTO subscription (id, tier, tier_id, seats, activated_at, valid_at) VALUES
       (1, 'bronze', 10, 3, '2026-02-15', tstzrange('2026-01-01','2026-02-01','[)')),
       (1, 'gold',   10, 5, '2026-02-15', tstzrange('2026-02-01','2026-04-01','[)'))
     """)
 
-    TestRepo.query!("""
+    TemporalTestRepo.query!("""
     INSERT INTO event (id, name, created) VALUES
       (1, 'early', '2026-01-01'), (2, 'late', '2026-02-01')
     """)
@@ -51,14 +51,14 @@ defmodule AshPostgres.TemporalTest do
   defp tiers(records), do: records |> Enum.map(& &1.tier) |> Enum.sort()
 
   defp subscription_timeline(id) do
-    TestRepo.query!(
+    TemporalTestRepo.query!(
       "SELECT tier, lower(valid_at), upper(valid_at) FROM subscription WHERE id = $1 ORDER BY lower(valid_at)",
       [id]
     ).rows
   end
 
   defp seats_timeline(id) do
-    TestRepo.query!(
+    TemporalTestRepo.query!(
       "SELECT tier, seats, lower(valid_at) FROM subscription WHERE id = $1 ORDER BY lower(valid_at)",
       [id]
     ).rows
@@ -67,7 +67,7 @@ defmodule AshPostgres.TemporalTest do
   describe "default as_of (reads are current-state by default)" do
     test "get / read without as_of returns the currently-valid period, not all history" do
       # id=8: a past period + a current open-ended one (tier_id NULL avoids the PERIOD FK)
-      TestRepo.query!("""
+      TemporalTestRepo.query!("""
       INSERT INTO subscription (id, tier, tier_id, valid_at) VALUES
         (8, 'old',     NULL, tstzrange('2020-01-01','2021-01-01','[)')),
         (8, 'current', NULL, tstzrange('2021-01-01', NULL, '[)'))
@@ -102,7 +102,7 @@ defmodule AshPostgres.TemporalTest do
   describe "sorting through a temporal relationship" do
     setup do
       # tier 30 changes name across two adjacent periods; tier 40 is constant.
-      TestRepo.query!("""
+      TemporalTestRepo.query!("""
       INSERT INTO tier (id, name, valid_at) VALUES
         (30, 'mango',  tstzrange('2026-01-01','2026-03-01','[)')),
         (30, 'apple',  tstzrange('2026-03-01','2026-06-01','[)')),
@@ -111,7 +111,7 @@ defmodule AshPostgres.TemporalTest do
 
       # Each subscription spans the whole window (sub 5's span covers BOTH of tier 30's
       # periods — the as_of join must pick the single covering tier row, not both).
-      TestRepo.query!("""
+      TemporalTestRepo.query!("""
       INSERT INTO subscription (id, tier, tier_id, valid_at) VALUES
         (5, 'sub5', 30, tstzrange('2026-01-01','2026-06-01','[)')),
         (6, 'sub6', 40, tstzrange('2026-01-01','2026-06-01','[)'))
@@ -173,13 +173,13 @@ defmodule AshPostgres.TemporalTest do
   describe "Ash.load carries as_of (like tenant)" do
     setup do
       # tier 50 changes name across periods; subscription 7 spans both.
-      TestRepo.query!("""
+      TemporalTestRepo.query!("""
       INSERT INTO tier (id, name, valid_at) VALUES
         (50, 'old', tstzrange('2026-01-01','2026-03-01','[)')),
         (50, 'new', tstzrange('2026-03-01','2026-06-01','[)'))
       """)
 
-      TestRepo.query!("""
+      TemporalTestRepo.query!("""
       INSERT INTO subscription (id, tier, tier_id, valid_at) VALUES
         (7, 'sub7', 50, tstzrange('2026-01-01','2026-06-01','[)'))
       """)
@@ -287,14 +287,14 @@ defmodule AshPostgres.TemporalTest do
     end
   end
 
-  describe "FOR PORTION OF preserves an unbounded (NULL) upper" do
+  describe "splitting preserves an unbounded (NULL) upper" do
     test "updating a current row yields [as_of, ∞) with no 'infinity' upper or junk row" do
       # open-ended "current" row ([2026-01-01, ∞)); tier_id NULL avoids the PERIOD FK
-      TestRepo.query!(
+      TemporalTestRepo.query!(
         "INSERT INTO subscription (id, tier, tier_id, seats, valid_at) VALUES (20, 'x', NULL, 1, tstzrange('2026-01-01', NULL, '[)'))"
       )
 
-      # An atomic FOR PORTION OF update at @mar1 splits [jan,∞) -> [jan,mar1) + [mar1,∞).
+      # An atomic update at @mar1 splits [jan,∞) -> [jan,mar1) + [mar1,∞).
       # Returning the new slice previously crashed (literal 'infinity' upper); now it's NULL.
       updated =
         Subscription
@@ -307,13 +307,13 @@ defmodule AshPostgres.TemporalTest do
 
       # exactly two contiguous periods, the later one unbounded, and NO [infinity, ) junk row
       rows =
-        TestRepo.query!(
+        TemporalTestRepo.query!(
           "SELECT upper_inf(valid_at), seats FROM subscription WHERE id = 20 ORDER BY lower(valid_at)"
         ).rows
 
       assert rows == [[false, 1], [true, 2]]
 
-      assert TestRepo.query!(
+      assert TemporalTestRepo.query!(
                "SELECT count(*) FROM subscription WHERE id = 20 AND lower(valid_at) = 'infinity'"
              ).rows == [[0]]
     end
@@ -357,7 +357,7 @@ defmodule AshPostgres.TemporalTest do
     test "reads the rows valid at the current instant" do
       # An open-ended ([2020, ∞)) row that always covers the wall clock.
       # `tier_id` is NULL so it isn't subject to the temporal PERIOD FK.
-      TestRepo.query!(
+      TemporalTestRepo.query!(
         "INSERT INTO subscription (id, tier, tier_id, valid_at) VALUES (9, 'current', NULL, tstzrange('2020-01-01', NULL, '[)'))"
       )
 
@@ -467,7 +467,7 @@ defmodule AshPostgres.TemporalTest do
     end
   end
 
-  describe "FOR PORTION OF mutations" do
+  describe "period-splitting mutations" do
     test "update splits the row at as_of, applying the new value forward" do
       [gold] = Subscription |> Ash.Query.filter(id == 1) |> Ash.Query.as_of(@mar1) |> Ash.read!()
 
@@ -721,7 +721,7 @@ defmodule AshPostgres.TemporalTest do
     end
 
     test "an atomic update through an exists at the current instant leaves later versions alone" do
-      TestRepo.query!("""
+      TemporalTestRepo.query!("""
       INSERT INTO subscription (id, tier, seats, valid_at) VALUES
         (2, 'now', 1, tstzrange(now() - interval '1 day', now() + interval '1 day', '[)')),
         (2, 'later', 9, tstzrange(now() + interval '1 day', NULL, '[)'))
@@ -1010,7 +1010,7 @@ defmodule AshPostgres.TemporalTest do
   describe "aggregates respect as_of" do
     test "a count over a temporal has_many counts only periods valid at as_of" do
       # add a second subscription (id=3) for tier 10, valid only [Mar,May)
-      TestRepo.query!("""
+      TemporalTestRepo.query!("""
       INSERT INTO subscription (id, tier, tier_id, seats, valid_at) VALUES
         (3, 'silver', 10, 1, tstzrange('2026-03-01','2026-05-01','[)'))
       """)
@@ -1036,7 +1036,7 @@ defmodule AshPostgres.TemporalTest do
   describe "bulk_create" do
     test "distinct keys succeed; same key at the same as_of raises the exclusion constraint" do
       # open-ended tier so each create's `[as_of, ∞)` satisfies the PERIOD FK
-      TestRepo.query!(
+      TemporalTestRepo.query!(
         "INSERT INTO tier (id, name, valid_at) VALUES (20, 'open', tstzrange('2026-01-01', NULL, '[)'))"
       )
 
@@ -1087,7 +1087,7 @@ defmodule AshPostgres.TemporalTest do
     end
   end
 
-  describe "temporal upsert (atomic FOR-PORTION-OF-or-insert CTE)" do
+  describe "temporal upsert (atomic split-or-insert CTE)" do
     test "match: splits the period valid at as_of, applying new values forward" do
       # id=1: bronze [jan,feb), gold [feb,apr). Upsert at mar1 (inside gold).
       assert {:ok, result} =
@@ -1108,7 +1108,7 @@ defmodule AshPostgres.TemporalTest do
     end
 
     test "match: future periods (that don't contain as_of) are untouched" do
-      TestRepo.query!(
+      TemporalTestRepo.query!(
         "INSERT INTO subscription (id, tier, tier_id, seats, valid_at) VALUES (1,'future',10,1,tstzrange('2026-05-01','2026-06-01','[)'))"
       )
 
@@ -1124,13 +1124,13 @@ defmodule AshPostgres.TemporalTest do
                |> Ash.create(upsert?: true)
 
       assert [["future", 1]] =
-               TestRepo.query!(
+               TemporalTestRepo.query!(
                  "SELECT tier, seats FROM subscription WHERE id=1 AND lower(valid_at)='2026-05-01'"
                ).rows
     end
 
     test "no match: inserts a new period gap-filled to infinity" do
-      TestRepo.query!(
+      TemporalTestRepo.query!(
         "INSERT INTO tier (id, name, valid_at) VALUES (20, 'open', tstzrange('2026-01-01', NULL, '[)'))"
       )
 
@@ -1143,14 +1143,28 @@ defmodule AshPostgres.TemporalTest do
       assert result.tier == "new"
 
       assert [["new", "2026-01-15 00:00:00+00", nil]] =
-               TestRepo.query!(
+               TemporalTestRepo.query!(
                  "SELECT tier, lower(valid_at)::text, upper(valid_at) FROM subscription WHERE id=2"
                ).rows
     end
 
+    test "an identity violation still raises (only the primary key is retried)" do
+      TemporalTestRepo.query!(
+        "INSERT INTO member (id, email, valid_at) VALUES (1, 'taken', tstzrange('2026-01-01', NULL))"
+      )
+
+      assert {:error, %Ash.Error.Invalid{} = error} =
+               Member
+               |> Ash.Changeset.for_create(:create, %{id: 2, email: "taken"})
+               |> Ash.Changeset.as_of(@mar1)
+               |> Ash.create(upsert?: true)
+
+      assert Exception.message(error) =~ "email"
+    end
+
     test "bulk: upserts many keys in one statement (match splits, miss inserts)" do
       # open-ended tier so the new id=5 row's [mar1, ∞) satisfies the PERIOD FK
-      TestRepo.query!(
+      TemporalTestRepo.query!(
         "INSERT INTO tier (id, name, valid_at) VALUES (20, 'open', tstzrange('2026-01-01', NULL, '[)'))"
       )
 
@@ -1176,9 +1190,9 @@ defmodule AshPostgres.TemporalTest do
   end
 
   describe "bulk_update / update_many" do
-    # The `FOR PORTION OF FROM` bound is sourced from the query context as_of
+    # The split point is sourced from the query context as_of
     # (via `temporal_from_bound`), which carries it for both single and bulk.
-    test "each matched row is split at as_of (FOR PORTION OF, set-wide)" do
+    test "each matched row is split at as_of (set-wide)" do
       Subscription
       |> Ash.Query.filter(id == 1)
       |> Ash.Query.as_of(@mar1)
@@ -1195,7 +1209,7 @@ defmodule AshPostgres.TemporalTest do
 
   describe "bulk_destroy" do
     # FROM bound sourced from query context as_of (via `temporal_from_bound`).
-    test "each matched row's validity is truncated at as_of (FOR PORTION OF DELETE)" do
+    test "each matched row's validity is truncated at as_of (temporal DELETE)" do
       Subscription
       |> Ash.Query.filter(id == 1)
       |> Ash.Query.as_of(@mar1)
@@ -1209,7 +1223,7 @@ defmodule AshPostgres.TemporalTest do
   end
 
   describe "temporal PERIOD foreign keys reject referential actions" do
-    test "on_delete on a temporal relationship fails to compile (PG19: NO ACTION only)" do
+    test "on_delete on a temporal relationship fails to compile (PERIOD FKs: NO ACTION only)" do
       assert_raise Spark.Error.DslError, ~r/PERIOD|NO ACTION/, fn ->
         defmodule BadTemporalRef do
           use Ash.Resource,
@@ -1219,7 +1233,7 @@ defmodule AshPostgres.TemporalTest do
 
           postgres do
             table("bad_temporal_ref")
-            repo(AshPostgres.TestRepo)
+            repo(AshPostgres.TemporalTestRepo)
 
             references do
               reference(:tier_record, on_delete: :delete)
@@ -1255,6 +1269,63 @@ defmodule AshPostgres.TemporalTest do
     end
   end
 
+  describe "a write that keeps conflicting" do
+    test "fails with an Ash framework error that says retrying is safe" do
+      error =
+        Ash.Error.to_error_class(
+          AshPostgres.Temporal.WriteConflict.exception(resource: Subscription, attempts: 25)
+        )
+
+      assert %Ash.Error.Framework{errors: [%AshPostgres.Temporal.WriteConflict{}]} = error
+      assert Exception.message(error) =~ "conflicted with concurrent writes"
+      assert Exception.message(error) =~ "Retrying the action is safe"
+    end
+  end
+
+  describe "a temporal resource's repo must target PostgreSQL 18+" do
+    defmodule PrePg18Repo do
+      @moduledoc false
+      use AshPostgres.Repo, otp_app: :ash_postgres
+
+      def min_pg_version, do: %Version{major: 17, minor: 0, patch: 0}
+      def installed_extensions, do: ["ash-functions", "btree_gist"]
+    end
+
+    # Verifiers run after a module is verified, which doesn't happen synchronously for one
+    # defined in a test, so this runs the verifier on the resource directly.
+    test "a repo whose min_pg_version is below 18 fails to compile" do
+      # Spark reports the verifier's error on its own once the module is verified.
+      ExUnit.CaptureIO.capture_io(:stderr, fn ->
+        defmodule TemporalOnPrePg18 do
+          use Ash.Resource,
+            domain: nil,
+            validate_domain_inclusion?: false,
+            data_layer: AshPostgres.DataLayer
+
+          postgres do
+            table("temporal_on_pre_pg18")
+            repo(PrePg18Repo)
+          end
+
+          temporal do
+            strategy(:context)
+            attribute(:valid_at)
+          end
+
+          attributes do
+            attribute(:id, :integer, primary_key?: true, allow_nil?: false, public?: true)
+          end
+        end
+      end)
+
+      assert_raise Spark.Error.DslError, ~r/requires PostgreSQL 18 or later/, fn ->
+        AshPostgres.Verifiers.VerifyTemporal.verify(
+          AshPostgres.TemporalTest.TemporalOnPrePg18.spark_dsl_config()
+        )
+      end
+    end
+  end
+
   describe "the period attribute is not settable as input" do
     test "a resource that accepts the temporal attribute fails to compile" do
       assert_raise Spark.Error.DslError, ~r/must not be accepted as input/, fn ->
@@ -1266,7 +1337,7 @@ defmodule AshPostgres.TemporalTest do
 
           postgres do
             table("bad_temporal")
-            repo(AshPostgres.TestRepo)
+            repo(AshPostgres.TemporalTestRepo)
           end
 
           temporal do
@@ -1313,7 +1384,7 @@ defmodule AshPostgres.TemporalTest do
 
       # and the tier itself ended at mar1
       assert [["2026-01-01 00:00:00+00", "2026-03-01 00:00:00+00"]] =
-               TestRepo.query!(
+               TemporalTestRepo.query!(
                  "SELECT lower(valid_at)::text, upper(valid_at)::text FROM tier WHERE id = 10"
                ).rows
     end
@@ -1321,7 +1392,7 @@ defmodule AshPostgres.TemporalTest do
 
   describe "cascade destroy carries the destroy's as_of" do
     defp tier_timeline do
-      TestRepo.query!(
+      TemporalTestRepo.query!(
         "SELECT lower(valid_at)::text, upper(valid_at)::text FROM tier WHERE id = 10 ORDER BY lower(valid_at)"
       ).rows
     end
@@ -1410,7 +1481,7 @@ defmodule AshPostgres.TemporalTest do
       # the subscription's [jan15, ∞) (the temporal PERIOD FK requires it). A child written
       # at "now" would be [now, ∞) and fail the FK / show the wrong lower bound.
       assert [["2026-01-15 00:00:00+00", nil]] =
-               TestRepo.query!(
+               TemporalTestRepo.query!(
                  "SELECT lower(valid_at)::text, upper(valid_at) FROM tier WHERE id = 30"
                ).rows
     end
@@ -1443,7 +1514,7 @@ defmodule AshPostgres.TemporalTest do
 
   describe "recorded_at is the wall clock, not as_of" do
     defp recorded_timeline(id) do
-      TestRepo.query!(
+      TemporalTestRepo.query!(
         "SELECT tier, recorded_at FROM subscription WHERE id = $1 ORDER BY lower(valid_at)",
         [id]
       ).rows
@@ -1461,7 +1532,7 @@ defmodule AshPostgres.TemporalTest do
     end
 
     test "a back-dated create is recorded now" do
-      TestRepo.query!(
+      TemporalTestRepo.query!(
         "INSERT INTO tier (id, name, valid_at) VALUES (20, 'open', tstzrange('2026-01-01', NULL, '[)'))"
       )
 
diff --git a/test/test_helper.exs b/test/test_helper.exs
index 82928b22..2687a430 100644
--- a/test/test_helper.exs
+++ b/test/test_helper.exs
@@ -32,6 +32,11 @@ AshPostgres.TestRepo.start_link()
 AshPostgres.DevTestRepo.start_link()
 AshPostgres.TestNoSandboxRepo.start_link()
 
+# Only migrated on PostgreSQL 18+ (see `AshPostgres.TemporalTestRepo`).
+if pg_version >= 18 do
+  AshPostgres.TemporalTestRepo.start_link()
+end
+
 format_sql_query =
   try do
     case System.shell("which pg_format") do
diff --git a/test_priv/resource_snapshots/test_repo/event/20260627032019.json b/test_priv/resource_snapshots/temporal_test_repo/event/20261004073135.json
similarity index 89%
rename from test_priv/resource_snapshots/test_repo/event/20260627032019.json
rename to test_priv/resource_snapshots/temporal_test_repo/event/20261004073135.json
index a07709a8..f6f152e9 100644
--- a/test_priv/resource_snapshots/test_repo/event/20260627032019.json
+++ b/test_priv/resource_snapshots/temporal_test_repo/event/20261004073135.json
@@ -43,14 +43,14 @@
   "custom_indexes": [],
   "custom_statements": [],
   "has_create_action": false,
-  "hash": "564460F32319359E8E086E341F05793F960D4A5AD72521313CFFA05C7CBDDD1A",
+  "hash": "EA6BD551D1DD7BB8A2CD2B99E3009290FF8F1AE54431BBF8699E289B65148F58",
   "identities": [],
   "multitenancy": {
     "attribute": null,
     "global": null,
     "strategy": null
   },
-  "repo": "Elixir.AshPostgres.TestRepo",
+  "repo": "Elixir.AshPostgres.TemporalTestRepo",
   "schema": null,
   "table": "event",
   "temporal": null
diff --git a/test_priv/resource_snapshots/test_repo/event/20260627032019.json.license b/test_priv/resource_snapshots/temporal_test_repo/event/20261004073135.json.license
similarity index 100%
rename from test_priv/resource_snapshots/test_repo/event/20260627032019.json.license
rename to test_priv/resource_snapshots/temporal_test_repo/event/20261004073135.json.license
diff --git a/test_priv/resource_snapshots/temporal_test_repo/extensions.json b/test_priv/resource_snapshots/temporal_test_repo/extensions.json
new file mode 100644
index 00000000..c958e4d1
--- /dev/null
+++ b/test_priv/resource_snapshots/temporal_test_repo/extensions.json
@@ -0,0 +1,8 @@
+{
+  "ash_functions_version": 7,
+  "installed": [
+    "ash-functions",
+    "citext",
+    "btree_gist"
+  ]
+}
\ No newline at end of file
diff --git a/test_priv/resource_snapshots/test_repo/event/20260824043711.json.license b/test_priv/resource_snapshots/temporal_test_repo/extensions.json.license
similarity index 100%
rename from test_priv/resource_snapshots/test_repo/event/20260824043711.json.license
rename to test_priv/resource_snapshots/temporal_test_repo/extensions.json.license
diff --git a/test_priv/resource_snapshots/test_repo/member/20260926192339.json b/test_priv/resource_snapshots/temporal_test_repo/member/20261004073136.json
similarity index 92%
rename from test_priv/resource_snapshots/test_repo/member/20260926192339.json
rename to test_priv/resource_snapshots/temporal_test_repo/member/20261004073136.json
index 90497a1f..7e9cb5a8 100644
--- a/test_priv/resource_snapshots/test_repo/member/20260926192339.json
+++ b/test_priv/resource_snapshots/temporal_test_repo/member/20261004073136.json
@@ -43,7 +43,7 @@
   "custom_indexes": [],
   "custom_statements": [],
   "has_create_action": true,
-  "hash": "82A6392870511BE46BFCBE836F3527E4D4969B8C418C08DA4FC7D34D423850B1",
+  "hash": "D2D92A32204963CA53D91F4E82EA30039FFEBDF4CCF3D62641344C50E8610746",
   "identities": [
     {
       "all_tenants?": false,
@@ -65,7 +65,7 @@
     "global": null,
     "strategy": null
   },
-  "repo": "Elixir.AshPostgres.TestRepo",
+  "repo": "Elixir.AshPostgres.TemporalTestRepo",
   "schema": null,
   "table": "member",
   "temporal": {
diff --git a/test_priv/resource_snapshots/test_repo/member/20260926192339.json.license b/test_priv/resource_snapshots/temporal_test_repo/member/20261004073136.json.license
similarity index 100%
rename from test_priv/resource_snapshots/test_repo/member/20260926192339.json.license
rename to test_priv/resource_snapshots/temporal_test_repo/member/20261004073136.json.license
diff --git a/test_priv/resource_snapshots/test_repo/subscription/20260927024130.json b/test_priv/resource_snapshots/temporal_test_repo/subscription/20261004073137.json
similarity index 95%
rename from test_priv/resource_snapshots/test_repo/subscription/20260927024130.json
rename to test_priv/resource_snapshots/temporal_test_repo/subscription/20261004073137.json
index 841f3658..9de1f143 100644
--- a/test_priv/resource_snapshots/test_repo/subscription/20260927024130.json
+++ b/test_priv/resource_snapshots/temporal_test_repo/subscription/20261004073137.json
@@ -116,14 +116,14 @@
   "custom_indexes": [],
   "custom_statements": [],
   "has_create_action": true,
-  "hash": "FA1ADDB90BF620ACD6E8BEFFE489DD56995132A8BE5F05281DFAC3C5858087BB",
+  "hash": "0D9D101FE4C593F8EFFC03DC84F5B4249ECEC5C08E2F699816D74C9D37FE3F0E",
   "identities": [],
   "multitenancy": {
     "attribute": null,
     "global": null,
     "strategy": null
   },
-  "repo": "Elixir.AshPostgres.TestRepo",
+  "repo": "Elixir.AshPostgres.TemporalTestRepo",
   "schema": null,
   "table": "subscription",
   "temporal": {
diff --git a/test_priv/resource_snapshots/test_repo/subscription/20260627032021.json.license b/test_priv/resource_snapshots/temporal_test_repo/subscription/20261004073137.json.license
similarity index 100%
rename from test_priv/resource_snapshots/test_repo/subscription/20260627032021.json.license
rename to test_priv/resource_snapshots/temporal_test_repo/subscription/20261004073137.json.license
diff --git a/test_priv/resource_snapshots/test_repo/tier/20260824043712.json b/test_priv/resource_snapshots/temporal_test_repo/tier/20261004073138.json
similarity index 90%
rename from test_priv/resource_snapshots/test_repo/tier/20260824043712.json
rename to test_priv/resource_snapshots/temporal_test_repo/tier/20261004073138.json
index 11549929..9ab54f2e 100644
--- a/test_priv/resource_snapshots/test_repo/tier/20260824043712.json
+++ b/test_priv/resource_snapshots/temporal_test_repo/tier/20261004073138.json
@@ -43,14 +43,14 @@
   "custom_indexes": [],
   "custom_statements": [],
   "has_create_action": true,
-  "hash": "94A1C6C459EB720CC502D358935D76F39DD1D36D15A5868B67BCA4A16860BA1E",
+  "hash": "4C05418764E40BBAC82177FD5660CB06CD04C36A81D47C40D53A1FBFA02BCF00",
   "identities": [],
   "multitenancy": {
     "attribute": null,
     "global": null,
     "strategy": null
   },
-  "repo": "Elixir.AshPostgres.TestRepo",
+  "repo": "Elixir.AshPostgres.TemporalTestRepo",
   "schema": null,
   "table": "tier",
   "temporal": {
diff --git a/test_priv/resource_snapshots/test_repo/subscription/20260824043713.json.license b/test_priv/resource_snapshots/temporal_test_repo/tier/20261004073138.json.license
similarity index 100%
rename from test_priv/resource_snapshots/test_repo/subscription/20260824043713.json.license
rename to test_priv/resource_snapshots/temporal_test_repo/tier/20261004073138.json.license
diff --git a/test_priv/resource_snapshots/test_repo/event/20260824043711.json b/test_priv/resource_snapshots/test_repo/event/20260824043711.json
deleted file mode 100644
index 13fa9083..00000000
--- a/test_priv/resource_snapshots/test_repo/event/20260824043711.json
+++ /dev/null
@@ -1,57 +0,0 @@
-{
-  "attributes": [
-    {
-      "allow_nil?": false,
-      "default": "nil",
-      "generated?": false,
-      "precision": null,
-      "primary_key?": true,
-      "references": null,
-      "scale": null,
-      "size": null,
-      "source": "id",
-      "type": "bigint"
-    },
-    {
-      "allow_nil?": true,
-      "default": "nil",
-      "generated?": false,
-      "precision": null,
-      "primary_key?": false,
-      "references": null,
-      "scale": null,
-      "size": null,
-      "source": "name",
-      "type": "text"
-    },
-    {
-      "allow_nil?": true,
-      "default": "nil",
-      "generated?": false,
-      "precision": null,
-      "primary_key?": false,
-      "references": null,
-      "scale": null,
-      "size": null,
-      "source": "created",
-      "type": "utc_datetime_usec"
-    }
-  ],
-  "base_filter": null,
-  "check_constraints": [],
-  "create_table_options": null,
-  "custom_indexes": [],
-  "custom_statements": [],
-  "has_create_action": false,
-  "hash": "E7967071B2D8054BFE5787408ED2DE625F213CCD28C592078468D413C2BB5B8C",
-  "identities": [],
-  "multitenancy": {
-    "attribute": null,
-    "global": null,
-    "strategy": null
-  },
-  "repo": "Elixir.AshPostgres.TestRepo",
-  "schema": null,
-  "table": "event",
-  "temporal": null
-}
\ No newline at end of file
diff --git a/test_priv/resource_snapshots/test_repo/subscription/20260627032021.json b/test_priv/resource_snapshots/test_repo/subscription/20260627032021.json
deleted file mode 100644
index f4b21c3c..00000000
--- a/test_priv/resource_snapshots/test_repo/subscription/20260627032021.json
+++ /dev/null
@@ -1,119 +0,0 @@
-{
-  "attributes": [
-    {
-      "allow_nil?": false,
-      "default": "nil",
-      "generated?": false,
-      "precision": null,
-      "primary_key?": true,
-      "references": null,
-      "scale": null,
-      "size": null,
-      "source": "id",
-      "type": "bigint"
-    },
-    {
-      "allow_nil?": true,
-      "default": "nil",
-      "generated?": false,
-      "precision": null,
-      "primary_key?": false,
-      "references": null,
-      "scale": null,
-      "size": null,
-      "source": "tier",
-      "type": "text"
-    },
-    {
-      "allow_nil?": true,
-      "default": "nil",
-      "generated?": false,
-      "precision": null,
-      "primary_key?": false,
-      "references": {
-        "deferrable": false,
-        "destination_attribute": "id",
-        "destination_attribute_default": null,
-        "destination_attribute_generated": null,
-        "index?": false,
-        "match_type": null,
-        "match_with": null,
-        "multitenancy": {
-          "attribute": null,
-          "global": null,
-          "strategy": null
-        },
-        "name": "subscription_tier_id_fkey",
-        "on_delete": null,
-        "on_update": null,
-        "primary_key?": true,
-        "schema": "public",
-        "table": "tier",
-        "temporal_period": [
-          "valid_at",
-          "valid_at"
-        ]
-      },
-      "scale": null,
-      "size": null,
-      "source": "tier_id",
-      "type": "bigint"
-    },
-    {
-      "allow_nil?": true,
-      "default": "0",
-      "generated?": false,
-      "precision": null,
-      "primary_key?": false,
-      "references": null,
-      "scale": null,
-      "size": null,
-      "source": "seats",
-      "type": "bigint"
-    },
-    {
-      "allow_nil?": true,
-      "default": "nil",
-      "generated?": false,
-      "precision": null,
-      "primary_key?": false,
-      "references": null,
-      "scale": null,
-      "size": null,
-      "source": "activated_at",
-      "type": "utc_datetime_usec"
-    },
-    {
-      "allow_nil?": true,
-      "default": "nil",
-      "generated?": false,
-      "precision": null,
-      "primary_key?": false,
-      "references": null,
-      "scale": null,
-      "size": null,
-      "source": "valid_at",
-      "type": "tstzrange"
-    }
-  ],
-  "base_filter": null,
-  "check_constraints": [],
-  "create_table_options": null,
-  "custom_indexes": [],
-  "custom_statements": [],
-  "has_create_action": true,
-  "hash": "7B9A78F8010D1D9C80CE76721335887B4C40C4554C82C93692332AA6DF7A84D6",
-  "identities": [],
-  "multitenancy": {
-    "attribute": null,
-    "global": null,
-    "strategy": null
-  },
-  "repo": "Elixir.AshPostgres.TestRepo",
-  "schema": null,
-  "table": "subscription",
-  "temporal": {
-    "attribute": "valid_at",
-    "strategy": "context"
-  }
-}
\ No newline at end of file
diff --git a/test_priv/resource_snapshots/test_repo/subscription/20260824043713.json b/test_priv/resource_snapshots/test_repo/subscription/20260824043713.json
deleted file mode 100644
index e1c66e20..00000000
--- a/test_priv/resource_snapshots/test_repo/subscription/20260824043713.json
+++ /dev/null
@@ -1,121 +0,0 @@
-{
-  "attributes": [
-    {
-      "allow_nil?": false,
-      "default": "nil",
-      "generated?": false,
-      "precision": null,
-      "primary_key?": true,
-      "references": null,
-      "scale": null,
-      "size": null,
-      "source": "id",
-      "type": "bigint"
-    },
-    {
-      "allow_nil?": true,
-      "default": "nil",
-      "generated?": false,
-      "precision": null,
-      "primary_key?": false,
-      "references": null,
-      "scale": null,
-      "size": null,
-      "source": "tier",
-      "type": "text"
-    },
-    {
-      "allow_nil?": true,
-      "default": "nil",
-      "generated?": false,
-      "precision": null,
-      "primary_key?": false,
-      "references": {
-        "deferrable": false,
-        "destination_attribute": "id",
-        "destination_attribute_default": null,
-        "destination_attribute_generated": null,
-        "index?": false,
-        "index_where": null,
-        "match_tenant?": false,
-        "match_type": null,
-        "match_with": null,
-        "multitenancy": {
-          "attribute": null,
-          "global": null,
-          "strategy": null
-        },
-        "name": "subscription_tier_id_fkey",
-        "on_delete": null,
-        "on_update": null,
-        "primary_key?": true,
-        "schema": "public",
-        "table": "tier",
-        "temporal_period": [
-          "valid_at",
-          "valid_at"
-        ]
-      },
-      "scale": null,
-      "size": null,
-      "source": "tier_id",
-      "type": "bigint"
-    },
-    {
-      "allow_nil?": true,
-      "default": "0",
-      "generated?": false,
-      "precision": null,
-      "primary_key?": false,
-      "references": null,
-      "scale": null,
-      "size": null,
-      "source": "seats",
-      "type": "bigint"
-    },
-    {
-      "allow_nil?": true,
-      "default": "nil",
-      "generated?": false,
-      "precision": null,
-      "primary_key?": false,
-      "references": null,
-      "scale": null,
-      "size": null,
-      "source": "activated_at",
-      "type": "utc_datetime_usec"
-    },
-    {
-      "allow_nil?": false,
-      "default": "nil",
-      "generated?": true,
-      "precision": null,
-      "primary_key?": false,
-      "references": null,
-      "scale": null,
-      "size": null,
-      "source": "valid_at",
-      "type": "tstzrange"
-    }
-  ],
-  "base_filter": null,
-  "check_constraints": [],
-  "create_table_options": null,
-  "custom_indexes": [],
-  "custom_statements": [],
-  "has_create_action": true,
-  "hash": "BB2835F7273E3D9FC169BC7715F0E8363991B560D5C23FCFB3CF9D7A41B2FAE5",
-  "identities": [],
-  "multitenancy": {
-    "attribute": null,
-    "global": null,
-    "strategy": null
-  },
-  "repo": "Elixir.AshPostgres.TestRepo",
-  "schema": null,
-  "table": "subscription",
-  "temporal": {
-    "attribute": "valid_at",
-    "strategy": "context"
-  }
-}
\ No newline at end of file
diff --git a/test_priv/resource_snapshots/test_repo/subscription/20260927024130.json.license b/test_priv/resource_snapshots/test_repo/subscription/20260927024130.json.license
deleted file mode 100644
index c5bd9989..00000000
--- a/test_priv/resource_snapshots/test_repo/subscription/20260927024130.json.license
+++ /dev/null
@@ -1,3 +0,0 @@
-SPDX-FileCopyrightText: 2019 ash_postgres contributors <https://github.com/ash-project/ash_postgres/graphs/contributors>
-
-SPDX-License-Identifier: MIT
diff --git a/test_priv/resource_snapshots/test_repo/tier/20260627032020.json b/test_priv/resource_snapshots/test_repo/tier/20260627032020.json
deleted file mode 100644
index ea494646..00000000
--- a/test_priv/resource_snapshots/test_repo/tier/20260627032020.json
+++ /dev/null
@@ -1,60 +0,0 @@
-{
-  "attributes": [
-    {
-      "allow_nil?": false,
-      "default": "nil",
-      "generated?": false,
-      "precision": null,
-      "primary_key?": true,
-      "references": null,
-      "scale": null,
-      "size": null,
-      "source": "id",
-      "type": "bigint"
-    },
-    {
-      "allow_nil?": true,
-      "default": "nil",
-      "generated?": false,
-      "precision": null,
-      "primary_key?": false,
-      "references": null,
-      "scale": null,
-      "size": null,
-      "source": "name",
-      "type": "text"
-    },
-    {
-      "allow_nil?": true,
-      "default": "nil",
-      "generated?": false,
-      "precision": null,
-      "primary_key?": false,
-      "references": null,
-      "scale": null,
-      "size": null,
-      "source": "valid_at",
-      "type": "tstzrange"
-    }
-  ],
-  "base_filter": null,
-  "check_constraints": [],
-  "create_table_options": null,
-  "custom_indexes": [],
-  "custom_statements": [],
-  "has_create_action": false,
-  "hash": "D29673CAB290355F8F1081580F32CA27552F9B5B68ECCA0E31EA453FD9426BD2",
-  "identities": [],
-  "multitenancy": {
-    "attribute": null,
-    "global": null,
-    "strategy": null
-  },
-  "repo": "Elixir.AshPostgres.TestRepo",
-  "schema": null,
-  "table": "tier",
-  "temporal": {
-    "attribute": "valid_at",
-    "strategy": "context"
-  }
-}
\ No newline at end of file
diff --git a/test_priv/resource_snapshots/test_repo/tier/20260627032020.json.license b/test_priv/resource_snapshots/test_repo/tier/20260627032020.json.license
deleted file mode 100644
index c5bd9989..00000000
--- a/test_priv/resource_snapshots/test_repo/tier/20260627032020.json.license
+++ /dev/null
@@ -1,3 +0,0 @@
-SPDX-FileCopyrightText: 2019 ash_postgres contributors <https://github.com/ash-project/ash_postgres/graphs/contributors>
-
-SPDX-License-Identifier: MIT
diff --git a/test_priv/resource_snapshots/test_repo/tier/20260824043712.json.license b/test_priv/resource_snapshots/test_repo/tier/20260824043712.json.license
deleted file mode 100644
index c5bd9989..00000000
--- a/test_priv/resource_snapshots/test_repo/tier/20260824043712.json.license
+++ /dev/null
@@ -1,3 +0,0 @@
-SPDX-FileCopyrightText: 2019 ash_postgres contributors <https://github.com/ash-project/ash_postgres/graphs/contributors>
-
-SPDX-License-Identifier: MIT
diff --git a/test_priv/temporal_test_repo/migrations/20261004073133_migrate_resources_extensions_1.exs b/test_priv/temporal_test_repo/migrations/20261004073133_migrate_resources_extensions_1.exs
new file mode 100644
index 00000000..ecdbae3d
--- /dev/null
+++ b/test_priv/temporal_test_repo/migrations/20261004073133_migrate_resources_extensions_1.exs
@@ -0,0 +1,136 @@
+# SPDX-FileCopyrightText: 2019 ash_postgres contributors <https://github.com/ash-project/ash_postgres/graphs/contributors>
+#
+# SPDX-License-Identifier: MIT
+
+defmodule AshPostgres.TemporalTestRepo.Migrations.MigrateResourcesExtensions1 do
+  @moduledoc """
+  Installs any extensions that are mentioned in the repo's `installed_extensions/0` callback
+
+  This file was autogenerated with `mix ash_postgres.generate_migrations`
+  """
+
+  use Ecto.Migration
+
+  def up do
+    execute("""
+    CREATE OR REPLACE FUNCTION ash_elixir_or(left BOOLEAN, in right ANYCOMPATIBLE, out f1 ANYCOMPATIBLE)
+    AS $$ SELECT COALESCE(NULLIF($1, FALSE), $2) $$
+    LANGUAGE SQL
+    SET search_path = ''
+    IMMUTABLE;
+    """)
+
+    execute("""
+    CREATE OR REPLACE FUNCTION ash_elixir_or(left ANYCOMPATIBLE, in right ANYCOMPATIBLE, out f1 ANYCOMPATIBLE)
+    AS $$ SELECT COALESCE($1, $2) $$
+    LANGUAGE SQL
+    SET search_path = ''
+    IMMUTABLE;
+    """)
+
+    execute("""
+    CREATE OR REPLACE FUNCTION ash_elixir_and(left BOOLEAN, in right ANYCOMPATIBLE, out f1 ANYCOMPATIBLE) AS $$
+      SELECT CASE
+        WHEN $1 IS TRUE THEN $2
+        ELSE $1
+      END $$
+    LANGUAGE SQL
+    SET search_path = ''
+    IMMUTABLE;
+    """)
+
+    execute("""
+    CREATE OR REPLACE FUNCTION ash_elixir_and(left ANYCOMPATIBLE, in right ANYCOMPATIBLE, out f1 ANYCOMPATIBLE) AS $$
+      SELECT CASE
+        WHEN $1 IS NOT NULL THEN $2
+        ELSE $1
+      END $$
+    LANGUAGE SQL
+    SET search_path = ''
+    IMMUTABLE;
+    """)
+
+    execute("""
+    CREATE OR REPLACE FUNCTION ash_trim_whitespace(arr text[])
+    RETURNS text[] AS $$
+    DECLARE
+        start_index INT = 1;
+        end_index INT = array_length(arr, 1);
+    BEGIN
+        WHILE start_index <= end_index AND arr[start_index] = '' LOOP
+            start_index := start_index + 1;
+        END LOOP;
+
+        WHILE end_index >= start_index AND arr[end_index] = '' LOOP
+            end_index := end_index - 1;
+        END LOOP;
+
+        IF start_index > end_index THEN
+            RETURN ARRAY[]::text[];
+        ELSE
+            RETURN arr[start_index : end_index];
+        END IF;
+    END; $$
+    LANGUAGE plpgsql
+    SET search_path = ''
+    IMMUTABLE;
+    """)
+
+    execute("""
+    CREATE OR REPLACE FUNCTION ash_raise_error(json_data jsonb)
+    RETURNS BOOLEAN AS $$
+    BEGIN
+        -- Raise an error with the provided JSON data.
+        -- The JSON object is converted to text for inclusion in the error message.
+        RAISE EXCEPTION 'ash_error: %', json_data::text;
+        RETURN NULL;
+    END;
+    $$ LANGUAGE plpgsql
+    STABLE
+    SET search_path = '';
+    """)
+
+    execute("""
+    CREATE OR REPLACE FUNCTION ash_raise_error(json_data jsonb, type_signal ANYCOMPATIBLE)
+    RETURNS ANYCOMPATIBLE AS $$
+    BEGIN
+        -- Raise an error with the provided JSON data.
+        -- The JSON object is converted to text for inclusion in the error message.
+        RAISE EXCEPTION 'ash_error: %', json_data::text;
+        RETURN NULL;
+    END;
+    $$ LANGUAGE plpgsql
+    STABLE
+    SET search_path = '';
+    """)
+
+    execute("""
+    CREATE OR REPLACE FUNCTION ash_required(value ANYCOMPATIBLE, payload jsonb)
+    RETURNS ANYCOMPATIBLE AS $$
+    BEGIN
+      IF value IS NULL THEN
+        RAISE EXCEPTION 'ash_error: %', payload::text;
+      END IF;
+
+      RETURN value;
+    END;
+    $$ LANGUAGE plpgsql
+    STABLE
+    SET search_path = '';
+    """)
+
+    execute("CREATE EXTENSION IF NOT EXISTS \"citext\"")
+    execute("CREATE EXTENSION IF NOT EXISTS \"btree_gist\"")
+  end
+
+  def down do
+    # Uncomment this if you actually want to uninstall the extensions
+    # when this migration is rolled back:
+    execute(
+      "DROP FUNCTION IF EXISTS uuid_generate_v7(), ash_raise_error(jsonb), ash_raise_error(jsonb, ANYCOMPATIBLE), ash_elixir_and(BOOLEAN, ANYCOMPATIBLE), ash_elixir_and(ANYCOMPATIBLE, ANYCOMPATIBLE), ash_elixir_or(ANYCOMPATIBLE, ANYCOMPATIBLE), ash_elixir_or(BOOLEAN, ANYCOMPATIBLE), ash_trim_whitespace(text[]), ash_required(ANYCOMPATIBLE, jsonb)"
+    )
+
+    # execute("DROP EXTENSION IF EXISTS \"citext\"")
+    # execute("DROP EXTENSION IF EXISTS \"btree_gist\"")
+  end
+end
diff --git a/test_priv/test_repo/migrations/20260824043710_migrate_resources78.exs b/test_priv/temporal_test_repo/migrations/20261004073134_migrate_resources1.exs
similarity index 51%
rename from test_priv/test_repo/migrations/20260824043710_migrate_resources78.exs
rename to test_priv/temporal_test_repo/migrations/20261004073134_migrate_resources1.exs
index 3c1ee750..c5704ce0 100644
--- a/test_priv/test_repo/migrations/20260824043710_migrate_resources78.exs
+++ b/test_priv/temporal_test_repo/migrations/20261004073134_migrate_resources1.exs
@@ -2,7 +2,7 @@
 #
 # SPDX-License-Identifier: MIT
 
-defmodule AshPostgres.TestRepo.Migrations.MigrateResources78 do
+defmodule AshPostgres.TemporalTestRepo.Migrations.MigrateResources1 do
   @moduledoc """
   Updates resources based on their most recent snapshots.
 
@@ -18,18 +18,17 @@ defmodule AshPostgres.TestRepo.Migrations.MigrateResources78 do
       add(:created, :utc_datetime_usec)
     end
 
-    create table(:tier, primary_key: false) do
+    create table(:member, primary_key: false) do
       add(:id, :bigint, null: false)
-      add(:name, :text)
+      add(:email, :citext)
       add(:valid_at, :tstzrange, null: false)
     end
 
-    if repo().query!("SHOW server_version_num").rows |> hd() |> hd() |> String.to_integer() >=
-         190_000 do
-      execute("ALTER TABLE \"tier\" ADD PRIMARY KEY (id, valid_at WITHOUT OVERLAPS)")
-    else
-      execute("ALTER TABLE \"tier\" ADD PRIMARY KEY (id)")
-    end
+    execute("ALTER TABLE \"member\" ADD PRIMARY KEY (id, valid_at WITHOUT OVERLAPS)")
+
+    execute(
+      "ALTER TABLE \"member\" ADD CONSTRAINT \"member_unique_email_index\" EXCLUDE USING gist ((lower(email::text)) WITH =, valid_at WITH &&)"
+    )
 
     create table(:subscription, primary_key: false) do
       add(:id, :bigint, null: false)
@@ -40,27 +39,42 @@ defmodule AshPostgres.TestRepo.Migrations.MigrateResources78 do
       add(:valid_at, :tstzrange, null: false)
     end
 
-    if repo().query!("SHOW server_version_num").rows |> hd() |> hd() |> String.to_integer() >=
-         190_000 do
-      execute("ALTER TABLE \"subscription\" ADD PRIMARY KEY (id, valid_at WITHOUT OVERLAPS)")
-    else
-      execute("ALTER TABLE \"subscription\" ADD PRIMARY KEY (id)")
-    end
+    execute("ALTER TABLE \"subscription\" ADD PRIMARY KEY (id, valid_at WITHOUT OVERLAPS)")
 
-    if repo().query!("SHOW server_version_num").rows |> hd() |> hd() |> String.to_integer() >=
-         190_000 do
-      execute(
-        "ALTER TABLE \"subscription\" ADD CONSTRAINT subscription_tier_id_fkey FOREIGN KEY (tier_id, PERIOD valid_at) REFERENCES \"public\".\"tier\" (id, PERIOD valid_at)"
+    alter table(:subscription) do
+      add(:recorded_at, :utc_datetime_usec,
+        null: false,
+        default: fragment("(now() AT TIME ZONE 'utc')")
       )
     end
+
+    create table(:tier, primary_key: false) do
+      add(:id, :bigint, null: false)
+      add(:name, :text)
+      add(:valid_at, :tstzrange, null: false)
+    end
+
+    execute("ALTER TABLE \"tier\" ADD PRIMARY KEY (id, valid_at WITHOUT OVERLAPS)")
+
+    execute(
+      "ALTER TABLE \"subscription\" ADD CONSTRAINT subscription_tier_id_fkey FOREIGN KEY (tier_id, PERIOD valid_at) REFERENCES \"public\".\"tier\" (id, PERIOD valid_at)"
+    )
   end
 
   def down do
     execute("ALTER TABLE \"subscription\" DROP CONSTRAINT subscription_tier_id_fkey")
 
+    drop(table(:tier))
+
+    alter table(:subscription) do
+      remove(:recorded_at)
+    end
+
     drop(table(:subscription))
 
-    drop(table(:tier))
+    execute("ALTER TABLE \"member\" DROP CONSTRAINT IF EXISTS \"member_unique_email_index\"")
+
+    drop(table(:member))
 
     drop(table(:event))
   end
diff --git a/test_priv/test_repo/migrations/20260926192338_migrate_resources76.exs b/test_priv/test_repo/migrations/20260926192338_migrate_resources76.exs
deleted file mode 100644
index 3987daf2..00000000
--- a/test_priv/test_repo/migrations/20260926192338_migrate_resources76.exs
+++ /dev/null
@@ -1,48 +0,0 @@
-# SPDX-FileCopyrightText: 2019 ash_postgres contributors <https://github.com/ash-project/ash_postgres/graphs/contributors>
-#
-# SPDX-License-Identifier: MIT
-
-defmodule AshPostgres.TestRepo.Migrations.MigrateResources76 do
-  @moduledoc """
-  Updates resources based on their most recent snapshots.
-
-  This file was autogenerated with `mix ash_postgres.generate_migrations`
-  """
-
-  use Ecto.Migration
-
-  def up do
-    create table(:member, primary_key: false) do
-      add(:id, :bigint, null: false)
-      add(:email, :citext)
-      add(:valid_at, :tstzrange, null: false)
-    end
-
-    if repo().query!("SHOW server_version_num").rows |> hd() |> hd() |> String.to_integer() >=
-         190_000 do
-      execute("ALTER TABLE \"member\" ADD PRIMARY KEY (id, valid_at WITHOUT OVERLAPS)")
-    else
-      execute("ALTER TABLE \"member\" ADD PRIMARY KEY (id)")
-    end
-
-    if repo().query!("SHOW server_version_num").rows |> hd() |> hd() |> String.to_integer() >=
-         190_000 do
-      execute(
-        "ALTER TABLE \"member\" ADD CONSTRAINT \"member_unique_email_index\" EXCLUDE USING gist ((lower(email::text)) WITH =, valid_at WITH &&)"
-      )
-    else
-      create(unique_index(:member, [:email], name: "member_unique_email_index"))
-    end
-  end
-
-  def down do
-    if repo().query!("SHOW server_version_num").rows |> hd() |> hd() |> String.to_integer() >=
-         190_000 do
-      execute("ALTER TABLE \"member\" DROP CONSTRAINT IF EXISTS \"member_unique_email_index\"")
-    else
-      drop_if_exists(unique_index(:member, [:email], name: "member_unique_email_index"))
-    end
-
-    drop(table(:member))
-  end
-end
diff --git a/test_priv/test_repo/migrations/20260927024129_migrate_resources77.exs b/test_priv/test_repo/migrations/20260927024129_migrate_resources77.exs
deleted file mode 100644
index 06892aae..00000000
--- a/test_priv/test_repo/migrations/20260927024129_migrate_resources77.exs
+++ /dev/null
@@ -1,28 +0,0 @@
-# SPDX-FileCopyrightText: 2019 ash_postgres contributors <https://github.com/ash-project/ash_postgres/graphs/contributors>
-#
-# SPDX-License-Identifier: MIT
-
-defmodule AshPostgres.TestRepo.Migrations.MigrateResources77 do
-  @moduledoc """
-  Updates resources based on their most recent snapshots.
-
-  This file was autogenerated with `mix ash_postgres.generate_migrations`
-  """
-
-  use Ecto.Migration
-
-  def up do
-    alter table(:subscription) do
-      add(:recorded_at, :utc_datetime_usec,
-        null: false,
-        default: fragment("(now() AT TIME ZONE 'utc')")
-      )
-    end
-  end
-
-  def down do
-    alter table(:subscription) do
-      remove(:recorded_at)
-    end
-  end
-end