diff --git a/.rubocop.yml b/.rubocop.yml
index 6458bc528..42d7fe5c2 100644
--- a/.rubocop.yml
+++ b/.rubocop.yml
@@ -257,6 +257,8 @@ Naming/PredicatePrefix:
 
 Metrics/ParameterLists:
   Exclude:
+    - 'packages/forest_admin_agent/lib/forest_admin_agent/audit_trail/capture.rb'
+    - 'packages/forest_admin_agent/lib/forest_admin_agent/audit_trail/store.rb'
     - 'packages/forest_admin_datasource_graphql_hasura/lib/forest_admin_datasource_graphql_hasura/configuration.rb'
     - 'packages/forest_admin_datasource_zendesk/lib/forest_admin_datasource_zendesk/collections/base_collection.rb'
     - 'packages/forest_admin_datasource_snowflake/lib/forest_admin_datasource_snowflake/datasource.rb'
@@ -454,6 +456,9 @@ Layout/LineLength:
 
 RSpec/VerifiedDoubles:
   Exclude:
+    - 'packages/forest_admin_agent/spec/lib/forest_admin_agent/audit_trail/capture_spec.rb'
+    - 'packages/forest_admin_agent/spec/lib/forest_admin_agent/routes/resources/audit_trail_spec.rb'
+    - 'packages/forest_admin_agent/spec/lib/forest_admin_agent/routes/resources/audit_trail_correlation_spec.rb'
     - 'packages/forest_admin_datasource_customizer/spec/lib/forest_admin_datasource_customizer/composite_datasource_spec.rb'
 
 RSpec/VerifiedDoubleReference:
diff --git a/packages/forest_admin_agent/AUDIT_TRAIL.md b/packages/forest_admin_agent/AUDIT_TRAIL.md
new file mode 100644
index 000000000..fdb47183f
--- /dev/null
+++ b/packages/forest_admin_agent/AUDIT_TRAIL.md
@@ -0,0 +1,168 @@
+# Audit trail
+
+Capture who changed what (before/after) for every change Forest performs through its data layer, and
+persist it into a SQL database. Built into the agent: it turns on as soon as an audit-trail **database
+is configured**, and stays completely off otherwise.
+
+Two parts, both internal:
+
+- **Capture** (`ForestAdminAgent::AuditTrail::Capture`) — datasource-agnostic. It instruments every
+  collection through the customizer hooks, so it behaves the same whether the audited datasource is
+  ActiveRecord, Mongoid, etc.
+- **Storage** (`ForestAdminAgent::AuditTrail::Store`) — ActiveRecord-backed. It creates the `forest`
+  schema and creates/evolves the `audit_logs` table through versioned migrations, and reads the
+  per-record history back for the routes below.
+
+Storage uses ActiveRecord: outside Rails, add `gem 'activerecord'` (and the adapter gem) to your
+Gemfile. Nothing is loaded and no connection is opened until the feature is configured.
+
+## Turn it on
+
+### Rails (forest_admin_rails)
+
+```ruby
+# config/initializers/forest_admin_rails.rb
+ForestAdminRails.configure do |config|
+  config.auth_secret = ENV['FOREST_AUTH_SECRET']
+  config.env_secret  = ENV['FOREST_ENV_SECRET']
+
+  config.audit_trail = {
+    database: { # or an ActiveRecord URL: ENV['AUDIT_TRAIL_DATABASE_URL']
+      adapter: 'postgresql', host: ENV['AUDIT_DB_HOST'], port: ENV['AUDIT_DB_PORT'],
+      username: ENV['AUDIT_DB_USER'], password: ENV['AUDIT_DB_PASSWORD'], database: ENV['AUDIT_DB_NAME']
+    }
+  }
+end
+```
+
+### Plain agent (no Rails)
+
+```ruby
+ForestAdminAgent::Builder::AgentFactory.instance.setup(
+  auth_secret: ENV['FOREST_AUTH_SECRET'],
+  env_secret: ENV['FOREST_ENV_SECRET'],
+  # ...usual options...
+  audit_trail: { database: ENV['AUDIT_TRAIL_DATABASE_URL'] }
+)
+```
+
+| option       | description                                                                          |
+| ------------ | ------------------------------------------------------------------------------------ |
+| `database`   | ActiveRecord URL or config hash. **Setting it activates the audit trail.**           |
+| `schema`     | Postgres schema holding the table (default `forest`; ignored on other adapters)      |
+| `table_name` | default `audit_logs`                                                                 |
+| `redact`     | `{ 'collection_name' => ['field', ...] }` — values masked while recording the change |
+
+On the first write or read the store ensures the schema exists and runs any pending migrations; every
+create / update / delete performed through Forest then writes one row per record, and the **Historic**
+tab in the UI reads from the same table.
+
+## Routes
+
+All routes live under `/forest/_audit-trail`, are registered only when `audit_trail[:database]` is
+set, and require read permission on the target collection (`can?(:read, collection)`).
+
+### Record-history route
+
+`GET /forest/_audit-trail/{collection}/{recordId}` returns the current page of history (newest first
+by default) together with the filtered total:
+
+```json
+{ "data": [ /* current page rows */ ], "meta": { "count": 137 } }
+```
+
+`meta.count` is the number of rows matching the active filters (not the absolute total) and is
+independent of the page. Optional filters (all combine with `AND`; omit them for the full history):
+
+| query param | format                           | effect                                          |
+| ----------- | -------------------------------- | ----------------------------------------------- |
+| `userIds`   | comma-separated integers `12,45` | keep only entries whose `user_id` is in the list |
+| `startDate` | `YYYY-MM-DD` or datetime (incl.) | keep entries from this lower bound onward       |
+| `endDate`   | `YYYY-MM-DD` or datetime (incl.) | keep entries up to this upper bound             |
+
+`startDate` / `endDate` are read as **local wall-clock time** in the request `timezone` query param
+(e.g. `Europe/Paris`, default `UTC`) and converted to a UTC instant before querying, so filtering
+happens in SQL. Two shapes are accepted:
+
+- **Bare day** `YYYY-MM-DD` — `startDate` snaps to `00:00:00.000`, `endDate` to `23:59:59.999`.
+- **Datetime** `YYYY-MM-DD[T| ]HH:mm[:ss]` — `T` or space separator, seconds optional; when seconds
+  are omitted `endDate` is completed to `:59.999` and `startDate` stays at `:00.000`.
+
+Both bounds are **inclusive**. Defensive parsing: non-numeric `userIds` tokens are dropped
+(`12,abc,45` → `12,45`), and a `startDate` / `endDate` matching no accepted format returns **HTTP
+400** (`ValidationError`); an invalid `timezone` likewise returns **400**.
+
+Pagination follows JSON:API: `page[number]` is 1-based (default `1`), `page[size]` defaults to `20`
+and is capped at `100`; out-of-bound or non-numeric values fall back to the defaults rather than
+erroring. Sorting follows JSON:API `sort` on `timestamp`: `sort=-timestamp` (or absent/unrecognized)
+is newest first, `sort=timestamp` is oldest first. Ties on equal timestamps fall back to insertion
+order (the auto-increment `id`), so paging is deterministic in either direction.
+
+All three routes serialize audit records the same way: top-level keys are camelCased
+(`recordId`, `userId`, `correlationKey`, `previousValues`, `new_values` → `newValues`), while the
+`previousValues` / `newValues` hashes keep the audited record's own column names.
+
+A record that no longer exists keeps its history: only a record that still exists *outside* the
+caller's permission scope is refused (404). Inspecting what was deleted is much of the point of an
+audit trail, and the delete event itself is the last thing recorded.
+
+### Correlation route
+
+`GET /forest/_audit-trail/correlation/{correlationKey}` returns `{ "data": [...] }` — the
+operation(s) recorded under one `correlation_key` for a single record (usually one), oldest first, or
+an empty array if none. Scoped through query params; same auth and gating as above.
+
+| query param  | required | effect                                                       |
+| ------------ | -------- | ------------------------------------------------------------ |
+| `collection` | yes      | collection the record belongs to (also the permission scope) |
+| `recordId`   | yes      | packed record id to scope the lookup                         |
+
+A missing `collection` or `recordId` returns **HTTP 400** (`ValidationError`).
+
+### Batch correlation route
+
+`GET /forest/_audit-trail/correlations` returns `{ "data": [...] }` — a **flat** list of every record
+whose `correlation_key` is in `correlationKeys`, scoped to one record (the client groups by
+`correlation_key`). Same auth and gating; empty array when nothing matches.
+
+| query param       | required | effect                                                       |
+| ----------------- | -------- | ------------------------------------------------------------ |
+| `correlationKeys` | yes\*    | comma-separated keys; blank tokens are dropped               |
+| `collection`      | yes      | collection the record belongs to (also the permission scope) |
+| `recordId`        | yes      | packed record id to scope the lookup                         |
+
+\* To dodge any URL length limit, the same path also accepts **`POST`** with a JSON body
+`{ "correlationKeys": [...], "collection": "...", "recordId": "..." }` (the body array takes
+precedence over the query param). An empty/absent key list returns `{ "data": [] }` without hitting
+the store. A missing `collection` or `recordId` returns **HTTP 400** (`ValidationError`).
+
+## What gets stored
+
+`forest.audit_logs`, one row per audited change:
+
+| column            | description                                                 |
+| ----------------- | ----------------------------------------------------------- |
+| `id`              | auto-increment primary key                                  |
+| `timestamp`       | when the change happened                                    |
+| `operation`       | `create` / `update` / `delete`                              |
+| `collection`      | audited collection name                                     |
+| `record_id`       | packed record id (primary keys joined by `\|`)              |
+| `user_id`         | the Forest user who made the change                         |
+| `correlation_key` | per-request id; groups every change made within one request |
+| `previous_values` | values before the change (JSON)                             |
+| `new_values`      | values after the change (JSON)                              |
+
+`previous_values` / `new_values` store **only the parts that actually changed**: nested hashes and
+arrays of hashes are diffed structurally, so a single sub-field change records just that leaf. Only
+writable columns are audited — read-only, computed and DB-managed fields are never written by Forest.
+
+The `correlation_key` is the agent's per-request id (`caller.request_id`), generated by the agent and
+echoed back to the client in the `X-Forest-Correlation-Id` response header — so every change made in
+one request shares a key, and the caller can tie it to its own activity log.
+
+## Schema migrations & concurrency
+
+The table is created/evolved through an ordered, append-only migration list tracked in a dedicated
+`forest.audit_migrations` table. On Postgres the migrations run inside a transaction-scoped advisory
+lock so several agents booting at once apply them one after another; the schema is created (and
+committed, idempotently) first since the lock can't cover a not-yet-existing schema.
diff --git a/packages/forest_admin_agent/Gemfile b/packages/forest_admin_agent/Gemfile
index 0e0b74330..d22fe37a2 100644
--- a/packages/forest_admin_agent/Gemfile
+++ b/packages/forest_admin_agent/Gemfile
@@ -3,6 +3,7 @@ source "https://rubygems.org"
 gemspec
 
 group :development, :test do
+  gem 'activerecord', '>= 6.1'
   gem 'forest_admin_datasource_customizer', path: '../forest_admin_datasource_customizer'
   gem 'forest_admin_datasource_toolkit', path: '../forest_admin_datasource_toolkit'
   gem 'forest_admin_test_toolkit', path: '../forest_admin_test_toolkit'
@@ -13,4 +14,5 @@ group :development, :test do
   gem 'simplecov', '~> 0.22', require: false
   gem 'simplecov-html', '~> 0.12.3'
   gem 'simplecov_json_formatter', '~> 0.1.4'
+  gem 'sqlite3', '>= 2.1'
 end
diff --git a/packages/forest_admin_agent/Gemfile-test b/packages/forest_admin_agent/Gemfile-test
index 0e0b74330..d22fe37a2 100644
--- a/packages/forest_admin_agent/Gemfile-test
+++ b/packages/forest_admin_agent/Gemfile-test
@@ -3,6 +3,7 @@ source "https://rubygems.org"
 gemspec
 
 group :development, :test do
