diff --git a/.changeset/mcp-aggregate-records.md b/.changeset/mcp-aggregate-records.md new file mode 100644 index 0000000000..f904a4e9f0 --- /dev/null +++ b/.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). diff --git a/content/docs/ai/natural-language-queries.mdx b/content/docs/ai/natural-language-queries.mdx index 970e79607e..7d9f3aea4a 100644 --- a/content/docs/ai/natural-language-queries.mdx +++ b/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. -**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). 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 diff --git a/packages/mcp/src/mcp-http-tools.scopes.test.ts b/packages/mcp/src/mcp-http-tools.scopes.test.ts index 9706a43e5b..37ef1806da 100644 --- a/packages/mcp/src/mcp-http-tools.scopes.test.ts +++ b/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 }; }, diff --git a/packages/mcp/src/mcp-http-tools.ts b/packages/mcp/src/mcp-http-tools.ts index cb5e63873a..ca595b4057 100644 --- a/packages/mcp/src/mcp-http-tools.ts +++ b/packages/mcp/src/mcp-http-tools.ts @@ -56,6 +56,22 @@ export interface McpDataBridge { create(object: string, data: Record): Promise; update(object: string, id: string, data: Record): Promise; remove(object: string, id: string): Promise; + /** + * 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; + groupBy?: Array; + aggregations: Array<{ function: string; field?: string; alias: string; distinct?: boolean }>; + timezone?: string; + }, + ): Promise; } 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', { diff --git a/packages/mcp/src/mcp-server-runtime.http.test.ts b/packages/mcp/src/mcp-server-runtime.http.test.ts index 4d3a98eec4..45a7f34bc9 100644 --- a/packages/mcp/src/mcp-server-runtime.http.test.ts +++ b/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'); + }); +}); diff --git a/packages/mcp/src/mcp-server-runtime.ts b/packages/mcp/src/mcp-server-runtime.ts index 63227f9f41..e8da5dad5d 100644 --- a/packages/mcp/src/mcp-server-runtime.ts +++ b/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', ]); diff --git a/packages/plugins/plugin-security/src/security-plugin.test.ts b/packages/plugins/plugin-security/src/security-plugin.test.ts index 1aae460437..c0ef924981 100644 --- a/packages/plugins/plugin-security/src/security-plugin.test.ts +++ b/packages/plugins/plugin-security/src/security-plugin.test.ts @@ -878,6 +878,81 @@ describe('SecurityPlugin', () => { expect(opCtx.result.salary).toBe(100); }); + // FLS aggregate-input gate (2.5b): result masking never runs for `aggregate` + // (output rows carry only aliases), so a protected field's statistics would + // leak straight through `sum(ssn) AS x` — the gate rejects on the INPUT. + it('FLS aggregate — aggregating an unreadable field is denied fail-closed', async () => { + const plugin = new SecurityPlugin({ fallbackPermissionSet: 'member_default' }); + const harness = makeMiddlewareCtx({ + permissionSets: [flsPolicySet], + objectFields: ['id', 'owner_id', 'name', 'salary', 'ssn'], + }); + await plugin.init(harness.ctx); + await plugin.start(harness.ctx); + const opCtx: any = { + object: 'task', + operation: 'aggregate', + ast: { + object: 'task', + aggregations: [{ function: 'count_distinct', field: 'ssn', alias: 'n' }], + }, + context: { userId: 'u1', tenantId: 'org-1', positions: [], permissions: ['member_default'] }, + }; + await expect(harness.run(opCtx)).rejects.toThrow(/Field read denied/); + await expect(harness.run(opCtx)).rejects.toMatchObject({ + details: { forbiddenFields: ['ssn'] }, + }); + }); + + it('FLS aggregate — grouping BY an unreadable field is denied (group keys reveal values)', async () => { + const plugin = new SecurityPlugin({ fallbackPermissionSet: 'member_default' }); + const harness = makeMiddlewareCtx({ + permissionSets: [flsPolicySet], + objectFields: ['id', 'owner_id', 'name', 'salary', 'ssn'], + }); + await plugin.init(harness.ctx); + await plugin.start(harness.ctx); + const opCtx: any = { + object: 'task', + operation: 'aggregate', + ast: { + object: 'task', + // Structured groupBy item — the gate must see through {field} objects. + groupBy: [{ field: 'ssn', dateGranularity: undefined }], + aggregations: [{ function: 'count', alias: 'n' }], + }, + context: { userId: 'u1', tenantId: 'org-1', positions: [], permissions: ['member_default'] }, + }; + await expect(harness.run(opCtx)).rejects.toMatchObject({ + details: { forbiddenFields: ['ssn'] }, + }); + }); + + it('FLS aggregate — readable fields aggregate fine (and count(*) needs no field)', async () => { + const plugin = new SecurityPlugin({ fallbackPermissionSet: 'member_default' }); + const harness = makeMiddlewareCtx({ + permissionSets: [flsPolicySet], + objectFields: ['id', 'owner_id', 'name', 'salary', 'ssn'], + }); + await plugin.init(harness.ctx); + await plugin.start(harness.ctx); + const opCtx: any = { + object: 'task', + operation: 'aggregate', + ast: { + object: 'task', + groupBy: ['name'], + // salary is readable:true (only editable:false) → aggregating it is a READ, allowed. + aggregations: [ + { function: 'sum', field: 'salary', alias: 'total' }, + { function: 'count', alias: 'n' }, + ], + }, + context: { userId: 'u1', tenantId: 'org-1', positions: [], permissions: ['member_default'] }, + }; + await expect(harness.run(opCtx)).resolves.toBeDefined(); + }); + it('fails CLOSED when permission resolution throws — denies, never bypasses (P0-2)', async () => { const plugin = new SecurityPlugin({ fallbackPermissionSet: 'member_default' }); const harness = makeMiddlewareCtx({ permissionSets: [tenantPolicySet] }); diff --git a/packages/plugins/plugin-security/src/security-plugin.ts b/packages/plugins/plugin-security/src/security-plugin.ts index 757a75e2bb..178963753c 100644 --- a/packages/plugins/plugin-security/src/security-plugin.ts +++ b/packages/plugins/plugin-security/src/security-plugin.ts @@ -958,6 +958,58 @@ export class SecurityPlugin implements Plugin { } } + // 2.5b. Field-Level Security READ enforcement for aggregate inputs. + // + // The read path relies on RESULT masking (step 4) to hide FLS-protected + // fields, but step 4 only covers find/findOne/insert/update — and an + // aggregate's output rows carry only aliases, so masking could never + // recover which source field fed `sum(salary) AS total`. Without an + // input-side gate a caller may read a protected field's statistics + // (sum/avg/min/max reveal the value outright on a single-row group). + // Enforce on the INPUT: any groupBy / aggregation reference to an + // FLS-unreadable field is rejected fail-closed with the offending names + // (mirrors the write gate in 2.5). `where`-filter probing is a + // platform-wide class shared with find() and is not widened here. + if (opCtx.operation === 'aggregate' && permissionSets.length > 0) { + let fieldPerms = this.permissionEvaluator.getFieldPermissions(opCtx.object, permissionSets); + // [ADR-0066 D3] AND-gate field-level requiredPermissions into the map. + fieldPerms = this.foldFieldRequiredPermissions(fieldPerms, secMeta.fieldRequiredPermissions, permissionSets); + // [ADR-0090 D10] Intersect with the delegator's field perms — a field + // the agent may read but the delegator may not stays forbidden. + if (delegatorSets) { + let delFieldPerms = this.permissionEvaluator.getFieldPermissions(opCtx.object, delegatorSets); + delFieldPerms = this.foldFieldRequiredPermissions(delFieldPerms, secMeta.fieldRequiredPermissions, delegatorSets); + fieldPerms = intersectFieldMasks(fieldPerms, delFieldPerms); + } + if (Object.keys(fieldPerms).length > 0) { + const ast: any = opCtx.ast ?? {}; + const referenced = new Set(); + for (const g of Array.isArray(ast.groupBy) ? ast.groupBy : []) { + const f = typeof g === 'string' ? g : g?.field; + if (typeof f === 'string' && f) referenced.add(f); + } + for (const a of Array.isArray(ast.aggregations) ? ast.aggregations : []) { + if (typeof a?.field === 'string' && a.field) referenced.add(a.field); + } + const forbidden = [...referenced].filter( + (f) => fieldPerms[f] && fieldPerms[f].readable === false, + ); + if (forbidden.length > 0) { + throw new PermissionDeniedError( + `[Security] Field read denied: not permitted to aggregate ` + + `[${forbidden.join(', ')}] on '${opCtx.object}'`, + { + operation: opCtx.operation, + object: opCtx.object, + positions, + permissionSets: explicitPermissionSets, + forbiddenFields: forbidden, + }, + ); + } + } + } + // 3.5. Auto-inject `owner_id` on insert from the // ExecutionContext. Without this, the row has `owner_id = NULL` // and the default `owner_only_writes` RLS policy hides it from diff --git a/packages/runtime/src/api-exposure.test.ts b/packages/runtime/src/api-exposure.test.ts index e1d7cc73ec..1e1de82886 100644 --- a/packages/runtime/src/api-exposure.test.ts +++ b/packages/runtime/src/api-exposure.test.ts @@ -41,6 +41,15 @@ describe('checkApiExposure (#1889)', () => { expect(checkApiExposure(ro, 'find').allowed).toBe(true); // find → list }); + it('gates aggregate as a list-class read', () => { + // An object whose whitelist excludes `list` must not leak row + // statistics through GROUP BY either. + expect(checkApiExposure({ apiMethods: ['list'] }, 'aggregate').allowed).toBe(true); + const d = checkApiExposure({ apiMethods: ['get'] }, 'aggregate'); + expect(d.allowed).toBe(false); + expect(d.status).toBe(405); + }); + it('an empty whitelist is treated as no restriction', () => { expect(checkApiExposure({ apiMethods: [] }, 'create').allowed).toBe(true); }); diff --git a/packages/runtime/src/api-exposure.ts b/packages/runtime/src/api-exposure.ts index 6bc79751c9..d74d9167f0 100644 --- a/packages/runtime/src/api-exposure.ts +++ b/packages/runtime/src/api-exposure.ts @@ -42,6 +42,9 @@ const ACTION_TO_API_METHOD: Record = { delete: 'delete', query: 'list', find: 'list', + // Aggregation is a list-class read: an object whose whitelist excludes + // `list` must not leak row statistics through GROUP BY either. + aggregate: 'list', batch: 'bulk', }; diff --git a/packages/runtime/src/http-dispatcher.ts b/packages/runtime/src/http-dispatcher.ts index 61ed2137b6..74994a70ac 100644 --- a/packages/runtime/src/http-dispatcher.ts +++ b/packages/runtime/src/http-dispatcher.ts @@ -412,6 +412,39 @@ export class HttpDispatcher { throw { statusCode: 503, message: 'Data service not available' }; } + if (action === 'aggregate') { + // Aggregate MUST run through the ObjectQL ENGINE (never the raw + // `dataDriver` the MCP bridge threads through for the other verbs): + // only the engine's middleware chain injects RLS/tenant scoping and + // the FLS aggregate-input gate. A raw driver.aggregate() would + // evaluate the query verbatim over every row. + // + // At least one aggregation is REQUIRED: with neither aggregations + // nor groupBy the engine's in-memory path degrades to raw rows, + // and the FLS result masker does not cover the `aggregate` op — + // grouped/aggregated output must stay the only thing this action + // can ever return. + if (!Array.isArray(params.aggregations) || params.aggregations.length === 0) { + throw { statusCode: 400, message: 'aggregate requires at least one aggregation' }; + } + const engine = (await this.getObjectQLService(scopeId)) + ?? await this.resolveService('objectql', scopeId).catch(() => null); + if (engine && typeof engine.aggregate === 'function') { + const rows = await engine.aggregate( + params.object, + { + ...(params.where ? { where: params.where } : {}), + ...(params.groupBy ? { groupBy: params.groupBy } : {}), + ...(params.aggregations ? { aggregations: params.aggregations } : {}), + ...(params.timezone ? { timezone: params.timezone } : {}), + ...(executionContext ? { context: executionContext } : {}), + }, + ); + return { object: params.object, rows: rows ?? [] }; + } + throw { statusCode: 503, message: 'Data service not available' }; + } + if (action === 'batch') { // Batch operations — not yet supported via direct service dispatch return { object: params.object, results: [] }; @@ -739,6 +772,26 @@ export class HttpDispatcher { const res: any = await callData('get', { object, id }, driver, envId, ec); return res?.record ?? res ?? null; }, + aggregate: async (object: string, o: any) => { + // NOTE: `driver` (the raw per-env db driver) is deliberately NOT + // passed — callData's aggregate branch resolves the ObjectQL + // engine itself so the security middleware (RLS + FLS aggregate + // gate) always runs. See the branch comment in callData. + const res: any = await callData( + 'aggregate', + { + object, + where: o?.where, + groupBy: o?.groupBy, + aggregations: o?.aggregations, + timezone: o?.timezone, + }, + undefined, + envId, + ec, + ); + return res?.rows ?? []; + }, create: async (object: string, data: any) => await callData('create', { object, data }, driver, envId, ec), update: async (object: string, id: string, data: any) =>