feat(mcp): aggregate_records MCP tool over the engine read path (#2976)

* feat(mcp,runtime,plugin-security): aggregate_records MCP tool over the engine read path

Adds GROUP BY aggregation to the MCP surface so agents can answer
quantitative questions in one call instead of paging raw rows through
query_records (capped + token-expensive).

- mcp: new data:read tool aggregate_records (count/sum/avg/min/max/
  count_distinct, groupBy incl. {field,dateGranularity} date bucketing,
  where, timezone). Optional McpDataBridge.aggregate seam — runtimes that
  omit it simply don't register the tool. Output rows capped at the query
  limit with a truncated flag.
- runtime: callData('aggregate') branch resolves the ObjectQL ENGINE
  (never the raw per-env dataDriver) so the security middleware — RLS,
  tenant scoping, ADR-0090 D10 delegator intersection — always runs;
  requires >=1 aggregation (the engine's in-memory path would otherwise
  degrade to raw rows the FLS masker does not cover). aggregate maps to
  the 'list' ApiMethod in the exposure gate.
- plugin-security: FLS aggregate-INPUT gate (2.5b) — result masking never
  runs for aggregate and aliases hide field provenance, so groupBy /
  aggregation references to FLS-unreadable fields are rejected
  fail-closed with the offending names, including the D10 delegator
  field-mask intersection.

Tests: mcp 72 passed (6 new: registration/degradation, delegation,
sys_* guard, min(1) schema, truncation, FLS error surface); plugin-
security 426 passed (3 new FLS aggregate gate cases); runtime 507
passed (new aggregate exposure-mapping case).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012RE5oFswe4DUvExuAgvbg2

* docs(ai): open MCP surface now ships server-side aggregation (aggregate_records)

natural-language-queries.mdx stated the open MCP surface has no
server-side aggregate tool and a BYO agent must sum client-side — the
aggregate_records tool added in this branch makes that false. Update the
tool enumeration + the aggregation callout, and keep the distinction
from the cloud dataset-level aggregate_data tool (cross-object / cubes).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012RE5oFswe4DUvExuAgvbg2

---------

Co-authored-by: Claude <noreply@anthropic.com>

Author: os-zhuang

.changeset/mcp-aggregate-records.md

@@ -0,0 +1,29 @@
+---
+'@objectstack/mcp': minor
+'@objectstack/runtime': minor
+'@objectstack/plugin-security': patch
+---
+
+feat(mcp): `aggregate_records` tool — GROUP BY aggregation over the engine read path
+
+New MCP tool `aggregate_records` (count/sum/avg/min/max/count_distinct, optional
+groupBy incl. date bucketing, where filter, IANA timezone) in the `data:read`
+family. Execution routes through the ObjectQL ENGINE (`callData('aggregate')`
+deliberately never uses the raw per-env driver), so RLS/tenant scoping and the
+D10 delegator intersection apply exactly as on find.
+
+Security hardening shipped with it:
+
+- plugin-security: new FLS aggregate-INPUT gate — result masking never runs for
+  `aggregate` (output rows carry only aliases), so any groupBy / aggregation
+  reference to an FLS-unreadable field is now rejected fail-closed with the
+  offending field names (mirrors the FLS write gate).
+- runtime: `aggregate` maps to the `list` ApiMethod in the object exposure gate
+  (an object whose `apiMethods` whitelist excludes `list` cannot leak row
+  statistics through GROUP BY), and the aggregate action requires at least one
+  aggregation (the engine's in-memory path would otherwise degrade to raw rows
+  that the FLS masker does not cover).
+
+The bridge seam is optional: a runtime that does not implement
+`McpDataBridge.aggregate` simply does not register the tool (graceful
+degradation, same contract as the action tools).

content/docs/ai/natural-language-queries.mdx

@@ -1,6 +1,6 @@
 ---
 title: Natural Language Queries
-description: How agents query live data through the open MCP data tools (query_records, get_record) under RLS
+description: How agents query live data through the open MCP data tools (query_records, get_record, aggregate_records) under RLS
 ---
 
 # Natural Language Queries
@@ -15,19 +15,23 @@ back this are part of the open framework — no ObjectOS runtime and no
 cloud Studio are required.
 
 Your AI queries your data through the open MCP **data tools** — `query_records`
-(filter, field selection, sort, pagination) and `get_record`, plus the
-discovery tools `list_objects` and `describe_object` — which the model calls
-with structured arguments. These run as ordinary [ObjectQL](/docs/protocol/objectql) queries over
+(filter, field selection, sort, pagination), `get_record`, and `aggregate_records`
+(group-by count/sum/avg/min/max/count_distinct, with optional date bucketing),
+plus the discovery tools `list_objects` and `describe_object` — which the model
+calls with structured arguments. These run as ordinary [ObjectQL](/docs/protocol/objectql) queries over
 your objects (ObjectStack uses ObjectQL, not SOQL), and they execute under the
 caller's `ExecutionContext`, so row-level security applies exactly as it does
 for the REST API.
 
 <Callout type="info">
-**Aggregation is a cloud data tool.** The open MCP surface reads and filters
-rows; it does not ship a server-side `aggregate_data` (group-by / roll-up)
-tool — `query_records` has no aggregation arguments. A BYO agent aggregates by
-fetching the rows it's allowed to see and summing client-side; the built-in
-`aggregate_data` tool is part of the cloud in-product chat runtime (below).
+**Server-side aggregation is open too.** `aggregate_records` runs a single-object
+GROUP BY (count/sum/avg/min/max/count_distinct, optional date bucketing and
+filter) through the same ObjectQL engine read path — so totals and breakdowns
+come back in one call under row-level **and** field-level security, instead of
+the agent paging every visible row and summing client-side. It does not join
+across objects or read datasets/cubes: the cross-object, dataset-level
+`aggregate_data` roll-up tool remains part of the cloud in-product chat runtime
+(below).
 </Callout>
 
 The open MCP server plugin exposes these tools automatically — it bridges the
@@ -45,7 +49,7 @@ await kernel.bootstrap();
 
 Point any MCP client at the server and ask questions in natural language: the
 model discovers `list_objects` / `describe_object` / `query_records` /
-`get_record` and calls them under RLS as the authenticated caller.
+`get_record` / `aggregate_records` and calls them under RLS as the authenticated caller.
 
 There is no separate natural-language-to-query metadata type to author — the
 model is prompted with the available objects and translates the user's question
@@ -55,7 +59,7 @@ into `query_records` calls at runtime.
 **ObjectOS — bundled in-product chat.** ObjectOS ships an in-UI
 AI runtime — the `ask` data-query assistant
 and the `/api/v1/ai/*` chat endpoints — that registers these same data tools,
-plus the group-by/roll-up `aggregate_data` tool, into its own chat loop. That
+plus the dataset-level (cross-object) `aggregate_data` roll-up tool, into its own chat loop. That
 runtime is not part of the open framework and is documented separately in the
 [ObjectOS docs](https://docs.objectos.app/docs/ai); this page covers the open path
 only. On the open-source framework, use

packages/mcp/src/mcp-http-tools.scopes.test.ts

@@ -22,7 +22,7 @@ import {
 import { MCPServerRuntime } from './mcp-server-runtime.js';
 import type { McpActionBridge, McpDataBridge } from './mcp-http-tools.js';
 
-const READ_TOOLS = ['list_objects', 'describe_object', 'query_records', 'get_record'];
+const READ_TOOLS = ['list_objects', 'describe_object', 'query_records', 'get_record', 'aggregate_records'];
 const WRITE_TOOLS = ['create_record', 'update_record', 'delete_record'];
 const ACTION_TOOLS = ['list_actions', 'run_action'];
 
@@ -34,6 +34,7 @@ function makeBridge(): McpDataBridge & McpActionBridge & { calls: any[] } {
     async describeObject(name: string) { calls.push(['describeObject', name]); return { name }; },
     async query(object: string, opts: any) { calls.push(['query', object, opts]); return { object, records: [] }; },
     async get(object: string, id: string) { calls.push(['get', object, id]); return { id }; },
+    async aggregate(object: string, opts: any) { calls.push(['aggregate', object, opts]); return []; },
     async create(object: string, data: any) { calls.push(['create', object, data]); return { object, id: 'n1' }; },
     async update(object: string, id: string, data: any) { calls.push(['update', object, id, data]); return { object, id }; },
     async remove(object: string, id: string) { calls.push(['remove', object, id]); return { object, id, deleted: true }; },

packages/mcp/src/mcp-http-tools.ts

@@ -56,6 +56,22 @@ export interface McpDataBridge {
   create(object: string, data: Record<string, unknown>): Promise<unknown>;
   update(object: string, id: string, data: Record<string, unknown>): Promise<unknown>;
   remove(object: string, id: string): Promise<unknown>;
+  /**
+   * GROUP BY aggregation through the ObjectQL engine's read path (RLS + the
+   * FLS aggregate-input gate). OPTIONAL: a runtime that cannot route
+   * aggregation through the engine simply omits it and the
+   * `aggregate_records` tool is not registered (graceful degradation, same
+   * contract as {@link McpActionBridge}).
+   */
+  aggregate?(
+    object: string,
+    opts: {
+      where?: Record<string, unknown>;
+      groupBy?: Array<string | { field: string; dateGranularity?: string; alias?: string }>;
+      aggregations: Array<{ function: string; field?: string; alias: string; distinct?: boolean }>;
+      timezone?: string;
+    },
+  ): Promise<unknown[]>;
 }
 
 export interface RegisterObjectToolsOptions {
@@ -282,6 +298,82 @@ export function registerObjectTools(
       },
     );
 
+    if (typeof bridge.aggregate === 'function') {
+      const aggregateFn = bridge.aggregate.bind(bridge);
+      server.registerTool(
+        'aggregate_records',
+        {
+          description:
+            'Aggregate records with GROUP BY: count/sum/avg/min/max/count_distinct over an object, ' +
+            'optionally grouped by fields (dates can be bucketed by day/week/month/quarter/year). ' +
+            'Use this instead of paging query_records when a question needs totals or breakdowns. ' +
+            'Runs under the caller\'s permissions, row-level security and field-level security.',
+          inputSchema: {
+            objectName: z.string().describe('The object/table name'),
+            aggregations: z
+              .array(
+                z.object({
+                  function: z
+                    .enum(['count', 'sum', 'avg', 'min', 'max', 'count_distinct'])
+                    .describe('Aggregation function'),
+                  field: z.string().optional().describe('Field to aggregate (omit for count(*))'),
+                  alias: z.string().describe('Result column name'),
+                }),
+              )
+              .min(1)
+              .describe('Metrics to compute, e.g. [{"function":"sum","field":"amount","alias":"total"}]'),
+            groupBy: z
+              .array(
+                z.union([
+                  z.string(),
+                  z.object({
+                    field: z.string().describe('Field to group by'),
+                    dateGranularity: z
+                      .enum(['day', 'week', 'month', 'quarter', 'year'])
+                      .optional()
+                      .describe('Bucket a date field into uniform periods'),
+                    alias: z.string().optional().describe('Alias for the projected group value'),
+                  }),
+                ]),
+              )
+              .optional()
+              .describe('Grouping fields; omit for a single overall row'),
+            where: z
+              .record(z.string(), z.unknown())
+              .optional()
+              .describe('Filter conditions applied before aggregation, e.g. {"status":"open"}'),
+            timezone: z
+              .string()
+              .optional()
+              .describe('IANA timezone for date bucketing (defaults to UTC)'),
+          },
+          annotations: { readOnlyHint: true, destructiveHint: false, openWorldHint: false },
+        },
+        async ({ objectName, aggregations, groupBy, where, timezone }) => {
+          const bad = guard(objectName);
+          if (bad) return errorResult(bad);
+          try {
+            const rows = (await aggregateFn(objectName, {
+              where,
+              groupBy,
+              aggregations,
+              timezone,
+            })) ?? [];
+            // Group count is unbounded (a high-cardinality groupBy can return
+            // thousands of rows) — cap the tool output like query_records does.
+            const truncated = rows.length > maxLimit;
+            return textResult({
+              rows: truncated ? rows.slice(0, maxLimit) : rows,
+              totalGroups: rows.length,
+              ...(truncated ? { truncated: true } : {}),
+            });
+          } catch (err) {
+            return errorResult(messageOf(err));
+          }
+        },
+      );
+    }
+
     server.registerTool(
       'get_record',
       {

packages/mcp/src/mcp-server-runtime.http.test.ts

@@ -223,3 +223,164 @@ describe('MCPServerRuntime.handleHttpRequest (Streamable HTTP)', () => {
     expect(json.result.content[0].text).toContain('RLS: not permitted');
   });
 });
+
+describe('aggregate_records (GROUP BY aggregation tool)', () => {
+  let runtime: MCPServerRuntime;
+
+  /** Bridge WITH the optional aggregate seam implemented. */
+  function makeAggBridge(rows: unknown[] = [{ status: 'open', n: 2 }]) {
+    const base = makeBridge();
+    const calls = base.calls;
+    return {
+      ...base,
+      calls,
+      async aggregate(object: string, opts: any) {
+        calls.push(['aggregate', object, opts]);
+        return rows;
+      },
+    };
+  }
+
+  beforeEach(() => {
+    runtime = new MCPServerRuntime({ name: 'objectstack-test', version: '9.9.9' });
+  });
+
+  it('registers only when the bridge implements aggregate (graceful degradation)', async () => {
+    const without = await call(runtime, { jsonrpc: '2.0', id: 1, method: 'tools/list' }, makeBridge());
+    expect(without.json.result.tools.map((t: any) => t.name)).not.toContain('aggregate_records');
+
+    const runtime2 = new MCPServerRuntime({ name: 'objectstack-test', version: '9.9.9' });
+    const withAgg = await call(runtime2, { jsonrpc: '2.0', id: 1, method: 'tools/list' }, makeAggBridge());
+    const byName = Object.fromEntries(withAgg.json.result.tools.map((t: any) => [t.name, t]));
+    expect(byName.aggregate_records).toBeDefined();
+    expect(byName.aggregate_records.annotations.readOnlyHint).toBe(true);
+  });
+
+  it('delegates groupBy/aggregations/where to the bridge and returns rows', async () => {
+    const bridge = makeAggBridge([{ status: 'open', n: 2 }, { status: 'done', n: 5 }]);
+    const { json } = await call(
+      runtime,
+      {
+        jsonrpc: '2.0',
+        id: 2,
+        method: 'tools/call',
+        params: {
+          name: 'aggregate_records',
+          arguments: {
+            objectName: 'task',
+            groupBy: ['status', { field: 'created_at', dateGranularity: 'month' }],
+            aggregations: [{ function: 'count', alias: 'n' }],
+            where: { active: true },
+          },
+        },
+      },
+      bridge,
+    );
+    expect(json.result.isError).toBeFalsy();
+    const payload = JSON.parse(json.result.content[0].text);
+    expect(payload.rows).toHaveLength(2);
+    expect(payload.totalGroups).toBe(2);
+    expect(payload.truncated).toBeUndefined();
+    const aggCall = bridge.calls.find((c: any[]) => c[0] === 'aggregate');
+    expect(aggCall[1]).toBe('task');
+    expect(aggCall[2].groupBy).toEqual(['status', { field: 'created_at', dateGranularity: 'month' }]);
+    expect(aggCall[2].aggregations).toEqual([{ function: 'count', alias: 'n' }]);
+    expect(aggCall[2].where).toEqual({ active: true });
+  });
+
+  it('rejects system objects fail-closed without consulting the bridge', async () => {
+    const bridge = makeAggBridge();
+    const { json } = await call(
+      runtime,
+      {
+        jsonrpc: '2.0',
+        id: 3,
+        method: 'tools/call',
+        params: {
+          name: 'aggregate_records',
+          arguments: {
+            objectName: 'sys_user',
+            aggregations: [{ function: 'count', alias: 'n' }],
+          },
+        },
+      },
+      bridge,
+    );
+    expect(json.result.isError).toBe(true);
+    expect(json.result.content[0].text).toMatch(/system object/i);
+    expect(bridge.calls.find((c: any[]) => c[0] === 'aggregate')).toBeUndefined();
+  });
+
+  it('rejects an empty aggregations array (schema min(1))', async () => {
+    const bridge = makeAggBridge();
+    const { json } = await call(
+      runtime,
+      {
+        jsonrpc: '2.0',
+        id: 4,
+        method: 'tools/call',
+        params: {
+          name: 'aggregate_records',
+          arguments: { objectName: 'task', aggregations: [] },
+        },
+      },
+      bridge,
+    );
+    // Zod input validation fails before the handler runs — surfaced as a
+    // tool error or JSON-RPC error depending on SDK version; both are fine,
+    // the invariant is that the bridge is never reached.
+    expect(json.result?.isError === true || json.error != null).toBe(true);
+    expect(bridge.calls.find((c: any[]) => c[0] === 'aggregate')).toBeUndefined();
+  });
+
+  it('caps the returned group rows at the query limit with a truncated flag', async () => {
+    const manyGroups = Array.from({ length: 60 }, (_, i) => ({ k: `g${i}`, n: i }));
+    const bridge = makeAggBridge(manyGroups);
+    const { json } = await call(
+      runtime,
+      {
+        jsonrpc: '2.0',
+        id: 5,
+        method: 'tools/call',
+        params: {
+          name: 'aggregate_records',
+          arguments: {
+            objectName: 'task',
+            groupBy: ['k'],
+            aggregations: [{ function: 'count', alias: 'n' }],
+          },
+        },
+      },
+      bridge,
+    );
+    const payload = JSON.parse(json.result.content[0].text);
+    expect(payload.rows).toHaveLength(50); // DEFAULT_MAX_LIMIT
+    expect(payload.totalGroups).toBe(60);
+    expect(payload.truncated).toBe(true);
+  });
+
+  it('surfaces bridge errors (e.g. FLS denial) as tool errors', async () => {
+    const bridge = makeAggBridge();
+    bridge.aggregate = async () => {
+      throw new Error('[Security] Field read denied: not permitted to aggregate [salary]');
+    };
+    const { json } = await call(
+      runtime,
+      {
+        jsonrpc: '2.0',
+        id: 6,
+        method: 'tools/call',
+        params: {
+          name: 'aggregate_records',
+          arguments: {
+            objectName: 'employee',
+            aggregations: [{ function: 'sum', field: 'salary', alias: 'total' }],
+          },
+        },
+      },
+      bridge,
+    );
+    expect(json.result.isError).toBe(true);
+    expect(json.result.content[0].text).toContain('Field read denied');
+  });
+});

packages/mcp/src/mcp-server-runtime.ts

@@ -51,6 +51,7 @@ const READ_ONLY_TOOLS = new Set([
   'describe_object',
   'query_records',
   'get_record',
+  'aggregate_records',
   'aggregate_data',
 ]);