+  gem 'activerecord', '>= 6.1'
   gem 'forest_admin_datasource_customizer', path: '../forest_admin_datasource_customizer'
   gem 'forest_admin_datasource_toolkit', path: '../forest_admin_datasource_toolkit'
   gem 'forest_admin_test_toolkit', path: '../forest_admin_test_toolkit'
@@ -13,4 +14,5 @@ group :development, :test do
   gem 'simplecov', '~> 0.22', require: false
   gem 'simplecov-html', '~> 0.12.3'
   gem 'simplecov_json_formatter', '~> 0.1.4'
+  gem 'sqlite3', '>= 2.1'
 end
diff --git a/packages/forest_admin_agent/lib/forest_admin_agent.rb b/packages/forest_admin_agent/lib/forest_admin_agent.rb
index be8270393..22d65f2b9 100644
--- a/packages/forest_admin_agent/lib/forest_admin_agent.rb
+++ b/packages/forest_admin_agent/lib/forest_admin_agent.rb
@@ -4,7 +4,11 @@
 
 loader = Zeitwerk::Loader.for_gem
 loader.inflector.inflect('oauth2' => 'OAuth2')
+loader.inflector.inflect('sql' => 'Sql')
 loader.inflector.inflect('sse_cache_invalidation' => 'SSECacheInvalidation')
+# ActiveRecord is only needed by agents configuring an audit-trail database, and Rails eager loads
+# every gem loader (Zeitwerk::Loader.eager_load_all), so these files must stay strictly autoloaded.
+loader.do_not_eager_load("#{__dir__}/forest_admin_agent/audit_trail/sql")
 loader.setup
 
 module ForestAdminAgent
