@@ -29,8 +29,10 @@
 // `aggregation-filter-json-column-refusal.test.ts`.
 
 import { describe, it, expect, vi } from 'vitest';
+import { lowerFilterCondition } from '@objectstack/spec/data';
 import { ObjectQL } from './engine.js';
 import { declaredJsonStoredFields, matchesAggregationFilter } from './having-filter.js';
+import { applyInMemoryAggregation } from './in-memory-aggregation.js';
 
 const OBJECT = 'os21007_doc';
 
@@ -249,3 +251,86 @@ describe('[#21007] the per-row floor — a caller evaluating rows directly meets
     expect(matchesAggregationFilter(ROWS[0], { owners: { $in: ['u1'] } } as never, 0)).toBe(false);
   });
 });
+
+describe('[#21007] applyInMemoryAggregation — a direct caller with a field map meets the same refusal, before any row is judged', () => {
+  // `applyInMemoryAggregation` is published (objectql's root and `./core`), so a
+  // host can reach the walker with a field map and no engine in front of it.
+  // It calls the engine's own gate once per filter; without that call the
+  // per-row backstop answered both cells below 200 — an empty row set has no
+  // row to reach it, and on the LOWERED filter (the shape every engine seam
+  // hands the evaluator: spec `lowerFilterCondition`, rule 3) a row with no
+  // value is decided by the `$null` arm before the comparison is walked.
+  const NEGATIONS: ReadonlyArray<readonly [string, Record<string, unknown>, string]> = [
+    ['$ne', { meta: { $ne: 'a' } }, '$ne'],
+    ['$nin', { meta: { $nin: ['a', 'b'] } }, '$nin'],
+    ['$not $in', { $not: { meta: { $in: ['a'] } } }, '$in'],
+  ];
+  const NULL_META_ROWS = ROWS.map((row) => ({ ...row, meta: null }));
+  const ast = (filter: unknown, groupBy?: string[]) => ({
+    ...(groupBy ? { groupBy } : {}),
+    aggregations: [
+      { function: 'count' as const, alias: 'n' },
+      { function: 'count' as const, alias: 'm', filter: filter as never },
+    ],
+  });
+
+  function refusalFrom(run: () => unknown): Error & { code?: string; status?: number } {
+    try {
+      run();
+    } catch (e) {
+      return e as Error & { code?: string; status?: number };
+    }
+    throw new Error('expected applyInMemoryAggregation to refuse this filter, but it answered');
+  }
+
+  async function engineMessageFor(filter: unknown): Promise<string> {
+    const { engine } = await makeEngine([]);
+    return (await refusalOf(() => engine.aggregate(OBJECT, perAggregation(filter)))).message;
+  }
+
+  for (const [name, filter, op] of NEGATIONS) {
+    const lowered = lowerFilterCondition(filter);
+
+    it(`${name} — the lowered filter carries the $null arm the walker short-circuits on`, () => {
+      expect(JSON.stringify(lowered)).toContain('"$null"');
+    });
+
+    const cells: ReadonlyArray<{ cell: string; rows: Record<string, unknown>[]; spelled: unknown; groupBy?: string[] }> = [
+      { cell: 'an EMPTY row set', rows: [], spelled: filter },
+      { cell: 'an EMPTY row set, grouped', rows: [], spelled: filter, groupBy: ['title'] },
+      { cell: 'meta null in every row, the lowered filter', rows: NULL_META_ROWS, spelled: lowered },
+      { cell: 'meta null in every row, the filter as written', rows: NULL_META_ROWS, spelled: filter },
+    ];
+    for (const { cell, rows, spelled, groupBy } of cells) {
+      it(`${name} on ${cell}: 400 INVALID_FILTER, engine.aggregate's very message, the diagnostic to reportWithheld`, async () => {
+        const reported: string[] = [];
+        const err = refusalFrom(() => applyInMemoryAggregation(
+          rows.map((row) => ({ ...row })), ast(spelled, groupBy), undefined, FIELDS,
+          (diagnostic) => { reported.push(diagnostic); },
+        ));
+        expect(err.code).toBe('INVALID_FILTER');
+        expect(err.status).toBe(400);
+        expect(err.message).toBe(await engineMessageFor(filter));
+        expect(err.message).toContain('{ "FIELD": { "$contains": "a" } }');
+        expect(err.message).not.toContain('"meta"');
+        expect(reported.join('\n')).toContain(`Operator "${op}" on field "meta" WAS NOT APPLIED`);
+        expect(reported.join('\n')).toContain('At aggregations[1].filter.');
+      });
+    }
+  }
+
+  it('no reportWithheld: the same refusal, the diagnostic dropped', () => {
+    const err = refusalFrom(() => applyInMemoryAggregation([], ast({ owners: { $in: ['u1'] } }), undefined, FIELDS));
+    expect(err.code).toBe('INVALID_FILTER');
+    expect(err.status).toBe(400);
+  });
+
+  it('no field map: nothing is judged, and the walker answers as it always did', () => {
+    expect(applyInMemoryAggregation(ROWS.map((row) => ({ ...row })), ast({ owners: { $in: ['u1'] } }))).toEqual([{ n: 6, m: 0 }]);
+  });
+
+  it('what still answers on a declared JSON-stored field answers here too', () => {
+    expect(applyInMemoryAggregation(ROWS.map((row) => ({ ...row })), ast({ owners: { $contains: 'u1' } }), undefined, FIELDS))
+      .toEqual([{ n: 6, m: 2 }]);
+  });
+});
@@ -1042,11 +1042,26 @@ export function assertAggregationFilterIsEvaluable(
  * here meet only declared operators. A plain object with no `$` key is not a
  * condition this rule reads; the engine's no-operator-object door refuses it
  * earlier. No usable field map (a registry-less host) ⇒ nothing is judged.
+ *
+ * THIS is the complete door. The per-row arm in {@link checkCondition} is a
+ * backstop that fires only on a row that reaches it, and many never do: an
+ * empty table has no row, and the spec's filter lowering (rule 3) rewrites a
+ * negation into `$or: [{ f: { $null: true } }, { f: { $ne: v } }]` (and gives a
+ * `$not` operand an `{ f: { $null: false } }` conjunct), so on a row with no
+ * value the `$null` arm decides and the walker never reaches the comparison.
+ * Measured with this call removed: a `json` field null in every row answered
+ * `$ne` / `$nin` / `$not $in` with every row counted. Hence [#21007]'s second
+ * caller, `applyInMemoryAggregation`, which a host may call with a field map
+ * and no engine in front of it — it calls this same function, once per
+ * aggregation, before any row is judged.
+ *
+ * `declared` needs only the field map and where the diagnostic goes; the
+ * object's name is the reference rule's, not this one's.
  */
-function assertAggregationFilterSparesJsonStoredFields(
+export function assertAggregationFilterSparesJsonStoredFields(
   filter: unknown,
   root: string,
-  declared: AggregationFilterDeclaration,
+  declared: Pick<AggregationFilterDeclaration, 'fields' | 'reportWithheld'>,
 ): void {
   const jsonStored = declaredJsonStoredFields(declared.fields);
   if (jsonStored.size === 0) return;
@@ -1637,11 +1652,16 @@ function storedArrayHasMember(value: unknown, comparand: unknown): boolean {
  *
  * [#21007] …and a scalar comparison on such a column — implicit equality, or an
  * operator in `JSON_COLUMN_INCOMPATIBLE_OPERATORS` — is REFUSED, in `where`'s
- * withheld words. The engine judges that once on the filter
- * ({@link assertAggregationFilterSparesJsonStoredFields}) before any row is
- * read, and reports the diagnostic there; this arm is the floor for a caller
- * that evaluates rows directly. Above the no-value exit, for the #7158 reason:
- * it is the filter's verdict, not the row's.
+ * withheld words. The COMPLETE door is
+ * {@link assertAggregationFilterSparesJsonStoredFields}, which judges the filter
+ * once, before any row is read, and reports the diagnostic; `engine.aggregate`
+ * and `applyInMemoryAggregation` both call it. This arm is only the BACKSTOP
+ * for a row that reaches it — a caller evaluating rows directly through
+ * `matchesAggregationFilter`. It sits above the no-value exit, so a row with no
+ * value that reaches it is refused too; but its REACH is the row's: an empty
+ * row set never gets here, and neither does a row on which a short-circuited
+ * `$or` / `$and` branch has already decided (the spec lowering's rule 3 puts a
+ * `$null` arm ahead of every negation).
  */
 function checkCondition(
   value: any,
@@ -1695,7 +1715,8 @@ function checkCondition(
       throw emptyFlagComparandError(field, target, `${path}.${op}`);
     }
     // [#21007] A scalar comparison on a declared JSON-stored column — refused,
-    // as `where` refuses it. The floor under the engine's one-time judgment.
+    // as `where` refuses it. The backstop under the one-time judgment
+    // (assertAggregationFilterSparesJsonStoredFields), for a row that gets here.
     if (jsonStored && JSON_COLUMN_INCOMPATIBLE_OPERATORS.has(op)) {
       throw invalidFilterError(jsonColumnOperatorRefusalText(field, op, false).message);
     }
 
@@ -86,7 +86,13 @@
 
 import { bucketDateKey, compensatedSum } from '@objectstack/core';
 import type { QueryAST, GroupByNode, AggregationNode, DateGranularityValue } from '@objectstack/spec/data';
-import { declaredFieldClasses, declaredJsonStoredFields, matchesAggregationFilter } from './having-filter.js';
+import {
+  aggregationFilterClause,
+  assertAggregationFilterSparesJsonStoredFields,
+  declaredFieldClasses,
+  declaredJsonStoredFields,
+  matchesAggregationFilter,
+} from './having-filter.js';
 
 /**
  * Group + aggregate raw rows according to the AST's `groupBy` /
@@ -106,18 +112,42 @@ import { declaredFieldClasses, declaredJsonStoredFields, matchesAggregationFilte
  * [#20873] …and `$contains` / `$notContains` on a declared JSON-stored field ask
  * MEMBERSHIP (having-filter.ts `declaredJsonStoredFields`), the reading the
  * same condition gets in a `where`. Absent ⇒ the substring reading, as before.
+ *
+ * [#21007] …and a scalar comparison on such a field (`$in`, `$nin`, `$eq`, an
+ * ordering, `$between`, implicit equality) is REFUSED `INVALID_FILTER` / 400,
+ * in the words `where` refuses it in — judged once per `aggregations[i].filter`
+ * by having-filter.ts `assertAggregationFilterSparesJsonStoredFields`, the same
+ * function `engine.aggregate` calls, BEFORE any row is judged. So an empty
+ * `rows` and a row set the per-row walk would short-circuit past refuse alike;
+ * the per-row arm is only the backstop. `reportWithheld` is where the withheld
+ * half of that refusal goes — the field, the operator and the position, which
+ * the message does not name. This entry point holds no logger of its own, so a
+ * host that wants the diagnostic passes its log here; absent ⇒ it is dropped,
+ * and the caller still gets the refusal's code, status and prescription.
+ * `engine.aggregate` has judged the filter already (and logged it) before it
+ * calls this, so it passes none.
  */
 export function applyInMemoryAggregation(
   rows: any[],
   ast: Pick<QueryAST, 'groupBy' | 'aggregations'>,
   timezone?: string,
   fields?: Record<string, unknown>,
+  reportWithheld?: (diagnostic: string) => void,
 ): any[] {
   const groupBy = (ast.groupBy ?? []) as GroupByNode[];
   const aggregations = (ast.aggregations ?? []) as AggregationNode[];
   if (groupBy.length === 0 && aggregations.length === 0) return rows;
   // [#20176] Read once per call, and only when some aggregation carries a filter.
   const anyFilter = aggregations.some((a) => a?.filter && Object.keys(a.filter).length > 0);
+  // [#21007] The JSON-column rule, judged on each FILTER before any row is —
+  // the complete door, not the per-row backstop. See the docblock above.
+  if (fields && anyFilter) {
+    const declared = { fields, reportWithheld: reportWithheld ?? (() => undefined) };
+    for (const [index, agg] of aggregations.entries()) {
+      if (!agg?.filter || Object.keys(agg.filter).length === 0) continue;
+      assertAggregationFilterSparesJsonStoredFields(agg.filter, aggregationFilterClause(index).root, declared);
+    }
+  }
   const filterClasses = fields && anyFilter ? declaredFieldClasses(fields) : undefined;
   // [#20873] Read once per call too, from the same declaration.
   const filterJsonStored = fields && anyFilter ? declaredJsonStoredFields(fields) : undefined;