diff --git a/packages/agent/src/audit-trail/README.md b/packages/agent/src/audit-trail/README.md
index b9edc60f90..e2ff2e19a0 100644
--- a/packages/agent/src/audit-trail/README.md
+++ b/packages/agent/src/audit-trail/README.md
@@ -214,6 +214,13 @@ That test only runs when the snapshot can actually answer it. The capture keeps
 a field stored redacted — has no honest answer in the snapshot and the values are withheld rather
 than matched against a missing key: absent is not the same as passing.
 
+A captured `null` is tested the way the database tests a `NULL`: it answers only a condition asking
+for the null itself (`Blank`, `Missing`, `Equal` null, an `In` list holding null), never a negated or
+ordered one. In memory `status != 'private'` holds for a null status and `null < 5` coerces to
+`0 < 5`, while the scoped read that guarded the live record left that record out — without this, a
+record the caller could never read alive would become readable once deleted. On a datasource whose
+own `!=` keeps a NULL (Mongo's `$ne`), this is stricter than the live read, never looser.
+
 Primary keys are the exception, read back from the row's own id — but only for the side that id
 speaks for. A row is filed under the identity the record ended up with, so its id answers for a
 `create`, a `delete` and the new side of an `update`, never for what an update moved away from. A
@@ -292,9 +299,13 @@ ever written, so the real value was never in the database to find.
 Nor can a search confirm a value the scope withholding hides. On a record gone for good under a
 caller's permission scope, `search` and `fields` are matched against the values as served, never as
 captured — in SQL, which rows come back, `meta.count` and `availableUsers` would each say whether a
-withheld value holds the term. The rows are read without those two filters, in batches of 500, and
-matched and paged in memory, keeping only the requested page and the authors. That holds too for a
-record deleted while the request was in flight, whose SQL-matched answer is discarded and re-read.
+withheld value holds the term. The rows are read without those two filters, in batches of 500 that
+each continue past the last row read, never at an offset, and bounded at the instant the scan
+starts. They are matched and paged as they go, keeping only the page asked for. Only a gone
+record's history pays that scan, and it is one record's history — the same rows the SQL search
+would have scanned without an index. That holds too for a record deleted while the request was in flight:
+the second read of the record decides the withholding, so the SQL-matched answer is discarded and
+the history scanned the same way.
 
 Matching the *serialized* text rather than a structural walk of the parsed value is cheap and still
 correct for "keys and scalar values" — but two things follow from it. A punctuation-only term (`,`,
diff --git a/packages/agent/src/audit-trail/sql-store.ts b/packages/agent/src/audit-trail/sql-store.ts
index fe588cf8a3..7e5d47554e 100644
--- a/packages/agent/src/audit-trail/sql-store.ts
+++ b/packages/agent/src/audit-trail/sql-store.ts
@@ -199,6 +199,8 @@ function buildHistoryWhereClause(
     endTimestamp,
     fields,
     search,
+    order,
+    after,
   }: AuditHistoryQuery,
   sequelize: Sequelize,
 ): Record<string | symbol, unknown> {
@@ -215,6 +217,16 @@ function buildHistoryWhereClause(
   const andConditions = [];
   if (fields?.length) andConditions.push(fieldsChangedCondition(sequelize, fields));
   if (search) andConditions.push(searchCondition(sequelize, search));
+
+  if (after) {
+    const past = order === 'desc' ? Op.lt : Op.gt;
+    const at = new Date(after.timestamp);
+
+    andConditions.push({
+      [Op.or]: [{ timestamp: { [past]: at } }, { timestamp: at, id: { [past]: after.id } }],
+    });
+  }
+
   if (andConditions.length) where[Op.and] = andConditions;
 
   return where;
diff --git a/packages/agent/src/audit-trail/types.ts b/packages/agent/src/audit-trail/types.ts
index 2850fd2272..04a504aaf6 100644
--- a/packages/agent/src/audit-trail/types.ts
+++ b/packages/agent/src/audit-trail/types.ts
@@ -71,6 +71,11 @@ export type AuditHistoryQuery = {
   search?: string;
   /** Sort direction on `timestamp` (ties broken by insertion order). Defaults to `'asc'`. */
   order?: 'asc' | 'desc';
+  /**
+   * Keep only the rows strictly past this one in `order`, so a scan keyed on the last row it read
+   * neither repeats nor skips one when entries are written between its reads, as `skip` would.
+   */
+  after?: Pick<AuditRecord, 'timestamp' | 'id'>;
 };
 
 export type AuditCorrelationQuery = {
diff --git a/packages/agent/src/audit-trail/withhold.ts b/packages/agent/src/audit-trail/withhold.ts
index 60ac08d489..1bea817644 100644
--- a/packages/agent/src/audit-trail/withhold.ts
+++ b/packages/agent/src/audit-trail/withhold.ts
@@ -1,5 +1,11 @@
 import type { AuditRecord } from './types';
-import type { Collection, ConditionTree, Logger } from '@forestadmin/datasource-toolkit';
+import type {
+  Collection,
+  ConditionTree,
+  ConditionTreeBranch,
+  ConditionTreeLeaf,
+  Logger,
+} from '@forestadmin/datasource-toolkit';
 
 import { SchemaUtils } from '@forestadmin/datasource-toolkit';
 
@@ -23,6 +29,50 @@ export type Withholding = {
  */
 type IdAnswersForSide = boolean;
 
+function asksForNull({ operator, value }: ConditionTreeLeaf): boolean {
+  switch (operator) {
+    case 'Blank':
+    case 'Missing':
+      return true;
+    case 'Equal':
+      return value === null || value === undefined;
+    case 'In':
+      return Array.isArray(value) && value.some(item => item === null || item === undefined);
+    default:
+      return false;
+  }
+}
+
+// `ConditionTree.match`, except that a null answers only a condition asking for the null itself, as
+// the database tests it. In memory `status != 'private'` holds for a null status and `null < 5`
+// coerces to `0 < 5`, while the scoped read that guarded the live record left that NULL out: a record
+// the caller could never read alive would become readable once deleted. Stricter than a datasource
+// that matches NULL there (Mongo's `$ne`), never looser.
+function matchesAsStored(
+  tree: ConditionTree,
+  values: Record<string, unknown>,
+  collection: Collection,
+  timezone: string,
+): boolean {
+  // By shape, not `instanceof`: a scope can be built by another copy of the toolkit.
+  if ('aggregator' in tree) {
+    const branch = tree as ConditionTreeBranch;
+    const evaluate = (condition: ConditionTree) =>
+      matchesAsStored(condition, values, collection, timezone);
+
+    return branch.aggregator === 'And'
+      ? branch.conditions.every(evaluate)
+      : branch.conditions.some(evaluate);
+  }
+
+  const leaf = tree as ConditionTreeLeaf;
+  const value = values[leaf.field];
+
+  return value === null || value === undefined
+    ? asksForNull(leaf)
+    : leaf.match(values, collection, timezone);
+}
+
 // Only a snapshot that answers every field the permission scope asks about is worth matching. The
 // capture keeps the writable columns, so a permission scope reaching for anything else — a
 // read-only column, a relation — reads `undefined` there and would answer for a value the row never
@@ -43,7 +93,7 @@ export function permissionScopeAccepts(
     Object.prototype.hasOwnProperty.call(values, field),
   );
 
-  return answered && permissionScope.match(values, collection, timezone);
+  return answered && matchesAsStored(permissionScope, values, collection, timezone);
 }
 
 /**
diff --git a/packages/agent/src/routes/access/audit-trail.ts b/packages/agent/src/routes/access/audit-trail.ts
index 5fddecb640..a2b206c168 100644
--- a/packages/agent/src/routes/access/audit-trail.ts
+++ b/packages/agent/src/routes/access/audit-trail.ts
@@ -38,7 +38,7 @@ const ISO_INSTANT = /[Zz]$|[+-]\d{2}:?\d{2}$/;
 
 const DEFAULT_PAGE_SIZE = 20;
 const MAX_PAGE_SIZE = 100;
-const SERVED_MATCH_BATCH_SIZE = 500;
+const SCAN_BATCH_SIZE = 500;
 
 const AUDIT_OPERATIONS: readonly AuditOperation[] = [
   'create',
@@ -57,14 +57,6 @@ type AuditHistoryFilters = {
   search?: string;
 };
 
-type ServedMatchQuery = Pick<AuditHistoryFilters, 'fields' | 'search'> & {
-  rowFilters: AuditHistoryQuery;
-  order: 'asc' | 'desc';
-  skip: number;
-  limit: number;
-  isFirstFetch: boolean;
-};
-
 export default class AuditTrailRoute extends CollectionRoute {
   setupRoutes(router: Router): void {
     router.get(`/_audit-trail/${this.collectionUrlSlug}/:id`, this.handleHistory.bind(this));
@@ -110,18 +102,28 @@ export default class AuditTrailRoute extends CollectionRoute {
     const isFirstFetch =
       (context.request.query as Record<string, unknown>)['page[number]'] === undefined;
 
-    // Matched in SQL, `search` and `fields` test the values as captured, so which rows come back,
-    // the count and the authors would still say what the withholding below hides — one probe per
-    // character. For a gone record they are matched against the values served instead.
     const filtersOnValues = Boolean(fields || search);
-    const servedMatchQuery = { rowFilters, order, fields, search, skip, limit, isFirstFetch };
 
-    if (permissionScope && filtersOnValues && goneEntirely) {
-      context.response.body = await this.listServedMatches(
-        servedMatchQuery,
-        permissionScope,
-        context,
-      );
+    const serveMatchedValues = async (scope: ConditionTree) => {
+      const matched = await this.scanServedValues(context, scope, rowFilters, {
+        fields,
+        search,
+        order,
+        skip,
+        limit,
+      });
+
+      context.response.body = {
+        data: matched.page,
+        meta: {
+          count: matched.count,
+          ...(isFirstFetch && { availableUsers: [...matched.authors.values()] }),
+        },
+      };
+    };
+
+    if (permissionScope && goneEntirely && filtersOnValues) {
+      await serveMatchedValues(permissionScope);
 
       return;
     }
@@ -154,13 +156,10 @@ export default class AuditTrailRoute extends CollectionRoute {
 
     const gone = after ? after.goneEntirely : goneEntirely;
 
-    // Gone between the check and the read: the rows, count and authors above were matched in SQL.
-    if (permissionScope && filtersOnValues && gone) {
-      context.response.body = await this.listServedMatches(
-        servedMatchQuery,
-        permissionScope,
-        context,
-      );
+    // Gone between the check and the read: this answer withholds, so the rows, count and authors
+    // matched in SQL above must not decide what is served either.
+    if (permissionScope && gone && filtersOnValues) {
+      await serveMatchedValues(permissionScope);
 
       return;
     }
@@ -179,56 +178,67 @@ export default class AuditTrailRoute extends CollectionRoute {
     };
   }
 
-  // Read in batches so a gone record's history never sits in memory whole: only the requested
-  // page and the distinct authors are kept while the count runs over every match.
-  private async listServedMatches(
-    { rowFilters, order, fields, search, skip, limit, isFirstFetch }: ServedMatchQuery,
-    permissionScope: ConditionTree,
+  // Matched in SQL, `search` and `fields` test the values as captured, so which rows come back, the
+  // count and the authors would still say what the withholding hides — one probe per character. For
+  // a gone record they are matched against the values served instead, which means
+  // scanning the whole history: in batches, keeping only the page asked for, each continuing past
+  // the last row read rather than at an offset that entries written in between would shift, and
+  // bounded at the instant the scan starts so an id taken since cannot keep it chasing new rows.
+  private async scanServedValues(
     context: Context,
-  ): Promise<{
-    data: AuditRecord[];
-    meta: { count: number; availableUsers?: AuditUserSummary[] };
-  }> {
+    permissionScope: ConditionTree,
+    rowFilters: Omit<AuditHistoryQuery, 'fields' | 'search' | 'skip' | 'limit' | 'order' | 'after'>,
+    {
+      fields,
+      search,
+      order,
+      skip,
+      limit,
+    }: AuditHistoryFilters & { order: 'asc' | 'desc'; skip: number; limit: number },
+  ): Promise<{ page: AuditRecord[]; count: number; authors: Map<number, AuditUserSummary> }> {
     const { store } = this.options.auditTrail;
+    const now = new Date().toISOString();
+    const endTimestamp =
+      rowFilters.endTimestamp && rowFilters.endTimestamp < now ? rowFilters.endTimestamp : now;
     const page: AuditRecord[] = [];
     const authors = new Map<number, AuditUserSummary>();
     let count = 0;
+    let after: AuditRecord | undefined;
+    let rows: AuditRecord[];
 
-    for (let offset = 0; ; offset += SERVED_MATCH_BATCH_SIZE) {
-      // Sequential on purpose: batches are held one at a time.
+    do {
       // eslint-disable-next-line no-await-in-loop
-      const batch = await store.listByRecord({
+      rows = await store.listByRecord({
         ...rowFilters,
+        endTimestamp,
         order,
-        skip: offset,
-        limit: SERVED_MATCH_BATCH_SIZE,
+        limit: SCAN_BATCH_SIZE,
+        ...(after && { after: { timestamp: after.timestamp, id: after.id } }),
       });
 
-      const matched = this.withhold(batch, permissionScope, context).filter(entry =>
+      const matched = this.withhold(rows, permissionScope, context).filter(entry =>
         AuditTrailRoute.matchesServedValues(entry, fields, search),
       );
-      const pageStart = Math.max(0, skip - count);
-
-      page.push(...matched.slice(pageStart, pageStart + limit - page.length));
-      matched.forEach(entry => {
-        if (authors.has(entry.userId)) return;
-
-        authors.set(entry.userId, {
-          id: entry.userId,
-          firstName: entry.userFirstName,
-          lastName: entry.userLastName,
-          email: entry.userEmail,
-        });
-      });
-      count += matched.length;
 
-      if (batch.length < SERVED_MATCH_BATCH_SIZE) break;
-    }
+      for (const entry of matched) {
+        if (count >= skip && page.length < limit) page.push(entry);
+        count += 1;
+
+        // An author reads as their latest identity whichever way the history is sorted.
+        if (order === 'asc' || !authors.has(entry.userId)) {
+          authors.set(entry.userId, {
+            id: entry.userId,
+            firstName: entry.userFirstName,
+            lastName: entry.userLastName,
+            email: entry.userEmail,
+          });
+        }
+      }
 
-    return {
-      data: page,
-      meta: { count, ...(isFirstFetch && { availableUsers: [...authors.values()] }) },
-    };
+      after = rows[rows.length - 1];
+    } while (rows.length === SCAN_BATCH_SIZE);
+
+    return { page, count, authors };
   }
 
   // The SQL store's `fieldsChangedCondition` and `searchCondition`, run on the values as served.
diff --git a/packages/agent/test/audit-trail/in-memory-store.test.ts b/packages/agent/test/audit-trail/in-memory-store.test.ts
index 112c1f37a8..286dd4f60f 100644
--- a/packages/agent/test/audit-trail/in-memory-store.test.ts
+++ b/packages/agent/test/audit-trail/in-memory-store.test.ts
@@ -153,6 +153,32 @@ describe('InMemoryAuditStore', () => {
     expect(page2.map(r => r.newValues)).toEqual([{ n: 3 }]);
   });
 
+  describe('listByRecord after a cursor', () => {
+    const history = { collection: 'accounts', recordId: '1' };
+    const seedTied = (store: InMemoryAuditStore) =>
+      [1, 2, 3].map(id =>
+        store.seed(record({ id, ...(id === 3 && { timestamp: '2026-01-02T00:00:00.000Z' }) })),
+      );
+
+    test('continues past the cursor oldest first, breaking a timestamp tie by id', () => {
+      const store = new InMemoryAuditStore();
+      const [first] = seedTied(store);
+
+      const rows = store.listByRecord({ ...history, order: 'asc', after: first });
+
+      expect(rows.map(row => row.id)).toEqual([2, 3]);
+    });
+
+    test('continues past the cursor newest first, breaking a timestamp tie by id', () => {
+      const store = new InMemoryAuditStore();
+      const [, second] = seedTied(store);
+
+      const rows = store.listByRecord({ ...history, order: 'desc', after: second });
+
+      expect(rows.map(row => row.id)).toEqual([1]);
+    });
+  });
+
   describe('countByRecord', () => {
     it('counts all matching entries, ignoring skip and limit', () => {
       const store = new InMemoryAuditStore();
diff --git a/packages/agent/test/audit-trail/in-memory-store.ts b/packages/agent/test/audit-trail/in-memory-store.ts
index 43b943d007..a271d07ddb 100644
--- a/packages/agent/test/audit-trail/in-memory-store.ts
+++ b/packages/agent/test/audit-trail/in-memory-store.ts
@@ -115,17 +115,19 @@ export default class InMemoryAuditStore implements AuditStore {
     startTimestamp,
     endTimestamp,
     order = 'asc',
+    after,
   }: AuditHistoryQuery): AuditRecord[] {
     const direction = order === 'desc' ? -1 : 1;
+    const compare = (a: Pick<AuditRecord, 'timestamp' | 'id'>, b: typeof a) =>
+      direction * (a.timestamp.localeCompare(b.timestamp) || a.id - b.id);
 
-    // Ties on equal timestamps fall back to insertion order (the stable sort keeps it), which is
-    // the in-memory equivalent of the SQL store's auto-increment id — deterministic and chronological.
     return this.records
       .filter(record => record.collection === collection && record.recordId === recordId)
       .filter(record => !userIds || userIds.includes(record.userId))
       .filter(record => !operations?.length || operations.includes(record.operation))
       .filter(record => !startTimestamp || record.timestamp >= startTimestamp)
       .filter(record => !endTimestamp || record.timestamp <= endTimestamp)
-      .sort((a, b) => direction * a.timestamp.localeCompare(b.timestamp));
+      .filter(record => !after || compare(record, after) > 0)
+      .sort(compare);
   }
 }
diff --git a/packages/agent/test/audit-trail/sql-store.test.ts b/packages/agent/test/audit-trail/sql-store.test.ts
index 1d2adda1ec..ed5d23e729 100644
--- a/packages/agent/test/audit-trail/sql-store.test.ts
+++ b/packages/agent/test/audit-trail/sql-store.test.ts
@@ -521,6 +521,28 @@ describe('createSqlAuditStore (sqlite round-trip)', () => {
     await close();
   });
 
+  // A cursor survives writes between reads that would shift an offset: nothing repeats, nothing is
+  // skipped.
+  it('continues past a given row in either order, whatever was written since', async () => {
+    const { store, close } = createSqlAuditStore({ connectionString: 'sqlite::memory:' });
+    const query = { collection: 'accounts', recordId: '1' };
+
+    await seed(store, record({ timestamp: '2026-01-01T00:00:00.000Z', correlationKey: 'a' }));
+    await seed(store, record({ timestamp: '2026-01-02T00:00:00.000Z', correlationKey: 'b' }));
+    await seed(store, record({ timestamp: '2026-01-01T00:00:00.000Z', correlationKey: 'a2' }));
+
+    const newest = await store.listByRecord({ ...query, order: 'desc', limit: 2 });
+    await seed(store, record({ timestamp: '2026-01-03T00:00:00.000Z', correlationKey: 'late' }));
+    const rest = await store.listByRecord({ ...query, order: 'desc', after: newest[1] });
+    const oldest = await store.listByRecord({ ...query, limit: 1 });
+    const later = await store.listByRecord({ ...query, after: oldest[0] });
+
+    expect([...newest, ...rest].map(r => r.correlationKey)).toEqual(['b', 'a2', 'a']);
+    expect([...oldest, ...later].map(r => r.correlationKey)).toEqual(['a', 'a2', 'b', 'late']);
+
+    await close();
+  });
+
   it('returns rows recorded under a correlationKey for a record, scoped and oldest first', async () => {
     const { store, close } = createSqlAuditStore({ connectionString: 'sqlite::memory:' });
 
diff --git a/packages/agent/test/routes/access/audit-trail.test.ts b/packages/agent/test/routes/access/audit-trail.test.ts
index 57b44951d4..0b3e4b61f7 100644
--- a/packages/agent/test/routes/access/audit-trail.test.ts
+++ b/packages/agent/test/routes/access/audit-trail.test.ts
@@ -1,4 +1,4 @@
-import { ConditionTreeLeaf } from '@forestadmin/datasource-toolkit';
+import { ConditionTreeBranch, ConditionTreeLeaf } from '@forestadmin/datasource-toolkit';
 import { createMockContext } from '@shopify/jest-koa-mocks';
 
 import { REDACTED } from '../../../src/audit-trail';
@@ -843,17 +843,22 @@ describe('AuditTrailRoute', () => {
     // Matched in SQL, a search would still answer what the withholding hides: whether the row comes
     // back, and the count, say whether the withheld value holds the term.
     describe('a genuinely gone, scoped record searched or filtered by field', () => {
-      const secretDelete = (userEmail = 'jane@acme.io') => ({
+      const secretDelete = (over: Record<string, unknown> = {}) => ({
+        id: 1,
+        timestamp: '2026-01-01T00:00:00.000Z',
         operation: 'delete',
         recordId: '2',
         userId: 7,
         userFirstName: null,
         userLastName: null,
-        userEmail,
+        userEmail: 'jane@acme.io',
         actionName: null,
         previousValues: { ownerId: 2, title: 'Secret' },
         newValues: {},
+        ...over,
       });
+      const kept = (over: Record<string, unknown> = {}) =>
+        secretDelete({ previousValues: { ownerId: 1, title: 'Mine' }, ...over });
 
       const searched = async (history: unknown[], query: Record<string, string>) => {
         const { services, dataSource, options, store } = setup(history);
@@ -879,18 +884,16 @@ describe('AuditTrailRoute', () => {
       });
 
       test('matches the values it serves', async () => {
-        const kept = { ...secretDelete(), previousValues: { ownerId: 1, title: 'Mine' } };
+        const { body } = await searched([kept()], { search: 'MINE' });
 
-        const { body } = await searched([kept], { search: 'MINE' });
-
-        expect(body.data).toEqual([kept]);
+        expect(body.data).toEqual([kept()]);
       });
 
       test('still matches what stays visible on a withheld row, such as its author', async () => {
         const { body } = await searched([secretDelete()], { search: 'acme' });
 
         expect(body).toEqual({
-          data: [{ ...secretDelete(), previousValues: {} }],
+          data: [secretDelete({ previousValues: {} })],
           meta: {
             count: 1,
             availableUsers: [{ id: 7, firstName: null, lastName: null, email: 'jane@acme.io' }],
@@ -899,9 +902,10 @@ describe('AuditTrailRoute', () => {
       });
 
       test('lists an author once even when their rows carry different identities', async () => {
-        const { body } = await searched([secretDelete(), secretDelete('jane@acme.com')], {
-          search: 'acme',
-        });
+        const { body } = await searched(
+          [secretDelete(), secretDelete({ id: 2, userEmail: 'jane@acme.com' })],
+          { search: 'acme' },
+        );
 
         expect(body.meta).toEqual({
           count: 2,
@@ -909,66 +913,48 @@ describe('AuditTrailRoute', () => {
         });
       });
 
+      test('lists an author as their latest identity when sorted oldest first', async () => {
+        const { body } = await searched(
+          [secretDelete(), secretDelete({ id: 2, userEmail: 'jane@acme.com' })],
+          { search: 'acme', sort: 'timestamp' },
+        );
+
+        expect(body.meta).toEqual({
+          count: 2,
+          availableUsers: [{ id: 7, firstName: null, lastName: null, email: 'jane@acme.com' }],
+        });
+      });
+
       test('does not match a field only a withheld side touched', async () => {
         const { body } = await searched([secretDelete()], { fields: 'title' });
 
         expect(body.data).toEqual([]);
       });
 
-      test('reads the rows without the value filters and pages what matched', async () => {
-        const kept = { ...secretDelete(), previousValues: { ownerId: 1, title: 'Mine' } };
+      test('reads the rows without the value filters, bounded at the instant the scan starts', async () => {
+        const before = new Date().toISOString();
 
-        const { body, store } = await searched([kept, { ...kept, operation: 'update' }], {
-          search: 'mine',
-          userIds: '12',
-          'page[size]': '1',
-          'page[number]': '2',
-        });
+        const { store } = await searched([kept()], { search: 'mine', userIds: '12' });
 
-        expect(store.listByRecord).toHaveBeenCalledWith({
+        const [query] = (store.listByRecord as jest.Mock).mock.calls[0];
+        expect(query).toEqual({
           collection: 'books',
           recordId: '2',
           userIds: [12],
           order: 'desc',
-          skip: 0,
           limit: 500,
+          endTimestamp: expect.any(String),
         });
+        expect(query.endTimestamp >= before).toBe(true);
         expect(store.countByRecord).not.toHaveBeenCalled();
-        expect(body).toEqual({ data: [{ ...kept, operation: 'update' }], meta: { count: 2 } });
       });
 
-      test('reads the history in batches and counts every match across them', async () => {
-        const kept = { ...secretDelete(), previousValues: { ownerId: 1, title: 'Mine' } };
-        const { services, dataSource, options, store } = setup();
-        store.listByRecord
-          .mockResolvedValueOnce(Array.from({ length: 500 }, () => kept))
-          .mockResolvedValueOnce([kept]);
-        (services.authorization.getScope as jest.Mock).mockResolvedValue(
-          new ConditionTreeLeaf('ownerId', 'Equal', 1),
-        );
-        jest.spyOn(dataSource.getCollection('books'), 'list').mockResolvedValue([]);
-        const route = new AuditTrailRoute(services, options, dataSource, 'books');
-        const context = createMockContext({
-          state: { user: { email: 'john.doe@domain.com' } },
-          customProperties: {
-            query: { timezone: 'Europe/Paris', search: 'mine', 'page[size]': '1' },
-            params: { id: '2' },
-          },
-        });
-
-        await route.handleHistory(context);
+      test('keeps an end date the caller asked for when it is earlier', async () => {
+        const { store } = await searched([], { search: 'mine', endDate: '2020-01-01' });
 
-        expect(store.listByRecord).toHaveBeenNthCalledWith(
-          2,
-          expect.objectContaining({ skip: 500, limit: 500 }),
+        expect(store.listByRecord).toHaveBeenCalledWith(
+          expect.objectContaining({ endTimestamp: expect.stringMatching(/^2020-01-01T/) }),
         );
-        expect(context.response.body).toEqual({
-          data: [kept],
-          meta: {
-            count: 501,
-            availableUsers: [{ id: 7, firstName: null, lastName: null, email: 'jane@acme.io' }],
-          },
-        });
       });
 
       test('matches served values for a record deleted while the audit read was in flight', async () => {
@@ -995,14 +981,47 @@ describe('AuditTrailRoute', () => {
           collection: 'books',
           recordId: '2',
           order: 'desc',
-          skip: 0,
           limit: 500,
+          endTimestamp: expect.any(String),
         });
         expect(context.response.body).toEqual({
           data: [],
           meta: { count: 0, availableUsers: [] },
         });
       });
+
+      // The page cap bounds what is served, not what is scanned: the history is read in batches,
+      // each continuing past the last row read, so a long one is never held in memory whole.
+      test('scans in batches past the last row read, counting across them and keeping only the page', async () => {
+        const rows = Array.from({ length: 1001 }, (_, index) => kept({ id: index + 1 }));
+        const { services, dataSource, options, store } = setup();
+        store.listByRecord.mockImplementation(
+          async ({ after, limit }: { after?: { id: number }; limit: number }) =>
+            rows.filter(row => !after || row.id > after.id).slice(0, limit),
+        );
+        (services.authorization.getScope as jest.Mock).mockResolvedValue(
+          new ConditionTreeLeaf('ownerId', 'Equal', 1),
+        );
+        jest.spyOn(dataSource.getCollection('books'), 'list').mockResolvedValue([]);
+        const route = new AuditTrailRoute(services, options, dataSource, 'books');
+        const context = createMockContext({
+          state: { user: { email: 'john.doe@domain.com' } },
+          customProperties: {
+            query: { search: 'mine', 'page[size]': '2', 'page[number]': '251' },
+            params: { id: '2' },
+          },
+        });
+
+        await route.handleHistory(context);
+
+        const calls = (store.listByRecord as jest.Mock).mock.calls.map(([query]) => query);
+        expect(calls.map(query => query.after?.id)).toEqual([undefined, 500, 1000]);
+        expect(calls.every(query => query.skip === undefined)).toBe(true);
+        expect(context.response.body).toEqual({
+          data: [rows[500], rows[501]],
+          meta: { count: 1001 },
+        });
+      });
     });
 
     test("keeps a genuinely-deleted delete row's previousValues when they match the caller's scope", async () => {
@@ -1340,6 +1359,102 @@ describe('AuditTrailRoute', () => {
         return (context.response.body as { data: unknown[] }).data;
       };
 
+      // In memory `!=` holds for a null and `null < 5` coerces to `0 < 5`, while the database leaves a
+      // NULL out of both: a record the caller could never read alive must not become readable once
+      // deleted.
+      describe('a captured null', () => {
+        const nullSecret = [
+          {
+            operation: 'delete',
+            recordId: '2',
+            previousValues: { ownerId: null, secret: null },
+            newValues: {},
+          },
+        ];
+        const withheld = [
+          { operation: 'delete', recordId: '2', previousValues: {}, newValues: {} },
+        ];
+
+        test('withholds the values under a negated scope, as the database would have left the record out', async () => {
+          const notEqual = await historyUnder(
+            new ConditionTreeLeaf('secret', 'NotEqual', 'private'),
+            nullSecret,
+          );
+          const notIn = await historyUnder(
+            new ConditionTreeLeaf('secret', 'NotIn', ['private']),
+            nullSecret,
+          );
+
+          expect([notEqual, notIn]).toEqual([withheld, withheld]);
+        });
+
+        test('withholds the values under an ordered scope a null would coerce through', async () => {
+          const data = await historyUnder(
+            new ConditionTreeLeaf('ownerId', 'LessThan', 5),
+            nullSecret,
+          );
+
+          expect(data).toEqual(withheld);
+        });
+
+        test('withholds when the negation sits inside a branch', async () => {
+          const scope = new ConditionTreeBranch('And', [
+            new ConditionTreeLeaf('ownerId', 'Equal', 1),
+            new ConditionTreeLeaf('secret', 'NotEqual', 'private'),
+          ]);
+
+          const data = await historyUnder(scope as unknown as ConditionTreeLeaf, [
+            { operation: 'delete', recordId: '2', previousValues: { ownerId: 1, secret: null } },
+          ]);
+
+          expect(data).toEqual(withheld);
+        });
+
+        test('keeps the values a scope asking for the null itself covers', async () => {
+          const missing = await historyUnder(
+            new ConditionTreeLeaf('secret', 'Missing'),
+            nullSecret,
+          );
+          const equalNull = await historyUnder(
+            new ConditionTreeLeaf('secret', 'Equal', null),
+            nullSecret,
+          );
+          const inWithNull = await historyUnder(
+            new ConditionTreeLeaf('secret', 'In', ['open', null]),
+            nullSecret,
+          );
+
+          expect([missing, equalNull, inWithNull]).toEqual([nullSecret, nullSecret, nullSecret]);
+        });
+
+        test('withholds the values under an In list that does not hold null', async () => {
+          const data = await historyUnder(
+            new ConditionTreeLeaf('secret', 'In', ['open']),
+            nullSecret,
+          );
+
+          expect(data).toEqual(withheld);
+        });
+
+        test('keeps a non-null value the negation covers', async () => {
+          const history = [
+            {
+              operation: 'delete',
+              recordId: '2',
+              previousValues: { ownerId: 1, secret: 'open' },
+              newValues: {},
+            },
+          ];
+
+          const data = await historyUnder(
+            new ConditionTreeLeaf('secret', 'NotEqual', 'private'),
+            history,
+          );
+
+          expect(data).toEqual(history);
+        });
+      });
+
       test('withholds a delete row when the scope reads a column the snapshot never captured', async () => {
         const data = await historyUnder(new ConditionTreeLeaf('status', 'NotEqual', 'private'), [
           { operation: 'delete', recordId: '2', previousValues: { ownerId: 1, secret: 'shh' } },
@@ -1972,11 +2087,11 @@ describe('AuditTrailRoute', () => {
     // fail; this route is those same values reassembled, so it has to answer the same way or the
     // withheld values are one request away.
     describe('a genuinely gone record, read by a scoped caller', () => {
-      const reconstructFor = async (scope: ConditionTreeLeaf) => {
+      const reconstructFor = async (scope: ConditionTreeLeaf, status: string | null = 'closed') => {
         const history = [
           {
             operation: 'delete',
-            previousValues: { id: 2, status: 'closed', name: 'Acme' },
+            previousValues: { id: 2, status, name: 'Acme' },
             newValues: {},
           },
         ];
@@ -2006,6 +2121,15 @@ describe('AuditTrailRoute', () => {
         expect(context.response.body).toEqual({ data: null });
       });
 
+      test('withholds a reconstruction whose NULL the scope negates, as the database would have refused the record', async () => {
+        const context = await reconstructFor(
+          new ConditionTreeLeaf('status', 'NotEqual', 'private'),
+          null,
+        );
+
+        expect(context.response.body).toEqual({ data: null });
+      });
+
       test('serves the reconstruction when it matches the scope', async () => {
         const context = await reconstructFor(new ConditionTreeLeaf('status', 'Equal', 'closed'));