diff --git a/packages/forest_admin_agent/lib/forest_admin_agent/audit_trail/audit_record.rb b/packages/forest_admin_agent/lib/forest_admin_agent/audit_trail/audit_record.rb
new file mode 100644
index 000000000..011d4238a
--- /dev/null
+++ b/packages/forest_admin_agent/lib/forest_admin_agent/audit_trail/audit_record.rb
@@ -0,0 +1,11 @@
+module ForestAdminAgent
+  module AuditTrail
+    # One audited change. Mirrors the columns of `forest.audit_logs`; only the actor's `user_id` is
+    # stored, the rest of the actor identity is correlated elsewhere through `correlation_key`.
+    AuditRecord = Struct.new(
+      :timestamp, :operation, :collection, :record_id, :user_id, :correlation_key,
+      :previous_values, :new_values,
+      keyword_init: true
+    )
+  end
+end
diff --git a/packages/forest_admin_agent/lib/forest_admin_agent/audit_trail/capture.rb b/packages/forest_admin_agent/lib/forest_admin_agent/audit_trail/capture.rb
new file mode 100644
index 000000000..deb970e5f
--- /dev/null
+++ b/packages/forest_admin_agent/lib/forest_admin_agent/audit_trail/capture.rb
@@ -0,0 +1,146 @@
+require 'securerandom'
+require 'time'
+
+module ForestAdminAgent
+  module AuditTrail
+    # Datasource-agnostic capture layer, installed by the agent factory as soon as an audit-trail
+    # database is configured. It instruments every collection through the Forest customizer hooks, so
+    # it behaves the same whatever the audited datasource is (ActiveRecord, Mongoid, ...), computes the
+    # minimal before/after diff for each change and appends an {AuditRecord} to the store.
+    class Capture
+      REDACTED = '[redacted]'.freeze
+      # ponytail: 16 deep is far past any legitimate nesting; raise it if one ever gets that far.
+      MAX_SNAPSHOTS = 16
+
+      # Signature imposed by DatasourceCustomizer#use; the agent always instruments the whole datasource.
+      def run(datasource_customizer, _collection_customizer = nil, options = {})
+        @store = options[:store]
+        @redact = options[:redact] || {}
+
+        datasource_customizer.collections.each_value { |collection| instrument(collection) }
+      end
+
+      private
+
+      def instrument(collection_customizer)
+        schema = collection_customizer.collection.schema
+        # Writable columns only: Forest audits what it writes. Read-only fields cover computed/virtual
+        # fields and DB-managed columns, none of which Forest mutates.
+        columns = schema[:fields].select do |_name, field|
+          field.type == 'Column' && !field.is_read_only
+        end.keys
+        primary_keys = ForestAdminDatasourceToolkit::Utils::Schema.primary_keys(collection_customizer.collection)
+        # Reads must carry the primary keys (even read-only ones) so the record id can be built; the
+        # diff itself stays restricted to the writable columns.
+        projection = ForestAdminDatasourceToolkit::Components::Query::Projection.new(
+          (primary_keys + columns).uniq
+        )
+        name = collection_customizer.name
+
+        add_create_hook(collection_customizer, columns, primary_keys, name)
+        add_update_hooks(collection_customizer, columns, primary_keys, name, projection)
+        add_delete_hooks(collection_customizer, columns, primary_keys, name, projection)
+      end
+
+      def add_create_hook(collection_customizer, columns, primary_keys, name)
+        collection_customizer.add_hook('After', 'Create') do |context|
+          emit(
+            context.caller, 'create', name, record_id(context.record, primary_keys),
+            {}, pick(context.record, columns)
+          )
+        end
+      end
+
+      def add_update_hooks(collection_customizer, columns, primary_keys, name, projection)
+        collection_customizer.add_hook('Before', 'Update') do |context|
+          # Snapshot the patch here: installed last, this hook sees what actually gets written, while
+          # the after-context is handed the original patch the caller sent.
+          push_snapshot(records: context.collection.list(context.filter, projection), patch: context.patch)
+        end
+
+        collection_customizer.add_hook('After', 'Update') do |context|
+          snapshot = snapshots.pop
+
+          snapshot&.fetch(:records)&.each do |record|
+            delta = Diff.changed_values(record, snapshot[:patch], columns)
+            next if delta[:new_values].empty?
+
+            emit(
+              context.caller, 'update', name, record_id(record, primary_keys),
+              delta[:previous_values], delta[:new_values]
+            )
+          end
+        end
+      end
+
+      def add_delete_hooks(collection_customizer, columns, primary_keys, name, projection)
+        collection_customizer.add_hook('Before', 'Delete') do |context|
+          push_snapshot(records: context.collection.list(context.filter, projection))
+        end
+
+        collection_customizer.add_hook('After', 'Delete') do |context|
+          snapshot = snapshots.pop
+
+          snapshot&.fetch(:records)&.each do |record|
+            emit(
+              context.caller, 'delete', name, record_id(record, primary_keys),
+              pick(record, columns), {}
+            )
+          end
+        end
+      end
+
+      # Snapshots taken in a "before" hook and consumed in the matching "after" hook. Both bracket one
+      # operation on one thread, so a LIFO stack pairs them without relying on the filter object
+      # reaching both hooks unchanged. An operation raising in between strands its entry, hence the cap.
+      def push_snapshot(snapshot)
+        stack = snapshots
+        stack.shift while stack.size >= MAX_SNAPSHOTS
+        stack.push(snapshot)
+      end
+
+      def snapshots
+        Thread.current[:forest_audit_trail_snapshots] ||= []
+      end
+
+      def emit(caller, operation, collection, record_id, previous_values, new_values)
+        redacted = @redact[collection] || []
+
+        @store.append(
+          AuditRecord.new(
+            timestamp: Time.now.utc.iso8601(3),
+            operation: operation,
+            collection: collection,
+            record_id: record_id,
+            user_id: caller&.id,
+            # Same id for every change made within one request — set on the caller by the agent
+            # (see CallerParser), mirroring the Node agent's caller.requestId.
+            correlation_key: correlation_key_for(caller),
+            previous_values: redact(previous_values, redacted),
+            new_values: redact(new_values, redacted)
+          )
+        )
+      end
+
+      def correlation_key_for(caller)
+        (caller.respond_to?(:request_id) && caller.request_id) || SecureRandom.uuid
+      end
+
+      def record_id(record, primary_keys)
+        primary_keys.map { |pk| record[pk].to_s }.join('|')
+      end
+
+      def pick(record, columns)
+        columns.to_h { |column| [column, record[column]] }
+      end
+
+      def redact(values, redacted_fields)
+        return values if redacted_fields.empty?
+
+        values.each_with_object({}) do |(field, value), result|
+          result[field] = redacted_fields.include?(field) ? REDACTED : value
+        end
+      end
+    end
+  end
+end
diff --git a/packages/forest_admin_agent/lib/forest_admin_agent/audit_trail/diff.rb b/packages/forest_admin_agent/lib/forest_admin_agent/audit_trail/diff.rb
new file mode 100644
index 000000000..8d8ed7a1a
--- /dev/null
+++ b/packages/forest_admin_agent/lib/forest_admin_agent/audit_trail/diff.rb
@@ -0,0 +1,91 @@
+module ForestAdminAgent
+  module AuditTrail
+    # Minimal structural diff. Nested hashes and arrays of hashes are recursed into, so only the
+    # keys/indexes whose leaf value actually changed are kept — a single sub-field change does not
+    # store the whole object/array. Scalars, primitive arrays, dates and other values are compared and
+    # kept as a whole.
+    #
+    # Ruby's `==` already performs deep, key-order-independent equality on Hash and (ordered) equality
+    # on Array, so it is used directly as the equality primitive.
+    module Diff
+      module_function
+
+      # @return [Hash{Symbol=>Object}, nil] { previous:, next: } of the changed leaves, or nil when equal.
+      def diff(before, after)
+        return nil if before == after
+
+        return diff_hashes(before, after) if before.is_a?(Hash) && after.is_a?(Hash)
+
+        return diff_object_arrays(before, after) if object_array?(before) && object_array?(after)
+
+        { previous: before.nil? ? nil : before, next: after.nil? ? nil : after }
+      end
+
+      # Build the previous/new value hashes for the writable columns that actually changed.
+      #
+      # @param before [Hash] snapshot of the record before the change (string keys)
+      # @param patch [Hash] the values being written (string keys); only present keys are considered
+      # @param columns [Array<String>] writable column names to inspect
+      # @return [Hash{Symbol=>Hash}] { previous_values:, new_values: }
+      def changed_values(before, patch, columns)
+        previous_values = {}
+        new_values = {}
+
+        columns.each do |column|
+          delta = patch.key?(column) ? diff(before[column], patch[column]) : nil
+          next unless delta
+
+          previous_values[column] = delta[:previous]
+          new_values[column] = delta[:next]
+        end
+
+        { previous_values: previous_values, new_values: new_values }
+      end
+
+      # Arrays whose every element is a hash (record-like collections, e.g. a workflow history).
+      def object_array?(value)
+        value.is_a?(Array) && !value.empty? && value.all?(Hash)
+      end
+
+      def diff_hashes(before, after)
+        previous = {}
+        next_values = {}
+
+        (before.keys | after.keys).each do |key|
+          sub = diff_at(before, after, key)
+          next unless sub
+
+          previous[key] = sub[:previous]
+          next_values[key] = sub[:next]
+        end
+
+        { previous: previous, next: next_values }
+      end
+
+      # A key held with a nil value is not the same thing as a missing key: recursing on the values
+      # alone reads both as nil and reports no change at all.
+      def diff_at(before, after, key)
+        return diff(before[key], after[key]) if before.key?(key) == after.key?(key)
+
+        { previous: before[key], next: after[key] }
+      end
+
+      def diff_object_arrays(before, after)
+        previous = {}
+        next_values = {}
+
+        [before.length, after.length].max.times do |index|
+          sub = diff(before[index], after[index])
+          next unless sub
+
+          previous[index] = sub[:previous]
+          next_values[index] = sub[:next]
+        end
+
+        { previous: previous, next: next_values }
+      end
+
+      private_class_method :diff_hashes, :diff_at, :diff_object_arrays
+    end
+  end
+end
diff --git a/packages/forest_admin_agent/lib/forest_admin_agent/audit_trail/sql/audit_connection_base.rb b/packages/forest_admin_agent/lib/forest_admin_agent/audit_trail/sql/audit_connection_base.rb
new file mode 100644
index 000000000..9005f1c40
--- /dev/null
+++ b/packages/forest_admin_agent/lib/forest_admin_agent/audit_trail/sql/audit_connection_base.rb
@@ -0,0 +1,19 @@
+begin
+  require 'active_record'
+rescue LoadError
+  raise LoadError, 'config.audit_trail needs the activerecord gem: add `gem "activerecord"` to your Gemfile.'
+end
+
+module ForestAdminAgent
+  module AuditTrail
+    module Sql
+      # Dedicated abstract base so the audit storage keeps its own connection pool, isolated from the
+      # host application's ActiveRecord::Base connection. Also the level the `attribute` overrides in
+      # AuditLog need: declaring them straight on an ActiveRecord::Base child resolves the type
+      # eagerly and blows up before any connection is established.
+      class AuditConnectionBase < ActiveRecord::Base
+        self.abstract_class = true
+      end
+    end
+  end
+end
diff --git a/packages/forest_admin_agent/lib/forest_admin_agent/audit_trail/sql/audit_log.rb b/packages/forest_admin_agent/lib/forest_admin_agent/audit_trail/sql/audit_log.rb
new file mode 100644
index 000000000..a0eb7c442
--- /dev/null
+++ b/packages/forest_admin_agent/lib/forest_admin_agent/audit_trail/sql/audit_log.rb
@@ -0,0 +1,16 @@
+module ForestAdminAgent
+  module AuditTrail
+    module Sql
+      # Abstract template for the audit model. Each Store builds its own concrete subclass bound to its
+      # own (schema-qualified) table, so stores with different `table_name`/`schema` can't clobber a
+      # shared one. The JSON attribute overrides force Hash <-> JSON casting on every adapter (Postgres
+      # json, or text on SQLite) and are inherited by every subclass.
+      class AuditLog < AuditConnectionBase
+        self.abstract_class = true
+
+        attribute :previous_values, :json
+        attribute :new_values, :json
+      end
+    end
+  end
+end
diff --git a/packages/forest_admin_agent/lib/forest_admin_agent/audit_trail/sql/migrator.rb b/packages/forest_admin_agent/lib/forest_admin_agent/audit_trail/sql/migrator.rb
new file mode 100644
index 000000000..d05419afb
--- /dev/null
+++ b/packages/forest_admin_agent/lib/forest_admin_agent/audit_trail/sql/migrator.rb
@@ -0,0 +1,127 @@
+module ForestAdminAgent
+  module AuditTrail
+    module Sql
+      # Creates and evolves the audit table through an ordered, append-only list of migrations, tracked
+      # in a dedicated `audit_migrations` table (namespaced in the `forest` schema on Postgres).
+      #
+      # On Postgres the migrations run inside a transaction-scoped advisory lock, so several agent
+      # instances booting at once apply them one after another instead of racing on the same DDL. The
+      # schema is created (and committed) first, made idempotent (CREATE SCHEMA IF NOT EXISTS +
+      # tolerating a concurrent create), because the lock cannot cover a not-yet-existing schema.
+      class Migrator
+        MIGRATIONS_TABLE = 'audit_migrations'.freeze
+        # Arbitrary but stable key pair identifying the audit-trail migration critical section.
+        ADVISORY_LOCK = [0x464f, 0x5254].freeze # "FO", "RT"
+
+        MIGRATIONS = [
+          {
+            name: '001-create-audit-logs',
+            up: lambda do |connection, table|
+              # if_not_exists: a non-PG race (no advisory lock) can let two instances both reach here.
+              connection.create_table(table, if_not_exists: true) do |t|
+                t.datetime :timestamp, null: false
+                t.string :operation, null: false
+                t.string :collection, null: false
+                t.string :record_id, null: false
+                t.integer :user_id
+                t.string :correlation_key
+                t.json :previous_values
+                t.json :new_values
+              end
+            end
+          },
+          {
+            name: '002-index-record-and-correlation',
+            up: lambda do |connection, table|
+              base = table.split('.').last
+              connection.add_index(table, :record_id, name: "#{base}_record_id", if_not_exists: true)
+              connection.add_index(table, :correlation_key, name: "#{base}_correlation_key", if_not_exists: true)
+              connection.add_index(table, :user_id, name: "#{base}_user_id", if_not_exists: true)
+            end
+          }
+        ].freeze
+
+        def initialize(connection, schema:, table_name:)
+          @connection = connection
+          @schema = schema # nil on adapters without schema support
+          @table_name = table_name
+        end
+
+        def run
+          ensure_schema
+
+          if postgres?
+            @connection.transaction do
+              @connection.execute("SELECT pg_advisory_xact_lock(#{ADVISORY_LOCK[0]}, #{ADVISORY_LOCK[1]})")
+              apply_pending
+            end
+          else
+            apply_pending
+          end
+        end
+
+        private
+
+        def postgres?
+          @connection.adapter_name.downcase.include?('postgres')
+        end
+
+        def schema?
+          postgres? && @schema.present?
+        end
+
+        def qualified(name)
+          schema? ? "#{@schema}.#{name}" : name
+        end
+
+        # Create the schema first and commit it: the migrations open DDL on the same connection, and a
+        # CREATE SCHEMA still pending in the lock transaction would not be visible to them.
+        def ensure_schema
+          return unless schema?
+
+          @connection.execute("CREATE SCHEMA IF NOT EXISTS #{@connection.quote_schema_name(@schema)}")
+        rescue ActiveRecord::StatementInvalid => e
+          # Another instance created it concurrently — CREATE SCHEMA is not fully race-free.
+          raise unless /already exists|duplicate/i.match?(e.message)
+        end
+
+        def apply_pending
+          done = applied_migrations
+          table = qualified(@table_name)
+
+          MIGRATIONS.each do |migration|
+            # The tracking table is shared by every audit table of the database/schema, so the target
+            # table belongs in the key: a store configured with another `table_name` must still get its
+            # own table created instead of reading someone else's migration as done. Rows written by
+            # earlier versions (bare migration name) simply replay, which the `if_not_exists` DDL above
+            # makes a no-op.
+            key = "#{table}:#{migration[:name]}"
+            next if done.include?(key)
+
+            migration[:up].call(@connection, table)
+            @connection.execute(
+              "INSERT INTO #{@connection.quote_table_name(migrations_table)} (name) " \
+              "VALUES (#{@connection.quote(key)})"
+            )
+          end
+        end
+
+        def applied_migrations
+          ensure_migrations_table
+
+          @connection.select_values("SELECT name FROM #{@connection.quote_table_name(migrations_table)}")
+        end
+
+        def ensure_migrations_table
+          @connection.create_table(migrations_table, id: false, if_not_exists: true) do |t|
+            t.string :name, null: false
+          end
+        end
+
+        def migrations_table
+          qualified(MIGRATIONS_TABLE)
+        end
+      end
+    end
+  end
+end
diff --git a/packages/forest_admin_agent/lib/forest_admin_agent/audit_trail/store.rb b/packages/forest_admin_agent/lib/forest_admin_agent/audit_trail/store.rb
new file mode 100644
index 000000000..8c5d9b4e7
--- /dev/null
+++ b/packages/forest_admin_agent/lib/forest_admin_agent/audit_trail/store.rb
@@ -0,0 +1,133 @@
+require 'time'
+
+module ForestAdminAgent
+  module AuditTrail
+    # SQL-backed storage that both writes every audited change and reads the per-record history back.
+    #
+    # Construction is cheap; the connection is opened lazily on the first append or read, at which
+    # point the `forest` schema and the `audit_logs` table are created/evolved through migrations.
+    class Store
+      DEFAULT_SCHEMA = 'forest'.freeze
+      DEFAULT_TABLE = 'audit_logs'.freeze
+
+      def initialize(database:, schema: DEFAULT_SCHEMA, table_name: DEFAULT_TABLE)
+        @database = database
+        @schema = schema
+        @table_name = table_name
+        @mutex = Mutex.new
+        @ready = false
+      end
+
+      def append(record)
+        model.create!(to_row(record))
+      end
+
+      def list_by_record(collection:, record_id:, skip: 0, limit: nil,
+                         user_ids: nil, start_timestamp: nil, end_timestamp: nil, order: 'asc')
+        # `id` (insertion order) breaks ties on equal timestamps in both directions, keeping pages
+        # deterministic and stable.
+        relation = scope(collection, record_id, user_ids, start_timestamp, end_timestamp)
+                   .order(timestamp: order.to_s == 'desc' ? :desc : :asc, id: :asc)
+                   .offset(skip || 0)
+        relation = relation.limit(limit) unless limit.nil?
+
+        relation.map { |row| from_row(row) }
+      end
+
+      def count_by_record(collection:, record_id:, user_ids: nil, start_timestamp: nil, end_timestamp: nil)
+        scope(collection, record_id, user_ids, start_timestamp, end_timestamp).count
+      end
+
+      def list_by_correlation(collection:, record_id:, correlation_key:)
+        model.where(collection: collection, record_id: record_id, correlation_key: correlation_key)
+             .order(:timestamp, :id)
+             .map { |row| from_row(row) }
+      end
+
+      def list_by_correlations(collection:, record_id:, correlation_keys:)
+        return [] if correlation_keys.empty?
+
+        model.where(collection: collection, record_id: record_id, correlation_key: correlation_keys)
+             .order(:timestamp, :id)
+             .map { |row| from_row(row) }
+      end
+
+      private
+
+      def scope(collection, record_id, user_ids, start_timestamp, end_timestamp)
+        relation = model.where(collection: collection, record_id: record_id)
+        relation = relation.where(user_id: user_ids) if user_ids
+        # Compare as Time so ActiveRecord casts the bound to the datetime column's storage format
+        # (raw ISO strings with a `Z` would compare lexically against the cast rows and never match).
+        relation = relation.where('timestamp >= ?', as_time(start_timestamp)) if start_timestamp
+        relation = relation.where('timestamp <= ?', as_time(end_timestamp)) if end_timestamp
+        relation
+      end
+
+      def as_time(value)
+        value.is_a?(::Time) ? value : ::Time.iso8601(value.to_s)
+      end
+
+      def model
+        ensure_ready
+        @model
+      end
+
+      def ensure_ready
+        return if @ready
+
+        @mutex.synchronize do
+          return if @ready
+
+          Sql::AuditConnectionBase.establish_connection(@database)
+          connection = Sql::AuditConnectionBase.connection
+          Sql::Migrator.new(connection, schema: schema_for(connection), table_name: @table_name).run
+          @model = build_model(qualified(connection))
+          @ready = true
+        end
+      end
+
+      # A per-instance concrete subclass bound to this store's own table, so distinct stores can't
+      # clobber each other's table name. reset_column_information drops stale metadata for the table
+      # the migration just created/evolved.
+      def build_model(table)
+        Class.new(Sql::AuditLog) { self.table_name = table }.tap(&:reset_column_information)
+      end
+
+      def schema_for(connection)
+        connection.adapter_name.downcase.include?('postgres') ? @schema : nil
+      end
+
+      def qualified(connection)
+        schema = schema_for(connection)
+        schema ? "#{schema}.#{@table_name}" : @table_name
+      end
+
+      def to_row(record)
+        {
+          timestamp: record.timestamp,
+          operation: record.operation,
+          collection: record.collection,
+          record_id: record.record_id,
+          user_id: record.user_id,
+          correlation_key: record.correlation_key,
+          previous_values: record.previous_values,
+          new_values: record.new_values
+        }
+      end
+
+      def from_row(row)
+        AuditRecord.new(
+          timestamp: row.timestamp.respond_to?(:iso8601) ? row.timestamp.iso8601(3) : row.timestamp.to_s,
+          operation: row.operation,
+          collection: row.collection,
+          record_id: row.record_id,
+          user_id: row.user_id,
+          correlation_key: row.correlation_key,
+          previous_values: row.previous_values || {},
+          new_values: row.new_values || {}
+        )
+      end
+    end
+  end
+end
diff --git a/packages/forest_admin_agent/lib/forest_admin_agent/builder/agent_factory.rb b/packages/forest_admin_agent/lib/forest_admin_agent/builder/agent_factory.rb
index 1974d6baa..8099232f4 100644
--- a/packages/forest_admin_agent/lib/forest_admin_agent/builder/agent_factory.rb
+++ b/packages/forest_admin_agent/lib/forest_admin_agent/builder/agent_factory.rb
@@ -58,6 +58,7 @@ def use(plugin, options = {})
       end
 
       def build
+        install_audit_trail
         @container.register(:datasource, @customizer.datasource(@logger))
 
         # Reset route cache to ensure routes are computed with all customizations
@@ -289,12 +290,30 @@ def build_cache
         @options[:customize_error_message] =
           clean_option_value(@options[:customize_error_message], 'config.customize_error_message =')
         @options[:logger] = clean_option_value(@options[:logger], 'config.logger =')
+        build_audit_trail_store
 
         @container.register(:config, @options.to_h)
 
         configure_rpc_polling_pool if @options[:rpc_max_polling_threads]
       end
 
+      # The audit trail switches on as soon as a database is configured. The store is built here
+      # (cheap: the connection only opens on the first write) and shared through the config, so the
+      # record-history routes read the very table the capture layer writes to.
+      def build_audit_trail_store
+        options = @options[:audit_trail]
+        return unless options && options[:database]
+
+        options[:store] = AuditTrail::Store.new(**options.slice(:database, :schema, :table_name).compact)
+      end
+
+      def install_audit_trail
+        options = @options[:audit_trail]
+        return if options.nil? || options[:store].nil?
+
+        @customizer.use(AuditTrail::Capture, { store: options[:store], redact: options[:redact] })
+      end
+
       def configure_rpc_polling_pool
         max_threads = @options[:rpc_max_polling_threads].to_i
         return unless max_threads.positive?
diff --git a/packages/forest_admin_agent/lib/forest_admin_agent/http/correlation_id.rb b/packages/forest_admin_agent/lib/forest_admin_agent/http/correlation_id.rb
new file mode 100644
index 000000000..ded3da334
--- /dev/null
+++ b/packages/forest_admin_agent/lib/forest_admin_agent/http/correlation_id.rb
@@ -0,0 +1,37 @@
+require 'securerandom'
+
+module ForestAdminAgent
+  module Http
+    # Per-request correlation id, generated by the agent. Shared between the caller (so the audit
+    # trail can group every change made within one request) and the response header echoed back to
+    # the client. Mirrors the Node agent's `context.state.requestId` + `x-forest-correlation-id`
+    # header: the agent generates the id, never reads it from the incoming request.
+    #
+    # Stored thread-locally and generated lazily on first read (e.g. when the caller is parsed). The
+    # host resets it at the start of each request so a pooled thread never reuses a previous id.
+    module CorrelationId
+      HEADER = 'x-forest-correlation-id'.freeze
+      KEY = :forest_admin_correlation_id
+
+      module_function
+
+      # Lazily generate and memoize the id for the current request/thread.
+      def current
+        Thread.current[KEY] ||= SecureRandom.uuid
+      end
+
+      # The id if one was generated during this request, otherwise nil (does not generate one).
+      def current?
+        Thread.current[KEY]
+      end
+
+      def current=(value)
+        Thread.current[KEY] = value
+      end
+
+      def reset!
+        Thread.current[KEY] = nil
+      end
+    end
+  end
+end
diff --git a/packages/forest_admin_agent/lib/forest_admin_agent/http/correlation_id_middleware.rb b/packages/forest_admin_agent/lib/forest_admin_agent/http/correlation_id_middleware.rb
new file mode 100644
index 000000000..fbd6c6e99
--- /dev/null
+++ b/packages/forest_admin_agent/lib/forest_admin_agent/http/correlation_id_middleware.rb
@@ -0,0 +1,29 @@
+module ForestAdminAgent
+  module Http
+    # Rack middleware echoing the agent-generated correlation id back to the client, mirroring the
+    # Node agent's `correlationIdMiddleware` (`router.use(...)`). Hosts mount it in their middleware
+    # stack; CORS exposure of the header is handled by the host's CORS config (see the Rails engine).
+    #
+    # The id itself is generated lazily by the agent during the request (see CorrelationId, called
+    # from CallerParser). This middleware only resets the thread-local around the request — so a
+    # pooled thread never reuses a previous id — and sets the response header when one was generated.
+    class CorrelationIdMiddleware
+      def initialize(app)
+        @app = app
+      end
+
+      def call(env)
+        CorrelationId.reset!
+
+        status, headers, body = @app.call(env)
+
+        id = CorrelationId.current?
+        headers[CorrelationId::HEADER] = id if id
+
+        [status, headers, body]
+      ensure
+        CorrelationId.reset!
+      end
+    end
+  end
+end
diff --git a/packages/forest_admin_agent/lib/forest_admin_agent/http/router.rb b/packages/forest_admin_agent/lib/forest_admin_agent/http/router.rb
index 4bd0b896f..1b12950ae 100644
--- a/packages/forest_admin_agent/lib/forest_admin_agent/http/router.rb
+++ b/packages/forest_admin_agent/lib/forest_admin_agent/http/router.rb
@@ -64,6 +64,10 @@ def self.routes
           { name: 'dissociate_related', handler: -> { Resources::Related::DissociateRelated.new.routes } },
           { name: 'update_related', handler: -> { Resources::Related::UpdateRelated.new.routes } },
           { name: 'update_field', handler: -> { Resources::UpdateField.new.routes } },
+          # Registered before `audit_trail` so `/_audit-trail/correlation(s)` matches here instead of
+          # the per-record `/_audit-trail/:collection_name/:id` (Rails matches in definition order).
+          { name: 'audit_trail_correlation', handler: -> { Resources::AuditTrailCorrelation.new.routes } },
+          { name: 'audit_trail', handler: -> { Resources::AuditTrail.new.routes } },
           { name: 'workflow_executor_proxy', handler: -> { Workflow::WorkflowExecutorProxy.new.routes } }
         ]
 
diff --git a/packages/forest_admin_agent/lib/forest_admin_agent/routes/resources/audit_trail.rb b/packages/forest_admin_agent/lib/forest_admin_agent/routes/resources/audit_trail.rb
new file mode 100644
index 000000000..3f577efa9
--- /dev/null
+++ b/packages/forest_admin_agent/lib/forest_admin_agent/routes/resources/audit_trail.rb
@@ -0,0 +1,148 @@
+require 'active_support/time'
+
+module ForestAdminAgent
+  module Routes
+    module Resources
+      # Record-history route, mirroring the Node agent's `/_audit-trail/{collection}/:id`.
+      #
+      # Registered only when `config.audit_trail[:database]` is set, in which case the agent factory
+      # built the store the capture layer writes to.
+      class AuditTrail < AbstractAuthenticatedRoute
+        include ForestAdminAgent::Utils
+        include AuditTrailRoute
+
+        DEFAULT_PAGE_SIZE = 20
+        MAX_PAGE_SIZE = 100
+        DATE_ONLY = /\A\d{4}-\d{2}-\d{2}\z/
+        # Wall-clock datetime, `T` or space separator, seconds optional: `YYYY-MM-DD[T ]HH:mm[:ss]`.
+        DATE_TIME = /\A(\d{4}-\d{2}-\d{2})[T ](\d{2}):(\d{2})(?::(\d{2}))?\z/
+
+        def setup_routes
+          return self unless store
+
+          add_route(
+            'forest_audit_trail',
+            'get',
+            '/_audit-trail/:collection_name/:id',
+            ->(args) { handle_request(args) }
+          )
+
+          self
+        end
+
+        def handle_request(args = {})
+          context = build(args)
+          context.permissions.can?(:read, context.collection)
+          assert_record_in_scope(context, context.collection, args[:params]['id'])
+
+          skip, limit = parse_pagination(args)
+          filters = {
+            collection: context.collection.name,
+            # args[:params]['id'] is already Forest's packed id, the form the audit store keys on.
+            record_id: args[:params]['id'],
+            **parse_filters(args)
+          }
+
+          history = store.list_by_record(**filters, skip: skip, limit: limit, order: parse_sort(args))
+          # `count` reflects the active filters (not the absolute total) and is independent of the page.
+          count = store.count_by_record(**filters)
+
+          {
+            name: args[:params]['collection_name'],
+            content: { data: history.map { |record| serialize_record(record) }, meta: { count: count } }
+          }
+        end
+
+        private
+
+        # JSON:API `sort`: `timestamp` → oldest first, anything else (absent/unsupported) → newest first.
+        def parse_sort(args)
+          args.dig(:params, 'sort').to_s == 'timestamp' ? 'asc' : 'desc'
+        end
+
+        # JSON:API pagination: 1-based page[number] (default 1) and page[size] (default 20, capped at
+        # 100). Out-of-bound or non-numeric values fall back to the defaults rather than erroring.
+        def parse_pagination(args)
+          # `?page=foo` reaches us as a bare String, which `dig` refuses to walk into.
+          page = args.dig(:params, 'page')
+          page = {} unless page.is_a?(Hash)
+
+          size = page['size'].to_i
+          size = DEFAULT_PAGE_SIZE if size < 1
+          size = MAX_PAGE_SIZE if size > MAX_PAGE_SIZE
+
+          number = page['number'].to_i
+          number = 1 if number < 1
+
+          [(number - 1) * size, size]
+        end
+
+        def parse_filters(args)
+          timezone = args.dig(:params, 'timezone').to_s
+          timezone = 'UTC' if timezone.empty?
+
+          {
+            user_ids: parse_user_ids(args.dig(:params, 'userIds')),
+            start_timestamp: parse_date_boundary(args.dig(:params, 'startDate'), timezone, :start),
+            end_timestamp: parse_date_boundary(args.dig(:params, 'endDate'), timezone, :end)
+          }.compact
+        end
+
+        # Comma-separated integer ids; non-numeric tokens are dropped. Empty after parsing → no filter.
+        def parse_user_ids(raw)
+          return nil if raw.nil? || raw.to_s.empty?
+
+          ids = raw.to_s.split(',').map(&:strip).grep(/\A\d+\z/).map(&:to_i)
+          ids.empty? ? nil : ids
+        end
+
+        # `startDate`/`endDate` accept a bare day (`YYYY-MM-DD`) or a wall-clock datetime
+        # (`YYYY-MM-DD[T ]HH:mm[:ss]`), read as local time in the request timezone and returned as a UTC
+        # ISO instant the store can compare against stored timestamps.
+        def parse_date_boundary(raw, timezone, boundary)
+          return nil if raw.nil? || raw.to_s.empty?
+
+          zone = Time.find_zone(timezone)
+          raise Http::Exceptions::ValidationError, "Invalid timezone: \"#{timezone}\"" if zone.nil?
+
+          instant = begin
+            local_instant(zone, raw.to_s, boundary)
+          rescue ArgumentError
+            nil
+          end
+
+          if instant.nil?
+            raise Http::Exceptions::ValidationError,
+                  "Invalid date: \"#{raw}\" (expected YYYY-MM-DD or YYYY-MM-DDTHH:mm)"
+          end
+
+          instant.utc.iso8601(3)
+        end
+
+        def local_instant(zone, raw, boundary)
+          if DATE_ONLY.match?(raw)
+            day = zone.parse(raw)
+            # Bare day → start (00:00:00.000) or end (23:59:59.999) of that local day.
+            boundary == :end ? day.end_of_day : day.beginning_of_day
+          elsif (match = DATE_TIME.match(raw))
+            date, hours, minutes, seconds = match.captures
+            base = zone.parse("#{date}T#{hours}:#{minutes}")
+            if seconds
+              base.change(sec: seconds.to_i, usec: 0)
+            elsif boundary == :end
+              # Minutes-only end boundary stays inclusive to :59.999; start stays at :00.000.
+              base.change(sec: 59, usec: 999_000)
+            else
+              base
+            end
+          end
+        end
+
+        def store
+          config = ForestAdminAgent::Facades::Container.config_from_cache
+          config && config[:audit_trail] && config[:audit_trail][:store]
+        end
+      end
+    end
+  end
+end
diff --git a/packages/forest_admin_agent/lib/forest_admin_agent/routes/resources/audit_trail_correlation.rb b/packages/forest_admin_agent/lib/forest_admin_agent/routes/resources/audit_trail_correlation.rb
new file mode 100644
index 000000000..2743e4a7b
--- /dev/null
+++ b/packages/forest_admin_agent/lib/forest_admin_agent/routes/resources/audit_trail_correlation.rb
@@ -0,0 +1,106 @@
+module ForestAdminAgent
+  module Routes
+    module Resources
+      # Correlation-scoped record-history routes, mirroring the Node agent's
+      # `/_audit-trail/correlation/:key` and `/_audit-trail/correlations`. Registered only when
+      # `config.audit_trail[:database]` is set. All three routes are scoped to a single record through the
+      # `collection`/`recordId` query (GET) or body (POST) params and share the per-record auth.
+      class AuditTrailCorrelation < AbstractAuthenticatedRoute
+        include AuditTrailRoute
+
+        def setup_routes
+          return self unless store
+
+          add_route(
+            'forest_audit_trail_correlation',
+            'get',
+            '/_audit-trail/correlation/:correlation_key',
+            ->(args) { handle_history(args) }
+          )
+          # GET carries the keys in `correlationKeys`; POST accepts a body list to dodge URL limits.
+          add_route(
+            'forest_audit_trail_correlations',
+            'get',
+            '/_audit-trail/correlations',
+            ->(args) { handle_batch(args) }
+          )
+          add_route(
+            'forest_audit_trail_correlations_batch',
+            'post',
+            '/_audit-trail/correlations',
+            ->(args) { handle_batch(args) }
+          )
+
+          self
+        end
+
+        def handle_history(args = {})
+          collection, record_id = assert_scope(args)
+
+          history = store.list_by_correlation(
+            collection: collection.name,
+            record_id: record_id,
+            correlation_key: args[:params]['correlation_key']
+          )
+
+          { name: collection.name, content: { data: history.map { |record| serialize_record(record) } } }
+        end
+
+        def handle_batch(args = {})
+          collection, record_id = assert_scope(args)
+          correlation_keys = parse_correlation_keys(args)
+
+          history = if correlation_keys.empty?
+                      []
+                    else
+                      store.list_by_correlations(
+                        collection: collection.name,
+                        record_id: record_id,
+                        correlation_keys: correlation_keys
+                      )
+                    end
+
+          { name: collection.name, content: { data: history.map { |record| serialize_record(record) } } }
+        end
+
+        private
+
+        def assert_scope(args)
+          context = build(args)
+          name = args.dig(:params, 'collection').to_s
+          record_id = args.dig(:params, 'recordId').to_s
+
+          raise Http::Exceptions::ValidationError, 'Missing collection' if name.empty?
+          raise Http::Exceptions::ValidationError, 'Missing recordId' if record_id.empty?
+
+          collection = get_collection(context, name)
+          context.permissions.can?(:read, collection)
+          assert_record_in_scope(context, collection, record_id)
+
+          [collection, record_id]
+        end
+
+        def get_collection(context, name)
+          context.datasource.get_collection(name)
+        rescue ForestAdminDatasourceToolkit::Exceptions::ForestException => e
+          raise Http::Exceptions::NotFoundError, e.message if e.message.include?('not found')
+
+          raise
+        end
+
+        # Body array (POST) takes precedence, otherwise the comma-separated query param (GET).
+        def parse_correlation_keys(args)
+          raw = args.dig(:params, 'correlationKeys')
+          keys = raw.is_a?(Array) ? raw : raw.to_s.split(',')
+
+          keys.map { |key| key.to_s.strip }.reject(&:empty?)
+        end
+
+        def store
+          config = ForestAdminAgent::Facades::Container.config_from_cache
+          config && config[:audit_trail] && config[:audit_trail][:store]
+        end
+      end
+    end
+  end
+end
diff --git a/packages/forest_admin_agent/lib/forest_admin_agent/routes/resources/audit_trail_route.rb b/packages/forest_admin_agent/lib/forest_admin_agent/routes/resources/audit_trail_route.rb
new file mode 100644
index 000000000..1efa078b6
--- /dev/null
+++ b/packages/forest_admin_agent/lib/forest_admin_agent/routes/resources/audit_trail_route.rb
@@ -0,0 +1,44 @@
+module ForestAdminAgent
+  module Routes
+    module Resources
+      # Behaviour shared by every audit-trail route: they all take a packed record id straight from the
+      # request, so the caller's permission scope has to be checked against that record before any
+      # history is returned (`can?(:read, collection)` alone only proves access to the collection — a
+      # role restricted to a subset of the records would otherwise read the history of any of them),
+      # and they all serialize audit records the same way.
+      module AuditTrailRoute
+        include ForestAdminDatasourceToolkit::Components::Query
+
+        def assert_record_in_scope(context, collection, packed_id)
+          condition = ConditionTree::ConditionTreeFactory.match_records(
+            collection, [Utils::Id.unpack_id(collection, packed_id, with_key: true)]
+          )
+          scope = context.permissions.get_scope(collection)
+
+          return if any_record?(context, collection, ConditionTree::ConditionTreeFactory.intersect([condition, scope]))
+
+          # Nothing in scope means the record is either outside it or gone for good. A deleted record
+          # keeps its history readable — inspecting what was deleted is much of the point of an audit
+          # trail — so only a record that still exists outside the scope is refused. Without a scope
+          # the first query already answered the question.
+          return if scope.nil? || !any_record?(context, collection, condition)
+
+          raise Http::Exceptions::NotFoundError, 'Record does not exists'
+        end
+
+        # Camelize only the top-level keys; previous/new value hashes keep the record's own column names.
+        def serialize_record(record)
+          record.to_h.transform_keys { |key| key.to_s.camelize(:lower) }
+        end
+
+        def any_record?(context, collection, condition_tree)
+          collection.list(
+            context.caller,
+            Filter.new(condition_tree: condition_tree),
+            Projection.new(ForestAdminDatasourceToolkit::Utils::Schema.primary_keys(collection))
+          ).any?
+        end
+      end
+    end
+  end
+end
diff --git a/packages/forest_admin_agent/lib/forest_admin_agent/utils/caller_parser.rb b/packages/forest_admin_agent/lib/forest_admin_agent/utils/caller_parser.rb
index 679c01fc5..72cb51905 100644
--- a/packages/forest_admin_agent/lib/forest_admin_agent/utils/caller_parser.rb
+++ b/packages/forest_admin_agent/lib/forest_admin_agent/utils/caller_parser.rb
@@ -18,6 +18,10 @@ def parse
         @token_data = decode_token
         @token_data[:timezone] = extract_timezone
         @token_data[:request] = { ip: @args[:headers]['action_dispatch.remote_ip'].to_s }
+        # One id per request, generated by the agent, shared by every operation it triggers (used to
+        # correlate the audit trail) and echoed back to the client in the response header.
+        # See ForestAdminAgent::Http::CorrelationId.
+        @token_data[:request_id] = Http::CorrelationId.current
         project, environment = extract_forest_context
         @token_data[:project] = project
         @token_data[:environment] = environment
diff --git a/packages/forest_admin_agent/spec/lib/forest_admin_agent/audit_trail/capture_spec.rb b/packages/forest_admin_agent/spec/lib/forest_admin_agent/audit_trail/capture_spec.rb
new file mode 100644
index 000000000..3b298af21
--- /dev/null
+++ b/packages/forest_admin_agent/spec/lib/forest_admin_agent/audit_trail/capture_spec.rb
@@ -0,0 +1,170 @@
+require 'spec_helper'
+
+module ForestAdminAgent
+  module AuditTrail
+    describe Capture do
+      let(:column_schema) { ForestAdminDatasourceToolkit::Schema::ColumnSchema }
+
+      let(:store) do
+        Class.new do
+          attr_reader :records
+
+          def initialize
+            @records = []
+          end
+
+          def append(record)
+            @records << record
+          end
+        end.new
+      end
+
+      let(:fields) do
+        {
+          'id' => column_schema.new(column_type: 'Number', is_primary_key: true, is_read_only: true),
+          'name' => column_schema.new(column_type: 'String'),
+          'address' => column_schema.new(column_type: 'Json')
+        }
+      end
+
+      let(:hooks) { {} }
+      let(:collection) { double('collection') }
+      let(:caller_double) { double('caller', id: 42, request_id: 'req-xyz') }
+
+      # What a hook context really hands out: the caller is already bound, so `list` takes
+      # (filter, projection). A verifying double keeps that contract honest.
+      let(:relaxed_collection) do
+        instance_double(ForestAdminDatasourceCustomizer::Context::RelaxedWrappers::RelaxedCollection)
+      end
+
+      let(:collection_customizer) do
+        customizer = double('CollectionCustomizer', name: 'companies', collection: collection)
+        allow(customizer).to receive(:add_hook) { |position, type, &block| hooks["#{position}_#{type}"] = block }
+        customizer
+      end
+
+      let(:datasource_customizer) do
+        double('DatasourceCustomizer', collections: { 'companies' => collection_customizer })
+      end
+
+      before do
+        Thread.current[:forest_audit_trail_snapshots] = nil
+        allow(collection).to receive(:schema).and_return({ fields: fields })
+        described_class.new.run(datasource_customizer, nil, store: store)
+      end
+
+      it 'records a create with only the writable columns' do
+        record = { 'id' => 1, 'name' => 'Acme', 'address' => { 'city' => 'Paris' } }
+
+        hooks['After_Create'].call(double('ctx', caller: caller_double, record: record))
+
+        audit = store.records.last
+        expect(audit.operation).to eq('create')
+        expect(audit.record_id).to eq('1')
+        expect(audit.user_id).to eq(42)
+        expect(audit.correlation_key).to eq('req-xyz')
+        expect(audit.previous_values).to eq({})
+        expect(audit.new_values).to eq({ 'name' => 'Acme', 'address' => { 'city' => 'Paris' } })
+      end
+
+      def before_hook(type, patch: nil)
+        hooks["Before_#{type}"].call(
+          double('before', caller: caller_double, filter: Object.new,
+                           collection: relaxed_collection, patch: patch)
+        )
+      end
+
+      def after_hook(type)
+        hooks["After_#{type}"].call(double('after', caller: caller_double, filter: Object.new))
+      end
+
+      it "shares the caller's request id as the correlation key across records of one operation" do
+        allow(relaxed_collection).to receive(:list).and_return(
+          [{ 'id' => 1, 'name' => 'A' }, { 'id' => 2, 'name' => 'B' }]
+        )
+
+        before_hook('Update', patch: { 'name' => 'Z' })
+        after_hook('Update')
+
+        expect(store.records.map(&:correlation_key)).to eq(%w[req-xyz req-xyz])
+      end
+
+      it 'records an update with the minimal nested diff' do
+        before_record = { 'id' => 1, 'name' => 'Acme', 'address' => { 'city' => 'Paris', 'zip' => '1' } }
+        allow(relaxed_collection).to receive(:list).and_return([before_record])
+
+        before_hook('Update', patch: { 'address' => { 'city' => 'Lyon', 'zip' => '1' } })
+        after_hook('Update')
+
+        audit = store.records.last
+        expect(audit.operation).to eq('update')
+        expect(audit.previous_values).to eq({ 'address' => { 'city' => 'Paris' } })
+        expect(audit.new_values).to eq({ 'address' => { 'city' => 'Lyon' } })
+      end
+
+      it 'does not record an update when nothing writable changed' do
+        allow(relaxed_collection).to receive(:list).and_return([{ 'id' => 1, 'name' => 'Acme' }])
+
+        before_hook('Update', patch: { 'name' => 'Acme' })
+        after_hook('Update')
+
+        expect(store.records).to be_empty
+      end
+
+      # The before and after hooks are handed different filter objects (the after context always
+      # carries the caller's original), so pairing must not depend on that object.
+      it 'pairs the snapshot with its after hook without relying on the filter object' do
+        allow(relaxed_collection).to receive(:list).and_return([{ 'id' => 1, 'name' => 'Acme' }])
+
+        before_hook('Update', patch: { 'name' => 'Effective' })
+        after_hook('Update')
+
+        expect(store.records.last.new_values).to eq({ 'name' => 'Effective' })
+      end
+
+      it 'still audits the next write after one raised between its hooks' do
+        allow(relaxed_collection).to receive(:list).and_return([{ 'id' => 1, 'name' => 'Acme' }])
+
+        before_hook('Update', patch: { 'name' => 'Never written' })
+        before_hook('Update', patch: { 'name' => 'Z' })
+        after_hook('Update')
+
+        expect(store.records.map { |record| record.new_values['name'] }).to eq(['Z'])
+      end
+
+      it 'drops the oldest stranded snapshots instead of growing the thread-local without bound' do
+        allow(relaxed_collection).to receive(:list).and_return([{ 'id' => 1, 'name' => 'Acme' }])
+
+        (described_class::MAX_SNAPSHOTS + 4).times { before_hook('Update', patch: { 'name' => 'stranded' }) }
+        before_hook('Update', patch: { 'name' => 'Z' })
+        after_hook('Update')
+
+        expect(Thread.current[:forest_audit_trail_snapshots].size).to eq(described_class::MAX_SNAPSHOTS - 1)
+        expect(store.records.map { |record| record.new_values['name'] }).to eq(['Z'])
+      end
+
+      it 'records a delete with the previous values' do
+        allow(relaxed_collection).to receive(:list).and_return([{ 'id' => 7, 'name' => 'Gone', 'address' => nil }])
+
+        before_hook('Delete')
+        after_hook('Delete')
+
+        audit = store.records.last
+        expect(audit.operation).to eq('delete')
+        expect(audit.record_id).to eq('7')
+        expect(audit.previous_values).to eq({ 'name' => 'Gone', 'address' => nil })
+        expect(audit.new_values).to eq({})
+      end
+
+      it 'masks redacted fields while still recording the change' do
+        described_class.new.run(datasource_customizer, nil, store: store, redact: { 'companies' => ['name'] })
+
+        hooks['After_Create'].call(
+          double('ctx', caller: caller_double, record: { 'id' => 1, 'name' => 'Secret', 'address' => nil })
+        )
+
+        expect(store.records.last.new_values['name']).to eq(described_class::REDACTED)
+      end
+    end
+  end
+end
diff --git a/packages/forest_admin_agent/spec/lib/forest_admin_agent/audit_trail/diff_spec.rb b/packages/forest_admin_agent/spec/lib/forest_admin_agent/audit_trail/diff_spec.rb
new file mode 100644
index 000000000..10405b2eb
--- /dev/null
+++ b/packages/forest_admin_agent/spec/lib/forest_admin_agent/audit_trail/diff_spec.rb
@@ -0,0 +1,68 @@
+require 'spec_helper'
+
+module ForestAdminAgent
+  module AuditTrail
+    describe Diff do
+      describe '.diff' do
+        it 'returns nil when values are deeply equal regardless of hash key order' do
+          expect(described_class.diff({ 'a' => 1, 'b' => 2 }, { 'b' => 2, 'a' => 1 })).to be_nil
+        end
+
+        it 'keeps only the changed leaf of a nested object' do
+          before = { 'city' => 'Paris', 'zip' => '75001' }
+          after = { 'city' => 'Lyon', 'zip' => '75001' }
+
+          expect(described_class.diff(before, after)).to eq(
+            previous: { 'city' => 'Paris' },
+            next: { 'city' => 'Lyon' }
+          )
+        end
+
+        it 'reports a key that disappeared, even when its value was nil' do
+          expect(described_class.diff({ 'flag' => nil }, {})).to eq(
+            previous: { 'flag' => nil },
+            next: { 'flag' => nil }
+          )
+        end
+
+        it 'reports a key that appeared holding nil' do
+          expect(described_class.diff({}, { 'flag' => nil })).to eq(
+            previous: { 'flag' => nil },
+            next: { 'flag' => nil }
+          )
+        end
+
+        it 'diffs an array of objects index by index' do
+          before = [{ 'name' => 'a' }, { 'name' => 'b' }]
+          after = [{ 'name' => 'a' }, { 'name' => 'c' }]
+
+          expect(described_class.diff(before, after)).to eq(
+            previous: { 1 => { 'name' => 'b' } },
+            next: { 1 => { 'name' => 'c' } }
+          )
+        end
+
+        it 'keeps scalars and primitive arrays whole' do
+          expect(described_class.diff(%w[a b], %w[a c])).to eq(previous: %w[a b], next: %w[a c])
+        end
+
+        it 'reports nil for a newly set or cleared value' do
+          expect(described_class.diff(nil, 'x')).to eq(previous: nil, next: 'x')
+          expect(described_class.diff('x', nil)).to eq(previous: 'x', next: nil)
+        end
+      end
+
+      describe '.changed_values' do
+        it 'only records writable columns present in the patch that actually changed' do
+          before = { 'status' => 'open', 'name' => 'Acme', 'ignored' => 1 }
+          patch = { 'status' => 'closed', 'name' => 'Acme' }
+
+          expect(described_class.changed_values(before, patch, %w[status name])).to eq(
+            previous_values: { 'status' => 'open' },
+            new_values: { 'status' => 'closed' }
+          )
+        end
+      end
+    end
+  end
+end
diff --git a/packages/forest_admin_agent/spec/lib/forest_admin_agent/audit_trail/store_spec.rb b/packages/forest_admin_agent/spec/lib/forest_admin_agent/audit_trail/store_spec.rb
new file mode 100644
index 000000000..4b766d463
--- /dev/null
+++ b/packages/forest_admin_agent/spec/lib/forest_admin_agent/audit_trail/store_spec.rb
@@ -0,0 +1,155 @@
+require 'spec_helper'
+require 'tempfile'
+
+module ForestAdminAgent
+  module AuditTrail
+    describe Store do
+      let(:db) { Tempfile.new(['audit', '.sqlite3']) }
+      let(:store) { described_class.new(database: { adapter: 'sqlite3', database: db.path }) }
+
+      after do
+        Sql::AuditConnectionBase.remove_connection
+        db.close!
+      end
+
+      def record(over = {})
+        AuditRecord.new(
+          operation: 'update', collection: 'accounts', record_id: '1',
+          previous_values: { 'status' => 'open' }, new_values: { 'status' => 'closed' },
+          timestamp: '2026-01-02T03:04:05.000Z', user_id: 42, correlation_key: 'req-1', **over
+        )
+      end
+
+      def connection
+        Sql::AuditConnectionBase.connection
+      end
+
+      it 'creates the audit table with the expected columns on first write' do
+        store.append(record)
+
+        expect(store.send(:model).column_names.sort).to eq(
+          %w[collection correlation_key id new_values operation previous_values record_id timestamp user_id]
+        )
+      end
+
+      it 'persists and reads back a record, decoding the JSON columns' do
+        store.append(record)
+
+        audit = store.list_by_record(collection: 'accounts', record_id: '1').first
+        expect(audit.operation).to eq('update')
+        expect(audit.user_id).to eq(42)
+        expect(audit.previous_values).to eq({ 'status' => 'open' })
+        expect(audit.new_values).to eq({ 'status' => 'closed' })
+      end
+
+      it 'returns a record history oldest-first, scoped to the record, honoring skip/limit' do
+        store.append(record(timestamp: '2026-01-02T03:04:06.000Z', correlation_key: 'b'))
+        store.append(record(timestamp: '2026-01-02T03:04:05.000Z', correlation_key: 'a'))
+        store.append(record(record_id: '2', correlation_key: 'other'))
+
+        history = store.list_by_record(collection: 'accounts', record_id: '1')
+        expect(history.map(&:correlation_key)).to eq(%w[a b])
+
+        page = store.list_by_record(collection: 'accounts', record_id: '1', skip: 1, limit: 1)
+        expect(page.map(&:correlation_key)).to eq(['b'])
+      end
+
+      it 'sorts newest first when order is desc, breaking ties by insertion order' do
+        store.append(record(timestamp: '2026-01-02T03:04:05.000Z', correlation_key: 'a'))
+        store.append(record(timestamp: '2026-01-02T03:04:06.000Z', correlation_key: 'b'))
+        store.append(record(timestamp: '2026-01-02T03:04:05.000Z', correlation_key: 'a2'))
+
+        history = store.list_by_record(collection: 'accounts', record_id: '1', order: 'desc')
+        expect(history.map(&:correlation_key)).to eq(%w[b a a2])
+      end
+
+      it 'filters by user_ids and inclusive timestamp range' do
+        store.append(record(timestamp: '2026-01-02T03:04:05.000Z', user_id: 7, correlation_key: 'keep'))
+        store.append(record(timestamp: '2026-01-02T03:04:09.000Z', user_id: 7, correlation_key: 'late'))
+        store.append(record(timestamp: '2026-01-02T03:04:05.000Z', user_id: 9, correlation_key: 'other'))
+
+        history = store.list_by_record(
+          collection: 'accounts', record_id: '1', user_ids: [7],
+          start_timestamp: '2026-01-02T03:04:04.000Z', end_timestamp: '2026-01-02T03:04:06.000Z'
+        )
+        expect(history.map(&:correlation_key)).to eq(['keep'])
+      end
+
+      it 'counts matches independently of skip/limit, respecting filters' do
+        store.append(record(user_id: 7))
+        store.append(record(user_id: 7))
+        store.append(record(user_id: 9))
+
+        expect(store.count_by_record(collection: 'accounts', record_id: '1')).to eq(3)
+        expect(store.count_by_record(collection: 'accounts', record_id: '1', user_ids: [7])).to eq(2)
+      end
+
+      it 'lists entries under a correlation key for the record, scoped and oldest first' do
+        store.append(record(record_id: '1', correlation_key: 'req-1', timestamp: '2026-01-01T00:00:02.000Z'))
+        store.append(record(record_id: '1', correlation_key: 'req-1', timestamp: '2026-01-01T00:00:01.000Z'))
+        store.append(record(record_id: '1', correlation_key: 'req-2'))
+        store.append(record(record_id: '2', correlation_key: 'req-1'))
+
+        history = store.list_by_correlation(collection: 'accounts', record_id: '1', correlation_key: 'req-1')
+        expect(history.map(&:timestamp)).to eq(['2026-01-01T00:00:01.000Z', '2026-01-01T00:00:02.000Z'])
+      end
+
+      it 'lists a flat history across multiple correlation keys, oldest first' do
+        store.append(record(correlation_key: 'a', timestamp: '2026-01-03T00:00:00.000Z'))
+        store.append(record(correlation_key: 'b', timestamp: '2026-01-01T00:00:00.000Z'))
+        store.append(record(correlation_key: 'a', timestamp: '2026-01-02T00:00:00.000Z'))
+        store.append(record(correlation_key: 'c', timestamp: '2026-01-04T00:00:00.000Z'))
+
+        history = store.list_by_correlations(collection: 'accounts', record_id: '1', correlation_keys: %w[a b])
+        expect(history.map(&:timestamp)).to eq(
+          ['2026-01-01T00:00:00.000Z', '2026-01-02T00:00:00.000Z', '2026-01-03T00:00:00.000Z']
+        )
+      end
+
+      it 'returns an empty array for an empty correlation key list' do
+        store.append(record(correlation_key: 'a'))
+
+        expect(store.list_by_correlations(collection: 'accounts', record_id: '1', correlation_keys: [])).to eq([])
+      end
+
+      it 'tracks applied migrations and is idempotent across stores' do
+        store.append(record)
+        described_class.new(database: { adapter: 'sqlite3', database: db.path }).append(record)
+
+        names = connection.select_values('SELECT name FROM audit_migrations ORDER BY name')
+        expect(names).to eq(['audit_logs:001-create-audit-logs', 'audit_logs:002-index-record-and-correlation'])
+      end
+
+      it 'migrates a second table in the same database instead of reading the first one as done' do
+        store.append(record)
+        other = described_class.new(database: { adapter: 'sqlite3', database: db.path }, table_name: 'other_logs')
+
+        expect { other.append(record) }.not_to raise_error
+        expect(other.list_by_record(collection: 'accounts', record_id: '1').size).to eq(1)
+        expect(connection.indexes('other_logs').map(&:name)).to include('other_logs_record_id')
+      end
+
+      it 'binds each store to its own model class instead of mutating a shared one' do
+        other = described_class.new(database: { adapter: 'sqlite3', database: db.path })
+        store.append(record(correlation_key: 'main'))
+        other.append(record(correlation_key: 'other'))
+
+        # No shared mutable model: AuditLog is an abstract template, each store owns a distinct subclass.
+        expect(Sql::AuditLog.abstract_class?).to be(true)
+        expect(store.send(:model)).not_to equal(other.send(:model))
+        expect(store.send(:model).table_name).to eq('audit_logs')
+        expect(store.list_by_record(collection: 'accounts', record_id: '1').map(&:correlation_key))
+          .to eq(%w[main other])
+      end
+
+      it 'indexes record_id, correlation_key and user_id' do
+        store.append(record)
+
+        index_names = connection.indexes('audit_logs').map(&:name)
+        expect(index_names).to include(
+          'audit_logs_record_id', 'audit_logs_correlation_key', 'audit_logs_user_id'
+        )
+      end
+    end
+  end
+end
diff --git a/packages/forest_admin_agent/spec/lib/forest_admin_agent/builder/agent_factory_spec.rb b/packages/forest_admin_agent/spec/lib/forest_admin_agent/builder/agent_factory_spec.rb
index 0382c327f..2f2559969 100644
--- a/packages/forest_admin_agent/spec/lib/forest_admin_agent/builder/agent_factory_spec.rb
+++ b/packages/forest_admin_agent/spec/lib/forest_admin_agent/builder/agent_factory_spec.rb
@@ -254,6 +254,53 @@ module Builder
           end
         end
 
+        describe 'audit trail' do
+          let(:instance) { described_class.instance }
+          let(:options) do
+            {
+              auth_secret: 'cba803d01a4d43b55010cab41fa1ea1f1f51a95e',
+              env_secret: '89719c6d8e2e2de2694c2f220fe2dbf02d5289487364daf1e4c6b13733ed0cdb',
+              is_production: false,
+              schema_path: File.join('tmp', '.forestadmin-schema.json')
+            }
+          end
+          let(:audit_options) { { database: { adapter: 'sqlite3', database: ':memory:' } } }
+
+          before { allow(instance).to receive(:send_schema) }
+
+          it 'stays off when no audit trail database is configured' do
+            instance.setup(options)
+            allow(instance.customizer).to receive(:use)
+
+            instance.build
+
+            expect(instance.container.resolve(:config)[:audit_trail]).to be_nil
+            expect(instance.customizer).not_to have_received(:use)
+          end
+
+          it 'stays off when the audit trail option carries no database' do
+            instance.setup(options.merge(audit_trail: { redact: { 'users' => ['email'] } }))
+            allow(instance.customizer).to receive(:use)
+
+            instance.build
+
+            expect(instance.container.resolve(:config)[:audit_trail][:store]).to be_nil
+            expect(instance.customizer).not_to have_received(:use)
+          end
+
+          it 'builds the store from the configured database and installs the capture layer' do
+            instance.setup(options.merge(audit_trail: audit_options.merge(redact: { 'users' => ['email'] })))
+            allow(instance.customizer).to receive(:use)
+
+            instance.build
+
+            store = instance.container.resolve(:config)[:audit_trail][:store]
+            expect(store).to be_a(AuditTrail::Store)
+            expect(instance.customizer).to have_received(:use)
+              .with(AuditTrail::Capture, { store: store, redact: { 'users' => ['email'] } })
+          end
+        end
+
         describe 'generate_schema_only' do
           it 'generates schema and writes to default path' do
             instance = described_class.instance
diff --git a/packages/forest_admin_agent/spec/lib/forest_admin_agent/http/correlation_id_middleware_spec.rb b/packages/forest_admin_agent/spec/lib/forest_admin_agent/http/correlation_id_middleware_spec.rb
new file mode 100644
index 000000000..474329e1c
--- /dev/null
+++ b/packages/forest_admin_agent/spec/lib/forest_admin_agent/http/correlation_id_middleware_spec.rb
@@ -0,0 +1,46 @@
+require 'spec_helper'
+
+module ForestAdminAgent
+  module Http
+    describe CorrelationIdMiddleware do
+      after { CorrelationId.reset! }
+
+      it 'echoes the id generated during the request on the response header' do
+        app = ->(_env) { [200, {}, [CorrelationId.current]] }
+
+        _status, headers, body = described_class.new(app).call({})
+
+        expect(headers[CorrelationId::HEADER]).to eq(body.first)
+      end
+
+      it 'does not set the header when no id was generated during the request' do
+        app = ->(_env) { [200, {}, ['ok']] }
+
+        _status, headers, = described_class.new(app).call({})
+
+        expect(headers).not_to have_key(CorrelationId::HEADER)
+      end
+
+      it 'resets any leaked id before handling the request' do
+        CorrelationId.current = 'stale'
+        seen = 'unset'
+        app = lambda do |_env|
+          seen = CorrelationId.current?
+          [200, {}, ['ok']]
+        end
+
+        described_class.new(app).call({})
+
+        expect(seen).to be_nil
+      end
+
+      it 'clears the id after the request so the thread is not reused with a stale id' do
+        app = ->(_env) { [200, {}, [CorrelationId.current]] }
+
+        described_class.new(app).call({})
+
+        expect(CorrelationId.current?).to be_nil
+      end
+    end
+  end
+end
diff --git a/packages/forest_admin_agent/spec/lib/forest_admin_agent/http/correlation_id_spec.rb b/packages/forest_admin_agent/spec/lib/forest_admin_agent/http/correlation_id_spec.rb
new file mode 100644
index 000000000..6809ce09f
--- /dev/null
+++ b/packages/forest_admin_agent/spec/lib/forest_admin_agent/http/correlation_id_spec.rb
@@ -0,0 +1,29 @@
+require 'spec_helper'
+
+module ForestAdminAgent
+  module Http
+    describe CorrelationId do
+      after { described_class.reset! }
+
+      it 'lazily generates and memoizes an id within the thread' do
+        id = described_class.current
+
+        expect(id).to match(/\A[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}\z/)
+        expect(described_class.current).to eq(id)
+      end
+
+      it 'can be seeded by the host' do
+        described_class.current = 'req-1'
+
+        expect(described_class.current).to eq('req-1')
+      end
+
+      it 'reset! clears it so a fresh id is generated next' do
+        first = described_class.current
+        described_class.reset!
+
+        expect(described_class.current).not_to eq(first)
+      end
+    end
+  end
+end
diff --git a/packages/forest_admin_agent/spec/lib/forest_admin_agent/routes/resources/audit_trail_correlation_spec.rb b/packages/forest_admin_agent/spec/lib/forest_admin_agent/routes/resources/audit_trail_correlation_spec.rb
new file mode 100644
index 000000000..66f5e9e7b
--- /dev/null
+++ b/packages/forest_admin_agent/spec/lib/forest_admin_agent/routes/resources/audit_trail_correlation_spec.rb
@@ -0,0 +1,137 @@
+require 'spec_helper'
+
+module ForestAdminAgent
+  module Routes
+    module Resources
+      include ForestAdminDatasourceToolkit::Schema
+      include ForestAdminDatasourceToolkit::Components::Query::ConditionTree
+
+      describe AuditTrailCorrelation do
+        let(:store) { double('store') }
+        let(:permissions) { double('permissions', can?: true, get_scope: nil) }
+        let(:collection) do
+          build_collection(
+            name: 'books',
+            schema: {
+              fields: {
+                'id' => ColumnSchema.new(
+                  column_type: 'Number', is_primary_key: true,
+                  filter_operators: [Operators::IN, Operators::EQUAL]
+                )
+              }
+            },
+            list: [{ 'id' => 2 }]
+          )
+        end
+
+        def route_with_store(history: [])
+          allow(ForestAdminAgent::Facades::Container).to receive(:config_from_cache)
+            .and_return({ audit_trail: { store: store } })
+          allow(store).to receive_messages(list_by_correlation: history, list_by_correlations: history)
+
+          route = described_class.new
+          datasource = double('datasource')
+          allow(datasource).to receive(:get_collection).with('books').and_return(collection)
+          context = double('context', datasource: datasource, caller: build_caller, permissions: permissions)
+          allow(route).to receive(:build).and_return(context)
+          route
+        end
+
+        it 'returns 404 without touching the store when the record exists outside the caller scope' do
+          allow(permissions).to receive(:get_scope).and_return(Nodes::ConditionTreeLeaf.new('id', Operators::EQUAL, 9))
+          allow(collection).to receive(:list).and_return([], [{ 'id' => 2 }])
+          route = route_with_store
+
+          expect do
+            route.handle_history(
+              { headers: {}, params: { 'collection' => 'books', 'recordId' => '2', 'correlation_key' => 'req-1' } }
+            )
+          end.to raise_error(Http::Exceptions::NotFoundError)
+
+          expect(store).not_to have_received(:list_by_correlation)
+        end
+
+        it 'registers the correlation routes when a store is configured' do
+          allow(ForestAdminAgent::Facades::Container).to receive(:config_from_cache)
+            .and_return({ audit_trail: { store: Object.new } })
+
+          expect(described_class.new.routes.keys).to include(
+            'forest_audit_trail_correlation', 'forest_audit_trail_correlations', 'forest_audit_trail_correlations_batch'
+          )
+        end
+
+        it 'does not register when no store is configured' do
+          allow(ForestAdminAgent::Facades::Container).to receive(:config_from_cache).and_return({})
+
+          expect(described_class.new.routes).to be_empty
+        end
+
+        it 'reads a single correlation history scoped to the record' do
+          entry = { operation: 'update', record_id: '2', new_values: { 'first_name' => 'Jo' } }
+          route = route_with_store(history: [double('entry', to_h: entry)])
+
+          result = route.handle_history(
+            { headers: {}, params: { 'collection' => 'books', 'recordId' => '2', 'correlation_key' => 'req-1' } }
+          )
+
+          expect(store).to have_received(:list_by_correlation).with(
+            collection: 'books', record_id: '2', correlation_key: 'req-1'
+          )
+          # Same serialization as the per-record route: camelCase on top, column names left alone.
+          expect(result[:content]).to eq(
+            { data: [{ 'operation' => 'update', 'recordId' => '2', 'newValues' => { 'first_name' => 'Jo' } }] }
+          )
+        end
+
+        it 'reads a batch history from comma-separated query keys (GET)' do
+          route = route_with_store(history: [double('entry', to_h: { operation: 'update' })])
+
+          route.handle_batch(
+            { headers: {}, params: { 'collection' => 'books', 'recordId' => '2', 'correlationKeys' => 'a, b' } }
+          )
+
+          expect(store).to have_received(:list_by_correlations).with(
+            collection: 'books', record_id: '2', correlation_keys: %w[a b]
+          )
+        end
+
+        it 'reads a batch history from a body array (POST)' do
+          route = route_with_store
+
+          route.handle_batch(
+            { headers: {}, params: { 'collection' => 'books', 'recordId' => '2', 'correlationKeys' => %w[a b] } }
+          )
+
+          expect(store).to have_received(:list_by_correlations).with(
+            collection: 'books', record_id: '2', correlation_keys: %w[a b]
+          )
+        end
+
+        it 'returns an empty batch without hitting the store when no keys are given' do
+          route = route_with_store
+
+          result = route.handle_batch({ headers: {}, params: { 'collection' => 'books', 'recordId' => '2' } })
+
+          expect(store).not_to have_received(:list_by_correlations)
+          expect(result[:content]).to eq({ data: [] })
+        end
+
+        it 'rejects a missing collection' do
+          route = route_with_store
+
+          expect do
+            route.handle_history({ headers: {}, params: { 'recordId' => '2', 'correlation_key' => 'req-1' } })
+          end.to raise_error(Http::Exceptions::ValidationError, /Missing collection/)
+        end
+
+        it 'rejects a missing recordId' do
+          route = route_with_store
+
+          expect do
+            route.handle_history({ headers: {}, params: { 'collection' => 'books', 'correlation_key' => 'req-1' } })
+          end.to raise_error(Http::Exceptions::ValidationError, /Missing recordId/)
+        end
+      end
+    end
+  end
+end
diff --git a/packages/forest_admin_agent/spec/lib/forest_admin_agent/routes/resources/audit_trail_spec.rb b/packages/forest_admin_agent/spec/lib/forest_admin_agent/routes/resources/audit_trail_spec.rb
new file mode 100644
index 000000000..9ade67934
--- /dev/null
+++ b/packages/forest_admin_agent/spec/lib/forest_admin_agent/routes/resources/audit_trail_spec.rb
@@ -0,0 +1,179 @@
+require 'spec_helper'
+
+module ForestAdminAgent
+  module Routes
+    module Resources
+      include ForestAdminDatasourceToolkit::Schema
+      include ForestAdminDatasourceToolkit::Components::Query::ConditionTree
+
+      describe AuditTrail do
+        let(:store) { double('store') }
+        let(:permissions) { double('permissions', can?: true, get_scope: nil) }
+        let(:collection) do
+          build_collection(
+            name: 'projects',
+            schema: {
+              fields: {
+                'id' => ColumnSchema.new(
+                  column_type: 'Number', is_primary_key: true,
+                  filter_operators: [Operators::IN, Operators::EQUAL]
+                )
+              }
+            },
+            list: [{ 'id' => 4 }]
+          )
+        end
+
+        def route_with_store(records: [])
+          allow(ForestAdminAgent::Facades::Container).to receive(:config_from_cache)
+            .and_return({ audit_trail: { store: store } })
+          allow(store).to receive_messages(list_by_record: records, count_by_record: records.length)
+
+          route = described_class.new
+          context = double('context', collection: collection, caller: build_caller, permissions: permissions)
+          allow(route).to receive(:build).and_return(context)
+          route
+        end
+
+        it 'registers the record-history route when an audit_trail store is configured' do
+          allow(ForestAdminAgent::Facades::Container).to receive(:config_from_cache)
+            .and_return({ audit_trail: { store: Object.new } })
+
+          expect(described_class.new.routes).to include('forest_audit_trail')
+        end
+
+        it 'does not register the route when no audit_trail store is configured' do
+          allow(ForestAdminAgent::Facades::Container).to receive(:config_from_cache).and_return({})
+
+          expect(described_class.new.routes).not_to include('forest_audit_trail')
+        end
+
+        it 'reads the history scoped to the packed id and returns data + filtered count' do
+          entry = { operation: 'update', record_id: '4', previous_values: { 'first_name' => 'Jo' } }
+          route = route_with_store(records: [double('entry', to_h: entry)])
+
+          result = route.handle_request({ headers: {}, params: { 'collection_name' => 'projects', 'id' => '4' } })
+
+          expect(store).to have_received(:list_by_record).with(
+            collection: 'projects', record_id: '4', skip: 0, limit: 20, order: 'desc'
+          )
+          expect(store).to have_received(:count_by_record).with(collection: 'projects', record_id: '4')
+          # Top-level keys are camelCased for the frontend; nested value hashes keep the column names.
+          expect(result[:content]).to eq(
+            {
+              data: [{ 'operation' => 'update', 'recordId' => '4', 'previousValues' => { 'first_name' => 'Jo' } }],
+              meta: { count: 1 }
+            }
+          )
+        end
+
+        it 'intersects the record with the permission scope before reading any history' do
+          scope = Nodes::ConditionTreeLeaf.new('id', Operators::EQUAL, 4)
+          allow(permissions).to receive(:get_scope).and_return(scope)
+          route = route_with_store
+
+          route.handle_request({ headers: {}, params: { 'collection_name' => 'projects', 'id' => '4' } })
+
+          expect(collection).to have_received(:list) do |_caller, filter, _projection|
+            expect(filter.condition_tree.conditions).to include(scope)
+          end
+        end
+
+        it 'returns 404 without touching the store when the record exists outside the caller scope' do
+          allow(permissions).to receive(:get_scope).and_return(Nodes::ConditionTreeLeaf.new('id', Operators::EQUAL, 9))
+          # Empty under the scope, found without it: the record is someone else's, not a deleted one.
+          allow(collection).to receive(:list).and_return([], [{ 'id' => 4 }])
+          route = route_with_store
+
+          expect do
+            route.handle_request({ headers: {}, params: { 'collection_name' => 'projects', 'id' => '4' } })
+          end.to raise_error(Http::Exceptions::NotFoundError)
+
+          expect(store).not_to have_received(:list_by_record)
+        end
+
+        it 'still serves the history of a deleted record, which is much of the point of an audit trail' do
+          allow(permissions).to receive(:get_scope).and_return(Nodes::ConditionTreeLeaf.new('id', Operators::EQUAL, 4))
+          allow(collection).to receive(:list).and_return([])
+          route = route_with_store(records: [double('entry', to_h: { operation: 'delete', record_id: '4' })])
+
+          result = route.handle_request({ headers: {}, params: { 'collection_name' => 'projects', 'id' => '4' } })
+
+          expect(result[:content][:data]).to eq([{ 'operation' => 'delete', 'recordId' => '4' }])
+        end
+
+        it 'defaults to newest-first and switches to oldest-first on sort=timestamp' do
+          route = route_with_store
+          route.handle_request({ headers: {}, params: { 'collection_name' => 'projects', 'id' => '4', 'sort' => 'timestamp' } })
+
+          expect(store).to have_received(:list_by_record).with(hash_including(order: 'asc'))
+        end
+
+        it 'caps page[size] at 100 and honors page[number]' do
+          route = route_with_store
+          route.handle_request({ headers: {},
+                                 params: { 'collection_name' => 'projects', 'id' => '4',
+                                           'page' => { 'size' => '500', 'number' => '3' } } })
+
+          expect(store).to have_received(:list_by_record).with(hash_including(skip: 200, limit: 100))
+        end
+
+        it 'falls back to the default page when page is not a hash' do
+          route = route_with_store
+          route.handle_request({ headers: {}, params: { 'collection_name' => 'projects', 'id' => '4', 'page' => 'foo' } })
+
+          expect(store).to have_received(:list_by_record).with(hash_including(skip: 0, limit: 20))
+        end
+
+        it 'parses userIds, dropping non-numeric tokens' do
+          route = route_with_store
+          route.handle_request({ headers: {},
+                                 params: { 'collection_name' => 'projects', 'id' => '4', 'userIds' => '7, x ,9' } })
+
+          expect(store).to have_received(:list_by_record).with(hash_including(user_ids: [7, 9]))
+        end
+
+        it 'parses a date range into inclusive UTC boundaries' do
+          route = route_with_store
+          route.handle_request({ headers: {},
+                                 params: { 'collection_name' => 'projects', 'id' => '4',
+                                           'startDate' => '2026-01-02', 'endDate' => '2026-01-02' } })
+
+          expect(store).to have_received(:list_by_record).with(
+            hash_including(start_timestamp: '2026-01-02T00:00:00.000Z',
+                           end_timestamp: '2026-01-02T23:59:59.999Z')
+          )
+        end
+
+        it 'reads dates as local time in the request timezone' do
+          route = route_with_store
+          route.handle_request({ headers: {},
+                                 params: { 'collection_name' => 'projects', 'id' => '4',
+                                           'timezone' => 'America/New_York', 'startDate' => '2026-01-02' } })
+
+          # 2026-01-02 00:00 in New York (UTC-5) is 05:00 UTC.
+          expect(store).to have_received(:list_by_record).with(hash_including(start_timestamp: '2026-01-02T05:00:00.000Z'))
+        end
+
+        it 'rejects an unparsable date' do
+          route = route_with_store
+
+          expect do
+            route.handle_request({ headers: {},
+                                   params: { 'collection_name' => 'projects', 'id' => '4', 'startDate' => 'nope' } })
+          end.to raise_error(Http::Exceptions::ValidationError, /Invalid date/)
+        end
+
+        it 'rejects an unknown timezone' do
+          route = route_with_store
+
+          expect do
+            route.handle_request({ headers: {},
+                                   params: { 'collection_name' => 'projects', 'id' => '4',
+                                             'timezone' => 'Mars/Phobos', 'startDate' => '2026-01-02' } })
+          end.to raise_error(Http::Exceptions::ValidationError, /Invalid timezone/)
+        end
+      end
+    end
+  end
+end
diff --git a/packages/forest_admin_agent/spec/spec_helper.rb b/packages/forest_admin_agent/spec/spec_helper.rb
index d7148dd8c..686c44ed5 100644
--- a/packages/forest_admin_agent/spec/spec_helper.rb
+++ b/packages/forest_admin_agent/spec/spec_helper.rb
@@ -1,4 +1,6 @@
 require 'filecache'
+require 'active_record'
+require 'sqlite3'
 require 'simplecov'
 require 'simplecov_json_formatter'
 require 'simplecov-html'
diff --git a/packages/forest_admin_datasource_toolkit/lib/forest_admin_datasource_toolkit/components/caller.rb b/packages/forest_admin_datasource_toolkit/lib/forest_admin_datasource_toolkit/components/caller.rb
index 183e6b5ea..1d6cad012 100644
--- a/packages/forest_admin_datasource_toolkit/lib/forest_admin_datasource_toolkit/components/caller.rb
+++ b/packages/forest_admin_datasource_toolkit/lib/forest_admin_datasource_toolkit/components/caller.rb
@@ -1,7 +1,8 @@
 module ForestAdminDatasourceToolkit
   module Components
     class Caller
-      attr_reader :id, :email, :first_name, :last_name, :tags, :team, :rendering_id, :timezone, :permission_level, :role
+      attr_reader :id, :email, :first_name, :last_name, :tags, :team, :rendering_id, :timezone,
+                  :permission_level, :role, :request, :request_id
 
       def initialize(
         id:,
@@ -15,6 +16,7 @@ def initialize(
         permission_level:,
         role: nil,
         request: {},
+        request_id: nil,
         project: nil,
         environment: nil,
         **_extra_args
@@ -30,6 +32,7 @@ def initialize(
         @permission_level = permission_level
         @role = role
         @request = request
+        @request_id = request_id
         @project = project
         @environment = environment
       end
diff --git a/packages/forest_admin_rails/lib/forest_admin_rails.rb b/packages/forest_admin_rails/lib/forest_admin_rails.rb
index 2606f91be..8d91b5c9b 100644
--- a/packages/forest_admin_rails/lib/forest_admin_rails.rb
+++ b/packages/forest_admin_rails/lib/forest_admin_rails.rb
@@ -34,6 +34,9 @@ module ForestAdminRails
   setting :disable_route_cache, default: false
   setting :rpc_max_polling_threads, default: nil
   setting :workflow_executor_url, default: nil
+  # { database: <ActiveRecord config or URL>, schema:, table_name:, redact: } — setting `database`
+  # turns the audit trail on: every change is captured and the `/_audit-trail` routes are registered.
+  setting :audit_trail, default: nil
 
   if defined?(Rails::Railtie)
     # logic for cors middleware,... here // or it might be into Engine
diff --git a/packages/forest_admin_rails/lib/forest_admin_rails/engine.rb b/packages/forest_admin_rails/lib/forest_admin_rails/engine.rb
index 829667caf..5d82dfa42 100644
--- a/packages/forest_admin_rails/lib/forest_admin_rails/engine.rb
+++ b/packages/forest_admin_rails/lib/forest_admin_rails/engine.rb
@@ -34,9 +34,16 @@ class Engine < ::Rails::Engine
         agent_factory.setup(ForestAdminRails.config)
         load_configuration
         load_cors
+        load_correlation_id
       end
     end
 
+    # Echo the agent-generated correlation id on every response (mirrors the Node agent's
+    # router.use(correlationIdMiddleware)); CORS exposure of the header is handled in load_cors.
+    def load_correlation_id
+      config.middleware.use ForestAdminAgent::Http::CorrelationIdMiddleware
+    end
+
     def load_configuration
       return unless running_web_server?
       return unless create_agent_file_exists?
@@ -102,7 +109,8 @@ def load_cors
           hostnames += ENV['CORS_ORIGINS'].split(',') if ENV['CORS_ORIGINS']
 
           origins hostnames
-          resource '*', headers: :any, methods: :any, credentials: true, max_age: 86_400
+          resource '*', headers: :any, methods: :any, credentials: true, max_age: 86_400,
+                        expose: [ForestAdminAgent::Http::CorrelationId::HEADER]
         end
       end
     end