From 3d56f3b0dcd94912b96a47cbfb022815d13d1cb3 Mon Sep 17 00:00:00 2001 From: Kevin Messiaen <114553769+kevinmessiaen@users.noreply.github.com> Date: Mon, 6 Jul 2026 16:15:40 +0700 Subject: [PATCH 01/17] feat(mongodb): add aggregate query surface to the connector --- .../cli/src/connectors/mongodb/connector.ts | 45 +++++++++++++++++++ .../test/connectors/mongodb/connector.test.ts | 37 +++++++++++++++ 2 files changed, 82 insertions(+) diff --git a/packages/cli/src/connectors/mongodb/connector.ts b/packages/cli/src/connectors/mongodb/connector.ts index f46d6038e..123508354 100644 --- a/packages/cli/src/connectors/mongodb/connector.ts +++ b/packages/cli/src/connectors/mongodb/connector.ts @@ -39,6 +39,20 @@ export interface KtxMongoListedCollection { type?: string; } +export interface KtxMongoQueryInput { + connectionId: string; + collection: string; + database?: string; + pipeline: Record[]; + limit: number; +} + +export interface KtxMongoQueryResult { + headers: string[]; + rows: unknown[][]; + rowCount: number; +} + interface KtxMongoFindOptions { sort: Record; limit: number; @@ -50,6 +64,12 @@ export interface KtxMongoClient { listCollections(databaseName: string): Promise; estimatedDocumentCount(databaseName: string, collectionName: string): Promise; find(databaseName: string, collectionName: string, options: KtxMongoFindOptions): Promise; + aggregate( + databaseName: string, + collectionName: string, + pipeline: Record[], + options: { limit: number }, + ): Promise; ping(databaseName: string): Promise; close(): Promise; } @@ -106,6 +126,20 @@ class DefaultMongoClient implements KtxMongoClient { .toArray() as Promise; } + async aggregate( + databaseName: string, + collectionName: string, + pipeline: Record[], + options: { limit: number }, + ): Promise { + const client = await this.connectedClient(); + return client + .db(databaseName) + .collection(collectionName) + .aggregate([...pipeline, { $limit: options.limit }], { maxTimeMS: SAMPLE_MAX_TIME_MS }) + .toArray() as Promise; + } + async ping(databaseName: string): Promise { const client = await this.connectedClient(); await client.db(databaseName).command({ ping: 1 }); @@ -362,6 +396,17 @@ export class KtxMongoDbScanConnector implements KtxScanConnector { return { values, nullCount, distinctCount: null }; } + async executeQuery(input: KtxMongoQueryInput, _ctx: KtxScanContext): Promise { + this.assertConnection(input.connectionId); + const database = input.database ?? this.databases[0]!; + const documents = await this.clientForQuery().aggregate(database, input.collection, input.pipeline, { + limit: input.limit, + }); + const headers = unionDocumentKeys(documents); + const rows = documents.map((document) => headers.map((header) => normalizeSampleValue(document[header]))); + return { headers, rows, rowCount: rows.length }; + } + async listSchemas(): Promise { return [...this.databases]; } diff --git a/packages/cli/test/connectors/mongodb/connector.test.ts b/packages/cli/test/connectors/mongodb/connector.test.ts index 43d60013d..3fd250390 100644 --- a/packages/cli/test/connectors/mongodb/connector.test.ts +++ b/packages/cli/test/connectors/mongodb/connector.test.ts @@ -40,6 +40,7 @@ function fakeClientFactory(): { factory: KtxMongoClientFactory; client: KtxMongo collectionName === 'users' ? 2 : 1, ), find: vi.fn(async (_databaseName: string, collectionName: string) => DOCUMENTS[collectionName] ?? []), + aggregate: vi.fn(async (_databaseName: string, collectionName: string) => DOCUMENTS[collectionName] ?? []), ping: vi.fn(async () => undefined), close: vi.fn(async () => undefined), }; @@ -184,6 +185,7 @@ describe('KtxMongoDbScanConnector.introspect', () => { return 2; }), find: vi.fn(async (_db: string, name: string) => DOCUMENTS[name] ?? DOCUMENTS.users!), + aggregate: vi.fn(async (_db: string, name: string) => DOCUMENTS[name] ?? DOCUMENTS.users!), ping: vi.fn(async () => undefined), close: vi.fn(async () => undefined), }; @@ -239,3 +241,38 @@ describe('ktx sql against a MongoDB connection', () => { ).rejects.toThrow(/does not support read-only SQL execution/); }); }); + +describe('KtxMongoDbScanConnector.executeQuery', () => { + it('runs an aggregation pipeline and returns normalized rows keyed by the union of document fields', async () => { + const { factory, client } = fakeClientFactory(); + const result = await connector(baseConnection, factory).executeQuery( + { + connectionId: 'mongo-prod', + collection: 'users', + pipeline: [{ $match: { age: { $gte: 18 } } }], + limit: 50, + }, + { runId: 't' }, + ); + + expect(client.aggregate).toHaveBeenCalledWith('app', 'users', [{ $match: { age: { $gte: 18 } } }], { + limit: 50, + }); + expect(result.headers).toEqual(['_id', 'email', 'age', 'address']); + // _id is a BSON wrapper → String(...); address is a sub-document → JSON string. + expect(result.rows[0]).toEqual(['a1', 'a@x.com', 31, JSON.stringify({ city: 'NY' })]); + // Second document is missing age and address → null-filled to keep rows rectangular. + expect(result.rows[1]).toEqual(['a2', 'b@x.com', null, null]); + expect(result.rowCount).toBe(2); + }); + + it('rejects a query whose connection id does not match this connector', async () => { + const { factory } = fakeClientFactory(); + await expect( + connector(baseConnection, factory).executeQuery( + { connectionId: 'other', collection: 'users', pipeline: [], limit: 10 }, + { runId: 't' }, + ), + ).rejects.toThrow(/cannot serve connection other/); + }); +}); From 90e78171f1c4d5bb4c37fd27b62bf4a5dd3cf251 Mon Sep 17 00:00:00 2001 From: Kevin Messiaen <114553769+kevinmessiaen@users.noreply.github.com> Date: Mon, 6 Jul 2026 16:21:18 +0700 Subject: [PATCH 02/17] feat(mcp): declare mongoQuery context port contract --- packages/cli/src/context/mcp/types.ts | 15 +++++++++++++++ 1 file changed, 15 insertions(+) diff --git a/packages/cli/src/context/mcp/types.ts b/packages/cli/src/context/mcp/types.ts index c8dbd480b..36a861153 100644 --- a/packages/cli/src/context/mcp/types.ts +++ b/packages/cli/src/context/mcp/types.ts @@ -180,6 +180,20 @@ export interface KtxSqlExecutionMcpPort { ): Promise; } +export interface KtxMongoQueryResponse { + headers: string[]; + rows: unknown[][]; + rowCount: number; +} + +/** @internal */ +export interface KtxMongoQueryMcpPort { + execute( + input: { connectionId: string; collection: string; database?: string; pipeline: Record[]; limit: number }, + options?: { onProgress?: KtxMcpProgressCallback }, + ): Promise; +} + /** @internal */ export interface KtxDialectNotesMcpPort { read(input: { connectionId: string }): Promise<{ connectionId: string; dialect: string; notes: string }>; @@ -193,6 +207,7 @@ export interface KtxMcpContextPorts { dictionarySearch?: KtxDictionarySearchMcpPort; discover?: KtxDiscoverDataMcpPort; sqlExecution?: KtxSqlExecutionMcpPort; + mongoQuery?: KtxMongoQueryMcpPort; dialectNotes?: KtxDialectNotesMcpPort; memoryIngest?: MemoryIngestPort; } From 985399593042b5b2d8fa22a64b6fa14254a0a6c9 Mon Sep 17 00:00:00 2001 From: Kevin Messiaen <114553769+kevinmessiaen@users.noreply.github.com> Date: Mon, 6 Jul 2026 16:24:42 +0700 Subject: [PATCH 03/17] feat(mcp): build mongoQuery port over the shared connector factory --- .../src/context/mcp/local-project-ports.ts | 46 ++++++++++++- .../cli/test/mcp/mongo-query-tool.test.ts | 67 +++++++++++++++++++ 2 files changed, 112 insertions(+), 1 deletion(-) create mode 100644 packages/cli/test/mcp/mongo-query-tool.test.ts diff --git a/packages/cli/src/context/mcp/local-project-ports.ts b/packages/cli/src/context/mcp/local-project-ports.ts index 684c002ca..71aa04569 100644 --- a/packages/cli/src/context/mcp/local-project-ports.ts +++ b/packages/cli/src/context/mcp/local-project-ports.ts @@ -15,6 +15,7 @@ import { KtxDaemonComputeError, type KtxSemanticLayerComputePort } from '../../c import type { KtxLocalProject } from '../../context/project/project.js'; import { createKtxEntityDetailsService } from '../../context/scan/entity-details.js'; import type { LocalScanMcpOptions } from '../../context/scan/local-scan.js'; +import type { KtxScanConnector } from '../../context/scan/types.js'; import { createKtxDiscoverDataService } from '../../context/search/discover.js'; import { sqlAnalysisDialectForDriver } from '../../context/sql-analysis/dialect.js'; import type { SqlAnalysisPort } from '../../context/sql-analysis/ports.js'; @@ -24,7 +25,8 @@ import { readLocalSlSource } from '../../context/sl/local-sl.js'; import { assertSafeConnectionId } from '../../context/sl/source-files.js'; import { assertConfiguredConnectionId } from '../../context/connections/configured-connections.js'; import { readLocalKnowledgePage, searchLocalKnowledgePages } from '../wiki/local-knowledge.js'; -import type { KtxMcpContextPorts, KtxMcpProgressCallback, KtxSqlExecutionResponse } from './types.js'; +import type { KtxMongoDbScanConnector } from '../../connectors/mongodb/connector.js'; +import type { KtxMcpContextPorts, KtxMcpProgressCallback, KtxMongoQueryResponse, KtxSqlExecutionResponse } from './types.js'; interface CreateLocalProjectMcpContextPortsOptions { semanticLayerCompute?: KtxSemanticLayerComputePort; @@ -54,6 +56,35 @@ function throwClassifiedQueryError(error: unknown): never { throw new KtxQueryError(error instanceof Error ? error.message : String(error), { cause: error }); } +function projectHasMongoConnection(project: KtxLocalProject): boolean { + return Object.values(project.config.connections).some( + (connection) => normalizeConnectionDriver(connection) === 'mongodb', + ); +} + +async function executeMongoQuery( + createConnector: (connectionId: string) => Promise | KtxScanConnector, + input: { connectionId: string; collection: string; database?: string; pipeline: Record[]; limit: number }, +): Promise { + const connectionId = assertSafeConnectionId(input.connectionId); + let connector: KtxScanConnector | null = null; + try { + connector = await createConnector(connectionId); + if (connector.driver !== 'mongodb') { + throw new KtxExpectedError( + `Connection "${connectionId}" driver "${connector.driver}" is not a MongoDB connection; mongo_query serves mongodb connections only.`, + ); + } + const result = await (connector as unknown as KtxMongoDbScanConnector).executeQuery( + { connectionId, collection: input.collection, database: input.database, pipeline: input.pipeline, limit: input.limit }, + { runId: 'mcp-mongo-query' }, + ); + return { headers: result.headers, rows: result.rows, rowCount: result.rowCount }; + } finally { + await connector?.cleanup?.(); + } +} + async function executeValidatedReadOnlySql( project: KtxLocalProject, options: CreateLocalProjectMcpContextPortsOptions, @@ -239,5 +270,18 @@ export function createLocalProjectMcpContextPorts( }; } + const mongoCreateConnector = options.localScan?.createConnector; + if (mongoCreateConnector && projectHasMongoConnection(project)) { + ports.mongoQuery = { + async execute(input) { + try { + return await executeMongoQuery(mongoCreateConnector, input); + } catch (error) { + throwClassifiedQueryError(error); + } + }, + }; + } + return ports; } diff --git a/packages/cli/test/mcp/mongo-query-tool.test.ts b/packages/cli/test/mcp/mongo-query-tool.test.ts new file mode 100644 index 000000000..4d316baf9 --- /dev/null +++ b/packages/cli/test/mcp/mongo-query-tool.test.ts @@ -0,0 +1,67 @@ +import { describe, expect, it, vi } from 'vitest'; +import { KtxQueryError } from '../../src/errors.js'; +import { KtxMongoDbScanConnector } from '../../src/connectors/mongodb/connector.js'; +import { createLocalProjectMcpContextPorts } from '../../src/context/mcp/local-project-ports.js'; +import type { KtxLocalProject } from '../../src/context/project/project.js'; + +const MONGO_DOCS = [ + { _id: 'a1', city: 'Indianapolis' }, + { _id: 'a2', city: 'Indianapolis' }, +]; + +function mongoConnector(): KtxMongoDbScanConnector { + return new KtxMongoDbScanConnector({ + connectionId: 'mongo', + connection: { driver: 'mongodb', url: 'mongodb://localhost:27017/app', databases: ['app'] }, + clientFactory: { + create: () => ({ + listCollections: vi.fn(async () => []), + estimatedDocumentCount: vi.fn(async () => 0), + find: vi.fn(async () => []), + aggregate: vi.fn(async () => MONGO_DOCS), + ping: vi.fn(async () => undefined), + close: vi.fn(async () => undefined), + }), + }, + }); +} + +function project(): KtxLocalProject { + return { + projectDir: '/tmp/ktx', + config: { + connections: { + mongo: { driver: 'mongodb', url: 'mongodb://localhost:27017/app', databases: ['app'] }, + warehouse: { driver: 'postgres' }, + }, + }, + } as unknown as KtxLocalProject; +} + +function portsFor(createConnector: (id: string) => KtxMongoDbScanConnector) { + return createLocalProjectMcpContextPorts(project(), { + embeddingService: null, + localScan: { createConnector } as never, + }); +} + +describe('mongoQuery MCP port', () => { + it('fetches real rows from a mongodb connection', async () => { + const ports = portsFor(() => mongoConnector()); + const result = await ports.mongoQuery!.execute({ + connectionId: 'mongo', + collection: 'business', + pipeline: [{ $match: { city: 'Indianapolis' } }], + limit: 100, + }); + expect(result.headers).toEqual(['_id', 'city']); + expect(result.rowCount).toBe(2); + }); + + it('rejects a non-mongodb connection id as an expected query error', async () => { + const ports = portsFor(() => mongoConnector()); + await expect( + ports.mongoQuery!.execute({ connectionId: 'warehouse', collection: 'x', pipeline: [], limit: 10 }), + ).rejects.toBeInstanceOf(KtxQueryError); + }); +}); From 8aa4f8f74a03d5ca13d1194c29a6341228715cce Mon Sep 17 00:00:00 2001 From: Kevin Messiaen <114553769+kevinmessiaen@users.noreply.github.com> Date: Mon, 6 Jul 2026 16:30:56 +0700 Subject: [PATCH 04/17] fix(test): exercise mongoQuery wrong-driver guard, not assertConnection The wrong-driver test's fake createConnector ignored the requested id and always returned a mongodb connector, so the driver guard never ran; the rejection came incidentally from assertConnection deep in the connector. It also asserted KtxQueryError, but the guard throws KtxExpectedError. Dispatch the fake by id and assert the guard's own error type and message. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01BdUJSmjqGoksdc45ZZJiVY --- .../cli/test/mcp/mongo-query-tool.test.ts | 25 +++++++++++++++---- 1 file changed, 20 insertions(+), 5 deletions(-) diff --git a/packages/cli/test/mcp/mongo-query-tool.test.ts b/packages/cli/test/mcp/mongo-query-tool.test.ts index 4d316baf9..ee9f788af 100644 --- a/packages/cli/test/mcp/mongo-query-tool.test.ts +++ b/packages/cli/test/mcp/mongo-query-tool.test.ts @@ -1,8 +1,9 @@ import { describe, expect, it, vi } from 'vitest'; -import { KtxQueryError } from '../../src/errors.js'; +import { KtxExpectedError } from '../../src/errors.js'; import { KtxMongoDbScanConnector } from '../../src/connectors/mongodb/connector.js'; import { createLocalProjectMcpContextPorts } from '../../src/context/mcp/local-project-ports.js'; import type { KtxLocalProject } from '../../src/context/project/project.js'; +import type { KtxScanConnector } from '../../src/context/scan/types.js'; const MONGO_DOCS = [ { _id: 'a1', city: 'Indianapolis' }, @@ -38,7 +39,18 @@ function project(): KtxLocalProject { } as unknown as KtxLocalProject; } -function portsFor(createConnector: (id: string) => KtxMongoDbScanConnector) { +function warehouseConnector(): KtxScanConnector { + return { + id: 'postgres:warehouse', + driver: 'postgres', + capabilities: { structuralIntrospection: true } as KtxScanConnector['capabilities'], + introspect: vi.fn(), + listSchemas: vi.fn(async () => []), + listTables: vi.fn(async () => []), + } as unknown as KtxScanConnector; +} + +function portsFor(createConnector: (id: string) => KtxScanConnector) { return createLocalProjectMcpContextPorts(project(), { embeddingService: null, localScan: { createConnector } as never, @@ -58,10 +70,13 @@ describe('mongoQuery MCP port', () => { expect(result.rowCount).toBe(2); }); - it('rejects a non-mongodb connection id as an expected query error', async () => { - const ports = portsFor(() => mongoConnector()); + it('rejects a non-mongodb connection id with an expected error naming the wrong driver', async () => { + const ports = portsFor((id) => (id === 'mongo' ? mongoConnector() : warehouseConnector())); + await expect( + ports.mongoQuery!.execute({ connectionId: 'warehouse', collection: 'x', pipeline: [], limit: 10 }), + ).rejects.toThrow(/is not a MongoDB connection/); await expect( ports.mongoQuery!.execute({ connectionId: 'warehouse', collection: 'x', pipeline: [], limit: 10 }), - ).rejects.toBeInstanceOf(KtxQueryError); + ).rejects.toBeInstanceOf(KtxExpectedError); }); }); From 5cfd64f15108ca64f976303fd70787f84b134533 Mon Sep 17 00:00:00 2001 From: Kevin Messiaen <114553769+kevinmessiaen@users.noreply.github.com> Date: Mon, 6 Jul 2026 16:33:56 +0700 Subject: [PATCH 05/17] feat(mcp): register mongo_query tool Expose the mongoQuery port as a callable MCP tool so agents can run aggregation pipelines against MongoDB connections directly, since MongoDB collections are not queryable via sql_execution. --- packages/cli/src/context/mcp/context-tools.ts | 56 +++++++++++++++++++ 1 file changed, 56 insertions(+) diff --git a/packages/cli/src/context/mcp/context-tools.ts b/packages/cli/src/context/mcp/context-tools.ts index 106167e11..e9b0898b8 100644 --- a/packages/cli/src/context/mcp/context-tools.ts +++ b/packages/cli/src/context/mcp/context-tools.ts @@ -52,6 +52,7 @@ const toolAnnotations = { sl_read_source: { title: 'Semantic Layer Read Source', readOnlyHint: true, idempotentHint: true, openWorldHint: false }, sl_query: { title: 'Semantic Layer Query', readOnlyHint: true, openWorldHint: false }, sql_execution: { title: 'SQL Execution', readOnlyHint: true, openWorldHint: false }, + mongo_query: { title: 'MongoDB Query', readOnlyHint: true, openWorldHint: false }, sql_dialect_notes: { title: 'SQL Dialect Notes', readOnlyHint: true, idempotentHint: true, openWorldHint: false }, memory_ingest: { title: 'Memory Ingest', destructiveHint: true, openWorldHint: false }, memory_ingest_status: { title: 'Memory Ingest Status', readOnlyHint: true, openWorldHint: false }, @@ -75,6 +76,8 @@ const toolDescriptions = { 'Execute a semantic-layer query and return headers, rows, and total row count, plus correctness notes (e.g. compile-only or fan-out) when relevant. The generated SQL and full query plan are omitted by default; request them with include: ["sql"] and/or include: ["plan"]. Example: sl_query({ connectionId: "warehouse", measures: ["orders.order_count"], dimensions: [{ field: "orders.created_at", granularity: "month" }], include: ["sql"] }).', sql_execution: 'Execute one parser-validated read-only SQL query against a configured ktx connection. Example: sql_execution({ connectionId: "warehouse", sql: "select count(*) from public.orders", maxRows: 100 }).', + mongo_query: + 'Fetch real rows from a live MongoDB connection by running an aggregation pipeline against one collection. MongoDB is not SQL-queryable, so use this — not sql_execution — to filter, project, group, or count Mongo documents. Example: mongo_query({ connectionId: "mongo", collection: "business", pipeline: [{ "$match": { "city": "Indianapolis" } }, { "$project": { "business_id": 1, "city": 1 } }], limit: 500 }).', sql_dialect_notes: 'Return the SQL syntax conventions for the dialect of a ktx connection: fully-qualified table-name form, identifier quoting and case-folding, date/time functions, top-N / window-filtering idiom, and JSON access. Call this before writing raw sql_execution SQL against a connection so the SQL matches that engine. Example: sql_dialect_notes({ connectionId: "warehouse" }).', memory_ingest: @@ -217,6 +220,27 @@ const sqlDialectNotesSchema = z.object({ connectionId: connectionIdSchema.describe('Connection id whose engine dialect conventions to return.'), }); +const mongoQuerySchema = z.object({ + connectionId: connectionIdSchema.describe('MongoDB connection id to query.'), + collection: z.string().min(1).describe('Collection to query, e.g. "business".'), + database: z + .string() + .min(1) + .optional() + .describe('Database name when the connection spans several; defaults to the connection\'s first configured database.'), + pipeline: z + .array(z.record(z.string(), z.unknown())) + .default([]) + .describe('MongoDB aggregation pipeline stages, e.g. [{ "$match": { "city": "Indianapolis" } }].'), + limit: z.number().int().min(1).max(10_000).default(1000).describe('Maximum documents to return.'), +}); + +const mongoQueryOutputSchema = z.object({ + headers: z.array(z.string()), + rows: z.array(z.array(z.unknown())), + rowCount: z.number(), +}); + const memoryIngestSchema = z.object({ content: z .string() @@ -971,6 +995,38 @@ export function registerKtxContextTools(deps: RegisterKtxContextToolsDeps): void ); } + if (ports.mongoQuery) { + const mongoQuery = ports.mongoQuery; + registerParsedTool( + server, + 'mongo_query', + { + title: toolAnnotations.mongo_query.title!, + description: toolDescriptions.mongo_query, + inputSchema: mongoQuerySchema.shape, + outputSchema: mongoQueryOutputSchema, + annotations: toolAnnotations.mongo_query, + }, + mongoQuerySchema, + async (input, context) => { + const onProgress = mcpProgressCallback(context); + return jsonToolResult( + await mongoQuery.execute( + { + connectionId: input.connectionId, + collection: input.collection, + ...(input.database ? { database: input.database } : {}), + pipeline: input.pipeline, + limit: input.limit, + }, + onProgress ? { onProgress } : undefined, + ), + ); + }, + toolTelemetry, + ); + } + if (ports.dialectNotes) { const dialectNotes = ports.dialectNotes; registerParsedTool( From bba645852f0aacefa22f52d508f8ce60b07f3f08 Mon Sep 17 00:00:00 2001 From: Kevin Messiaen <114553769+kevinmessiaen@users.noreply.github.com> Date: Mon, 6 Jul 2026 16:40:06 +0700 Subject: [PATCH 06/17] test(mcp): add mongo_query server registration test Closes a coverage gap for the mongo_query MCP tool and fixes Knip's default-mode unused-export flag on KtxMongoQueryMcpPort, which had no cross-file consumer before this test imported it. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01BdUJSmjqGoksdc45ZZJiVY --- packages/cli/test/context/mcp/server.test.ts | 52 ++++++++++++++++++++ 1 file changed, 52 insertions(+) diff --git a/packages/cli/test/context/mcp/server.test.ts b/packages/cli/test/context/mcp/server.test.ts index 6a7e756a4..69091aab9 100644 --- a/packages/cli/test/context/mcp/server.test.ts +++ b/packages/cli/test/context/mcp/server.test.ts @@ -21,6 +21,8 @@ import type { KtxKnowledgeMcpPort, KtxMcpContextPorts, KtxMcpToolHandlerContext, + KtxMongoQueryMcpPort, + KtxMongoQueryResponse, KtxSemanticLayerMcpPort, KtxSqlExecutionMcpPort, KtxSqlExecutionResponse, @@ -495,6 +497,56 @@ describe('createKtxMcpServer', () => { ); }); + it('registers mongo_query when the host provides a MongoDB query port', async () => { + const fake = makeFakeServer(); + const response: KtxMongoQueryResponse = { + headers: ['_id', 'city'], + rows: [ + ['a1', 'Indianapolis'], + ['a2', 'Indianapolis'], + ], + rowCount: 2, + }; + const mongoQuery: KtxMongoQueryMcpPort = { + execute: vi.fn().mockResolvedValue(response), + }; + + createKtxMcpServer({ + server: fake.server, + userContext: { userId: 'local-user' }, + contextTools: { + mongoQuery, + }, + }); + + expect(fake.tools.map((tool) => tool.name)).toEqual(['mongo_query']); + await expect( + getTool(fake.tools, 'mongo_query').handler({ + connectionId: 'mongo', + collection: 'business', + pipeline: [{ $match: { city: 'Indianapolis' } }], + limit: 100, + }), + ).resolves.toEqual({ + content: [ + { + type: 'text', + text: JSON.stringify(response), + }, + ], + structuredContent: response, + }); + expect(mongoQuery.execute).toHaveBeenCalledWith( + { + connectionId: 'mongo', + collection: 'business', + pipeline: [{ $match: { city: 'Indianapolis' } }], + limit: 100, + }, + undefined, + ); + }); + it('registers entity_details when the host provides an entity-details port', async () => { const fake = makeFakeServer(); const entityDetails: KtxEntityDetailsMcpPort = { From f5d431a398102c477eb472a1125646857d7a63ef Mon Sep 17 00:00:00 2001 From: Kevin Messiaen <114553769+kevinmessiaen@users.noreply.github.com> Date: Mon, 6 Jul 2026 17:00:35 +0700 Subject: [PATCH 07/17] refactor(mongo-query): dedupe result type and enforce read-only pipeline --- .../cli/src/connectors/mongodb/connector.ts | 10 ++++++++++ .../src/context/mcp/local-project-ports.ts | 3 +-- packages/cli/src/context/mcp/types.ts | 7 ++----- .../test/connectors/mongodb/connector.test.ts | 20 +++++++++++++++++++ 4 files changed, 33 insertions(+), 7 deletions(-) diff --git a/packages/cli/src/connectors/mongodb/connector.ts b/packages/cli/src/connectors/mongodb/connector.ts index 123508354..d6d185c82 100644 --- a/packages/cli/src/connectors/mongodb/connector.ts +++ b/packages/cli/src/connectors/mongodb/connector.ts @@ -222,6 +222,15 @@ function unionDocumentKeys(documents: readonly KtxMongoDocument[]): string[] { return keys; } +// MongoDB aggregation write stages ($out/$merge) persist results; mongo_query is read-only. +function assertReadOnlyPipeline(pipeline: Record[]): void { + for (const stage of pipeline) { + if ('$out' in stage || '$merge' in stage) { + throw new Error('mongo_query pipelines must be read-only; $out and $merge stages are not allowed.'); + } + } +} + export class KtxMongoDbScanConnector implements KtxScanConnector { readonly id: string; readonly driver = 'mongodb' as const; @@ -398,6 +407,7 @@ export class KtxMongoDbScanConnector implements KtxScanConnector { async executeQuery(input: KtxMongoQueryInput, _ctx: KtxScanContext): Promise { this.assertConnection(input.connectionId); + assertReadOnlyPipeline(input.pipeline); const database = input.database ?? this.databases[0]!; const documents = await this.clientForQuery().aggregate(database, input.collection, input.pipeline, { limit: input.limit, diff --git a/packages/cli/src/context/mcp/local-project-ports.ts b/packages/cli/src/context/mcp/local-project-ports.ts index 71aa04569..c366a6580 100644 --- a/packages/cli/src/context/mcp/local-project-ports.ts +++ b/packages/cli/src/context/mcp/local-project-ports.ts @@ -75,11 +75,10 @@ async function executeMongoQuery( `Connection "${connectionId}" driver "${connector.driver}" is not a MongoDB connection; mongo_query serves mongodb connections only.`, ); } - const result = await (connector as unknown as KtxMongoDbScanConnector).executeQuery( + return (connector as unknown as KtxMongoDbScanConnector).executeQuery( { connectionId, collection: input.collection, database: input.database, pipeline: input.pipeline, limit: input.limit }, { runId: 'mcp-mongo-query' }, ); - return { headers: result.headers, rows: result.rows, rowCount: result.rowCount }; } finally { await connector?.cleanup?.(); } diff --git a/packages/cli/src/context/mcp/types.ts b/packages/cli/src/context/mcp/types.ts index 36a861153..4ef12272a 100644 --- a/packages/cli/src/context/mcp/types.ts +++ b/packages/cli/src/context/mcp/types.ts @@ -5,6 +5,7 @@ import type { KtxEntityDetailsInput, KtxEntityDetailsResponse } from '../scan/en import type { KtxDiscoverDataInput, KtxDiscoverDataResponse } from '../../context/search/discover.js'; import type { KtxDictionarySearchInput, KtxDictionarySearchResponse } from '../../context/sl/dictionary-search.js'; import type { SemanticLayerQueryInput } from '../../context/sl/types.js'; +import type { KtxMongoQueryResult } from '../../connectors/mongodb/connector.js'; import type { WikiSearchLaneSummary, WikiSearchMatchReason } from '../../context/wiki/types.js'; interface KtxMcpTextContent { @@ -180,11 +181,7 @@ export interface KtxSqlExecutionMcpPort { ): Promise; } -export interface KtxMongoQueryResponse { - headers: string[]; - rows: unknown[][]; - rowCount: number; -} +export type KtxMongoQueryResponse = KtxMongoQueryResult; /** @internal */ export interface KtxMongoQueryMcpPort { diff --git a/packages/cli/test/connectors/mongodb/connector.test.ts b/packages/cli/test/connectors/mongodb/connector.test.ts index 3fd250390..7089ee218 100644 --- a/packages/cli/test/connectors/mongodb/connector.test.ts +++ b/packages/cli/test/connectors/mongodb/connector.test.ts @@ -275,4 +275,24 @@ describe('KtxMongoDbScanConnector.executeQuery', () => { ), ).rejects.toThrow(/cannot serve connection other/); }); + + it('rejects a pipeline containing a $out write stage', async () => { + const { factory } = fakeClientFactory(); + await expect( + connector(baseConnection, factory).executeQuery( + { connectionId: 'mongo-prod', collection: 'users', pipeline: [{ $out: 'exfiltrated' }], limit: 10 }, + { runId: 't' }, + ), + ).rejects.toThrow(/must be read-only/); + }); + + it('rejects a pipeline containing a $merge write stage', async () => { + const { factory } = fakeClientFactory(); + await expect( + connector(baseConnection, factory).executeQuery( + { connectionId: 'mongo-prod', collection: 'users', pipeline: [{ $merge: { into: 'x' } }], limit: 10 }, + { runId: 't' }, + ), + ).rejects.toThrow(/must be read-only/); + }); }); From 9a2e215bdc50b1a59662be0dbcf9eb688ee72402 Mon Sep 17 00:00:00 2001 From: Kevin Messiaen <114553769+kevinmessiaen@users.noreply.github.com> Date: Mon, 6 Jul 2026 19:54:41 +0700 Subject: [PATCH 08/17] refactor(mongo-query): extract shared runMongoQuery execution seam --- .../src/context/mcp/local-project-ports.ts | 19 ++-------- .../cli/src/context/mcp/mongo-query-runner.ts | 35 +++++++++++++++++++ .../cli/test/mcp/mongo-query-tool.test.ts | 23 ++++++++++++ 3 files changed, 60 insertions(+), 17 deletions(-) create mode 100644 packages/cli/src/context/mcp/mongo-query-runner.ts diff --git a/packages/cli/src/context/mcp/local-project-ports.ts b/packages/cli/src/context/mcp/local-project-ports.ts index c366a6580..43901685a 100644 --- a/packages/cli/src/context/mcp/local-project-ports.ts +++ b/packages/cli/src/context/mcp/local-project-ports.ts @@ -25,7 +25,7 @@ import { readLocalSlSource } from '../../context/sl/local-sl.js'; import { assertSafeConnectionId } from '../../context/sl/source-files.js'; import { assertConfiguredConnectionId } from '../../context/connections/configured-connections.js'; import { readLocalKnowledgePage, searchLocalKnowledgePages } from '../wiki/local-knowledge.js'; -import type { KtxMongoDbScanConnector } from '../../connectors/mongodb/connector.js'; +import { runMongoQuery } from './mongo-query-runner.js'; import type { KtxMcpContextPorts, KtxMcpProgressCallback, KtxMongoQueryResponse, KtxSqlExecutionResponse } from './types.js'; interface CreateLocalProjectMcpContextPortsOptions { @@ -66,22 +66,7 @@ async function executeMongoQuery( createConnector: (connectionId: string) => Promise | KtxScanConnector, input: { connectionId: string; collection: string; database?: string; pipeline: Record[]; limit: number }, ): Promise { - const connectionId = assertSafeConnectionId(input.connectionId); - let connector: KtxScanConnector | null = null; - try { - connector = await createConnector(connectionId); - if (connector.driver !== 'mongodb') { - throw new KtxExpectedError( - `Connection "${connectionId}" driver "${connector.driver}" is not a MongoDB connection; mongo_query serves mongodb connections only.`, - ); - } - return (connector as unknown as KtxMongoDbScanConnector).executeQuery( - { connectionId, collection: input.collection, database: input.database, pipeline: input.pipeline, limit: input.limit }, - { runId: 'mcp-mongo-query' }, - ); - } finally { - await connector?.cleanup?.(); - } + return runMongoQuery(createConnector, input); } async function executeValidatedReadOnlySql( diff --git a/packages/cli/src/context/mcp/mongo-query-runner.ts b/packages/cli/src/context/mcp/mongo-query-runner.ts new file mode 100644 index 000000000..fdd0b0820 --- /dev/null +++ b/packages/cli/src/context/mcp/mongo-query-runner.ts @@ -0,0 +1,35 @@ +import type { KtxMongoDbScanConnector, KtxMongoQueryResult } from '../../connectors/mongodb/connector.js'; +import { KtxExpectedError } from '../../errors.js'; +import { assertSafeConnectionId } from '../sl/source-files.js'; +import type { KtxScanConnector } from '../scan/types.js'; + +export interface KtxMongoQueryRequest { + connectionId: string; + collection: string; + database?: string; + pipeline: Record[]; + limit: number; +} + +/** Single Mongo-read execution seam: resolve the connector, guard the driver, run the pipeline. */ +export async function runMongoQuery( + createConnector: (connectionId: string) => Promise | KtxScanConnector, + input: KtxMongoQueryRequest, +): Promise { + const connectionId = assertSafeConnectionId(input.connectionId); + let connector: KtxScanConnector | null = null; + try { + connector = await createConnector(connectionId); + if (connector.driver !== 'mongodb') { + throw new KtxExpectedError( + `Connection "${connectionId}" driver "${connector.driver}" is not a MongoDB connection; mongo_query serves mongodb connections only.`, + ); + } + return await (connector as unknown as KtxMongoDbScanConnector).executeQuery( + { connectionId, collection: input.collection, database: input.database, pipeline: input.pipeline, limit: input.limit }, + { runId: 'mongo-query' }, + ); + } finally { + await connector?.cleanup?.(); + } +} diff --git a/packages/cli/test/mcp/mongo-query-tool.test.ts b/packages/cli/test/mcp/mongo-query-tool.test.ts index ee9f788af..e0af6023c 100644 --- a/packages/cli/test/mcp/mongo-query-tool.test.ts +++ b/packages/cli/test/mcp/mongo-query-tool.test.ts @@ -2,6 +2,7 @@ import { describe, expect, it, vi } from 'vitest'; import { KtxExpectedError } from '../../src/errors.js'; import { KtxMongoDbScanConnector } from '../../src/connectors/mongodb/connector.js'; import { createLocalProjectMcpContextPorts } from '../../src/context/mcp/local-project-ports.js'; +import { runMongoQuery } from '../../src/context/mcp/mongo-query-runner.js'; import type { KtxLocalProject } from '../../src/context/project/project.js'; import type { KtxScanConnector } from '../../src/context/scan/types.js'; @@ -80,3 +81,25 @@ describe('mongoQuery MCP port', () => { ).rejects.toBeInstanceOf(KtxExpectedError); }); }); + +describe('runMongoQuery shared helper', () => { + it('fetches rows from a mongodb connector', async () => { + const result = await runMongoQuery( + (id) => (id === 'mongo' ? mongoConnector() : warehouseConnector()), + { connectionId: 'mongo', collection: 'business', pipeline: [], limit: 100 }, + ); + expect(result.rowCount).toBe(2); + expect(result.headers).toEqual(['_id', 'city']); + }); + + it('throws an expected error naming the wrong driver for a non-mongodb connector', async () => { + await expect( + runMongoQuery((id) => (id === 'mongo' ? mongoConnector() : warehouseConnector()), { + connectionId: 'warehouse', + collection: 'x', + pipeline: [], + limit: 10, + }), + ).rejects.toThrow(/is not a MongoDB connection/); + }); +}); From 89b87299508a93f6b7a42ab48d63bfe588e40501 Mon Sep 17 00:00:00 2001 From: Kevin Messiaen <114553769+kevinmessiaen@users.noreply.github.com> Date: Mon, 6 Jul 2026 19:59:34 +0700 Subject: [PATCH 09/17] refactor(io): extract shared result-table formatters from sql Move formatValue/printJson/printPlain/printPretty/printResultTable and the KtxResultTable interface out of sql.ts into io/result-table.ts so the upcoming mongo-query command can reuse them verbatim. --- packages/cli/src/io/result-table.ts | 59 ++++++++++++++++++++++ packages/cli/src/sql.ts | 61 ++--------------------- packages/cli/test/io/result-table.test.ts | 46 +++++++++++++++++ 3 files changed, 108 insertions(+), 58 deletions(-) create mode 100644 packages/cli/src/io/result-table.ts create mode 100644 packages/cli/test/io/result-table.test.ts diff --git a/packages/cli/src/io/result-table.ts b/packages/cli/src/io/result-table.ts new file mode 100644 index 000000000..98f301d08 --- /dev/null +++ b/packages/cli/src/io/result-table.ts @@ -0,0 +1,59 @@ +import type { KtxCliIo } from '../cli-runtime.js'; +import type { KtxOutputMode } from './mode.js'; + +export interface KtxResultTable { + connectionId: string; + headers: string[]; + headerTypes?: string[]; + rows: unknown[][]; + rowCount: number; +} + +/** @internal */ +export function formatValue(value: unknown): string { + if (value === null || value === undefined) return ''; + if (typeof value === 'string') return value; + if (typeof value === 'number' || typeof value === 'boolean' || typeof value === 'bigint') return String(value); + return JSON.stringify(value); +} + +function printJson(output: KtxResultTable, io: KtxCliIo): void { + io.stdout.write(`${JSON.stringify(output, null, 2)}\n`); +} + +function printPlain(output: KtxResultTable, io: KtxCliIo): void { + io.stdout.write(`${output.headers.join('\t')}\n`); + for (const row of output.rows) { + io.stdout.write(`${row.map(formatValue).join('\t')}\n`); + } +} + +function printPretty(output: KtxResultTable, io: KtxCliIo): void { + const rows = output.rows.map((row) => row.map(formatValue)); + const widths = output.headers.map((header, index) => + Math.max(header.length, ...rows.map((row) => row[index]?.length ?? 0)), + ); + const renderRow = (cells: string[]): string => + cells.map((cell, index) => cell.padEnd(widths[index] ?? cell.length)).join(' ').trimEnd(); + + if (output.headers.length > 0) { + io.stdout.write(`${renderRow(output.headers)}\n`); + io.stdout.write(`${renderRow(widths.map((width) => '-'.repeat(width)))}\n`); + } + for (const row of rows) { + io.stdout.write(`${renderRow(row)}\n`); + } + io.stdout.write(`\n${output.rowCount} ${output.rowCount === 1 ? 'row' : 'rows'}\n`); +} + +export function printResultTable(output: KtxResultTable, mode: KtxOutputMode, io: KtxCliIo): void { + if (mode === 'json') { + printJson(output, io); + return; + } + if (mode === 'plain') { + printPlain(output, io); + return; + } + printPretty(output, io); +} diff --git a/packages/cli/src/sql.ts b/packages/cli/src/sql.ts index fbf059757..bd29245a2 100644 --- a/packages/cli/src/sql.ts +++ b/packages/cli/src/sql.ts @@ -9,6 +9,7 @@ import { sqlAnalysisDialectForDriver } from './context/sql-analysis/dialect.js'; import type { SqlAnalysisDialect, SqlAnalysisPort } from './context/sql-analysis/ports.js'; import type { KtxCliIo } from './cli-runtime.js'; import { type KtxOutputMode, resolveOutputMode } from './io/mode.js'; +import { type KtxResultTable, printResultTable } from './io/result-table.js'; import { createKtxCliScanConnector } from './local-scan-connectors.js'; import { createManagedDaemonSqlAnalysisPort } from './managed-python-http.js'; import { profileMark } from './startup-profile.js'; @@ -39,14 +40,6 @@ export interface KtxSqlDeps { executeFederated?: typeof executeFederatedQuery; } -interface SqlExecutionOutput { - connectionId: string; - headers: string[]; - headerTypes?: string[]; - rows: unknown[][]; - rowCount: number; -} - function queryVerb(sql: string): 'select' | 'explain' | 'show' | 'with' | 'other' { const first = sql.trim().split(/\s+/, 1)[0]?.toLowerCase(); if (first === 'select' || first === 'explain' || first === 'show' || first === 'with') { @@ -68,55 +61,7 @@ async function safeReferencedTableCount( } } -function formatValue(value: unknown): string { - if (value === null || value === undefined) return ''; - if (typeof value === 'string') return value; - if (typeof value === 'number' || typeof value === 'boolean' || typeof value === 'bigint') return String(value); - return JSON.stringify(value); -} - -function printJson(output: SqlExecutionOutput, io: KtxCliIo): void { - io.stdout.write(`${JSON.stringify(output, null, 2)}\n`); -} - -function printPlain(output: SqlExecutionOutput, io: KtxCliIo): void { - io.stdout.write(`${output.headers.join('\t')}\n`); - for (const row of output.rows) { - io.stdout.write(`${row.map(formatValue).join('\t')}\n`); - } -} - -function printPretty(output: SqlExecutionOutput, io: KtxCliIo): void { - const rows = output.rows.map((row) => row.map(formatValue)); - const widths = output.headers.map((header, index) => - Math.max(header.length, ...rows.map((row) => row[index]?.length ?? 0)), - ); - const renderRow = (cells: string[]): string => - cells.map((cell, index) => cell.padEnd(widths[index] ?? cell.length)).join(' ').trimEnd(); - - if (output.headers.length > 0) { - io.stdout.write(`${renderRow(output.headers)}\n`); - io.stdout.write(`${renderRow(widths.map((width) => '-'.repeat(width)))}\n`); - } - for (const row of rows) { - io.stdout.write(`${renderRow(row)}\n`); - } - io.stdout.write(`\n${output.rowCount} ${output.rowCount === 1 ? 'row' : 'rows'}\n`); -} - -function printSqlResult(output: SqlExecutionOutput, mode: KtxSqlOutputMode, io: KtxCliIo): void { - if (mode === 'json') { - printJson(output, io); - return; - } - if (mode === 'plain') { - printPlain(output, io); - return; - } - printPretty(output, io); -} - -function resultOutput(connectionId: string, result: KtxSqlQueryExecutionResult): SqlExecutionOutput { +function resultOutput(connectionId: string, result: KtxSqlQueryExecutionResult): KtxResultTable { return { connectionId, headers: result.headers, @@ -168,7 +113,7 @@ export async function runKtxSql(args: KtxSqlArgs, io: KtxCliIo = process, deps: }); const referencedTableCount = await safeReferencedTableCount(analysisPort, args.sql, dialect); const mode = resolveOutputMode({ explicit: args.output, json: args.json, io }); - printSqlResult(resultOutput(args.connectionId, result), mode, io); + printResultTable(resultOutput(args.connectionId, result), mode, io); await emitTelemetryEvent({ name: 'sql_completed', projectDir: args.projectDir, diff --git a/packages/cli/test/io/result-table.test.ts b/packages/cli/test/io/result-table.test.ts new file mode 100644 index 000000000..39508d370 --- /dev/null +++ b/packages/cli/test/io/result-table.test.ts @@ -0,0 +1,46 @@ +import { describe, expect, it } from 'vitest'; +import { formatValue, printResultTable } from '../../src/io/result-table.js'; +import type { KtxCliIo } from '../../src/cli-runtime.js'; + +function captureIo(): { io: KtxCliIo; out: () => string } { + let buffer = ''; + const io = { + stdout: { write: (s: string) => { buffer += s; return true; } }, + stderr: { write: () => true }, + } as unknown as KtxCliIo; + return { io, out: () => buffer }; +} + +describe('formatValue', () => { + it('renders null/undefined as empty, scalars as strings, objects as JSON', () => { + expect(formatValue(null)).toBe(''); + expect(formatValue(undefined)).toBe(''); + expect(formatValue('x')).toBe('x'); + expect(formatValue(42)).toBe('42'); + expect(formatValue(true)).toBe('true'); + expect(formatValue({ a: 1 })).toBe(JSON.stringify({ a: 1 })); + }); +}); + +describe('printResultTable', () => { + const table = { connectionId: 'mongo', headers: ['_id', 'city'], rows: [['a1', 'NY']], rowCount: 1 }; + + it('json mode prints the structured payload', () => { + const { io, out } = captureIo(); + printResultTable(table, 'json', io); + expect(JSON.parse(out())).toEqual(table); + }); + + it('plain mode prints tab-separated headers and rows', () => { + const { io, out } = captureIo(); + printResultTable(table, 'plain', io); + expect(out()).toBe('_id\tcity\na1\tNY\n'); + }); + + it('pretty mode prints a header rule and a row count', () => { + const { io, out } = captureIo(); + printResultTable(table, 'pretty', io); + expect(out()).toContain('_id'); + expect(out()).toContain('1 row'); + }); +}); From cd9a6f5dc203cc02dfba19cf3b0890ac64828bb8 Mon Sep 17 00:00:00 2001 From: Kevin Messiaen <114553769+kevinmessiaen@users.noreply.github.com> Date: Mon, 6 Jul 2026 20:05:22 +0700 Subject: [PATCH 10/17] feat(telemetry): add mongo_query_completed event Adds the strict mongo_query_completed schema (driver, isDemoConnection, stageCount, durationMs, outcome, errorClass) for the upcoming ktx mongo-query command, registered in both the catalog object and the name/description/fields catalog list, and mirrors it to the Python daemon schema copy. --- packages/cli/src/telemetry/events.schema.json | 83 +++++++++++++++++++ packages/cli/src/telemetry/events.ts | 17 ++++ packages/cli/test/telemetry/events.test.ts | 1 + .../ktx_daemon/telemetry/events.schema.json | 83 +++++++++++++++++++ .../tests/test_telemetry_schema_sync.py | 1 + 5 files changed, 185 insertions(+) diff --git a/packages/cli/src/telemetry/events.schema.json b/packages/cli/src/telemetry/events.schema.json index 405f02400..0bd22bb3d 100644 --- a/packages/cli/src/telemetry/events.schema.json +++ b/packages/cli/src/telemetry/events.schema.json @@ -144,6 +144,18 @@ "errorClass" ] }, + { + "name": "mongo_query_completed", + "description": "Emitted after ktx mongo-query completes validation and execution.", + "fields": [ + "driver", + "isDemoConnection", + "stageCount", + "durationMs", + "outcome", + "errorClass" + ] + }, { "name": "wiki_query_completed", "description": "Emitted after a wiki query completes.", @@ -1056,6 +1068,77 @@ ], "additionalProperties": false }, + "mongo_query_completed": { + "$schema": "https://json-schema.org/draft/2020-12/schema", + "type": "object", + "properties": { + "cliVersion": { + "type": "string" + }, + "nodeVersion": { + "type": "string" + }, + "osPlatform": { + "type": "string" + }, + "osRelease": { + "type": "string" + }, + "arch": { + "type": "string" + }, + "runtime": { + "type": "string", + "enum": [ + "node", + "daemon-py" + ] + }, + "isCi": { + "type": "boolean" + }, + "driver": { + "type": "string" + }, + "isDemoConnection": { + "type": "boolean" + }, + "stageCount": { + "type": "integer", + "minimum": 0, + "maximum": 9007199254740991 + }, + "durationMs": { + "type": "number", + "minimum": 0 + }, + "outcome": { + "type": "string", + "enum": [ + "ok", + "error" + ] + }, + "errorClass": { + "type": "string" + } + }, + "required": [ + "cliVersion", + "nodeVersion", + "osPlatform", + "osRelease", + "arch", + "runtime", + "isCi", + "driver", + "isDemoConnection", + "stageCount", + "durationMs", + "outcome" + ], + "additionalProperties": false + }, "wiki_query_completed": { "$schema": "https://json-schema.org/draft/2020-12/schema", "type": "object", diff --git a/packages/cli/src/telemetry/events.ts b/packages/cli/src/telemetry/events.ts index 6d311ac53..84b8617b8 100644 --- a/packages/cli/src/telemetry/events.ts +++ b/packages/cli/src/telemetry/events.ts @@ -146,6 +146,17 @@ const sqlCompletedSchema = telemetryCommonEnvelopeSchema }) .strict(); +const mongoQueryCompletedSchema = telemetryCommonEnvelopeSchema + .extend({ + driver: z.string(), + isDemoConnection: z.boolean(), + stageCount: z.number().int().nonnegative(), + durationMs: z.number().nonnegative(), + outcome: outcomeSchema, + errorClass: z.string().optional(), + }) + .strict(); + const wikiQueryCompletedSchema = telemetryCommonEnvelopeSchema .extend({ queryLength: z.number().int().nonnegative(), @@ -230,6 +241,7 @@ export const telemetryEventSchemas = { sl_validate_completed: slValidateCompletedSchema, sl_query_completed: slQueryCompletedSchema, sql_completed: sqlCompletedSchema, + mongo_query_completed: mongoQueryCompletedSchema, wiki_query_completed: wikiQueryCompletedSchema, mcp_request_completed: mcpRequestCompletedSchema, daemon_started: daemonStartedSchema, @@ -342,6 +354,11 @@ export const telemetryEventCatalog = [ 'errorClass', ], }, + { + name: 'mongo_query_completed', + description: 'Emitted after ktx mongo-query completes validation and execution.', + fields: ['driver', 'isDemoConnection', 'stageCount', 'durationMs', 'outcome', 'errorClass'], + }, { name: 'wiki_query_completed', description: 'Emitted after a wiki query completes.', diff --git a/packages/cli/test/telemetry/events.test.ts b/packages/cli/test/telemetry/events.test.ts index 033c2def5..1ed5f1f1e 100644 --- a/packages/cli/test/telemetry/events.test.ts +++ b/packages/cli/test/telemetry/events.test.ts @@ -31,6 +31,7 @@ describe('telemetry event schemas', () => { 'sl_validate_completed', 'sl_query_completed', 'sql_completed', + 'mongo_query_completed', 'wiki_query_completed', 'mcp_request_completed', 'daemon_started', diff --git a/python/ktx-daemon/src/ktx_daemon/telemetry/events.schema.json b/python/ktx-daemon/src/ktx_daemon/telemetry/events.schema.json index 405f02400..0bd22bb3d 100644 --- a/python/ktx-daemon/src/ktx_daemon/telemetry/events.schema.json +++ b/python/ktx-daemon/src/ktx_daemon/telemetry/events.schema.json @@ -144,6 +144,18 @@ "errorClass" ] }, + { + "name": "mongo_query_completed", + "description": "Emitted after ktx mongo-query completes validation and execution.", + "fields": [ + "driver", + "isDemoConnection", + "stageCount", + "durationMs", + "outcome", + "errorClass" + ] + }, { "name": "wiki_query_completed", "description": "Emitted after a wiki query completes.", @@ -1056,6 +1068,77 @@ ], "additionalProperties": false }, + "mongo_query_completed": { + "$schema": "https://json-schema.org/draft/2020-12/schema", + "type": "object", + "properties": { + "cliVersion": { + "type": "string" + }, + "nodeVersion": { + "type": "string" + }, + "osPlatform": { + "type": "string" + }, + "osRelease": { + "type": "string" + }, + "arch": { + "type": "string" + }, + "runtime": { + "type": "string", + "enum": [ + "node", + "daemon-py" + ] + }, + "isCi": { + "type": "boolean" + }, + "driver": { + "type": "string" + }, + "isDemoConnection": { + "type": "boolean" + }, + "stageCount": { + "type": "integer", + "minimum": 0, + "maximum": 9007199254740991 + }, + "durationMs": { + "type": "number", + "minimum": 0 + }, + "outcome": { + "type": "string", + "enum": [ + "ok", + "error" + ] + }, + "errorClass": { + "type": "string" + } + }, + "required": [ + "cliVersion", + "nodeVersion", + "osPlatform", + "osRelease", + "arch", + "runtime", + "isCi", + "driver", + "isDemoConnection", + "stageCount", + "durationMs", + "outcome" + ], + "additionalProperties": false + }, "wiki_query_completed": { "$schema": "https://json-schema.org/draft/2020-12/schema", "type": "object", diff --git a/python/ktx-daemon/tests/test_telemetry_schema_sync.py b/python/ktx-daemon/tests/test_telemetry_schema_sync.py index 0cc822f99..21c076cdb 100644 --- a/python/ktx-daemon/tests/test_telemetry_schema_sync.py +++ b/python/ktx-daemon/tests/test_telemetry_schema_sync.py @@ -30,6 +30,7 @@ def test_python_schema_copy_matches_node_schema() -> None: "sl_validate_completed", "sl_query_completed", "sql_completed", + "mongo_query_completed", "wiki_query_completed", "mcp_request_completed", "daemon_started", From d32dd1de5f0cd803fe89e5510b1987f5f5c5590e Mon Sep 17 00:00:00 2001 From: Kevin Messiaen <114553769+kevinmessiaen@users.noreply.github.com> Date: Mon, 6 Jul 2026 20:09:16 +0700 Subject: [PATCH 11/17] feat(cli): add runKtxMongoQuery runner --- packages/cli/src/mongo-query.ts | 110 ++++++++++++++++++++++++++ packages/cli/test/mongo-query.test.ts | 69 ++++++++++++++++ 2 files changed, 179 insertions(+) create mode 100644 packages/cli/src/mongo-query.ts create mode 100644 packages/cli/test/mongo-query.test.ts diff --git a/packages/cli/src/mongo-query.ts b/packages/cli/src/mongo-query.ts new file mode 100644 index 000000000..f529df889 --- /dev/null +++ b/packages/cli/src/mongo-query.ts @@ -0,0 +1,110 @@ +import { runMongoQuery } from './context/mcp/mongo-query-runner.js'; +import { resolveConfiguredConnection } from './context/connections/resolve-connection.js'; +import { loadKtxProject, type KtxLocalProject } from './context/project/project.js'; +import type { KtxCliIo } from './cli-runtime.js'; +import { type KtxOutputMode, resolveOutputMode } from './io/mode.js'; +import { type KtxResultTable, printResultTable } from './io/result-table.js'; +import { createKtxCliScanConnector } from './local-scan-connectors.js'; +import { profileMark } from './startup-profile.js'; +import { isDemoConnection } from './telemetry/demo-detect.js'; +import { emitTelemetryEvent, reportException } from './telemetry/index.js'; +import { collectTelemetryRedactionSecrets } from './telemetry/redaction-secrets.js'; +import { scrubErrorClass } from './telemetry/scrubber.js'; + +profileMark('module:mongo-query'); + +export type KtxMongoQueryArgs = { + projectDir: string; + connectionId: string; + collection: string; + database?: string; + pipeline: Record[]; + limit: number; + output?: KtxOutputMode; + json?: boolean; + cliVersion: string; +}; + +export interface KtxMongoQueryCliDeps { + loadProject?: typeof loadKtxProject; + createScanConnector?: typeof createKtxCliScanConnector; +} + +export async function runKtxMongoQuery( + args: KtxMongoQueryArgs, + io: KtxCliIo = process, + deps: KtxMongoQueryCliDeps = {}, +): Promise { + const startedAt = performance.now(); + let driver = 'unknown'; + let demoConnection = false; + let project: KtxLocalProject | undefined; + try { + project = await (deps.loadProject ?? loadKtxProject)({ projectDir: args.projectDir }); + const connection = resolveConfiguredConnection(project.config, args.connectionId); + driver = String(connection?.driver ?? 'unknown').toLowerCase(); + demoConnection = isDemoConnection(args.connectionId, connection); + + const createScanConnector = deps.createScanConnector ?? createKtxCliScanConnector; + const result = await runMongoQuery((connectionId) => createScanConnector(project!, connectionId), { + connectionId: args.connectionId, + collection: args.collection, + database: args.database, + pipeline: args.pipeline, + limit: args.limit, + }); + + const output: KtxResultTable = { + connectionId: args.connectionId, + headers: result.headers, + rows: result.rows, + rowCount: result.rowCount, + }; + const mode = resolveOutputMode({ explicit: args.output, json: args.json, io }); + printResultTable(output, mode, io); + await emitTelemetryEvent({ + name: 'mongo_query_completed', + projectDir: args.projectDir, + io, + fields: { + driver, + isDemoConnection: demoConnection, + stageCount: args.pipeline.length, + durationMs: Math.max(0, performance.now() - startedAt), + outcome: 'ok', + }, + }); + return 0; + } catch (error) { + const errorClass = scrubErrorClass(error); + await emitTelemetryEvent({ + name: 'mongo_query_completed', + projectDir: args.projectDir, + io, + fields: { + driver, + isDemoConnection: demoConnection, + stageCount: args.pipeline.length, + durationMs: Math.max(0, performance.now() - startedAt), + outcome: 'error', + ...(errorClass ? { errorClass } : {}), + }, + }); + await reportException({ + error, + context: { source: 'mongo-query run', handled: true, fatal: false }, + projectDir: args.projectDir, + io, + redactionSecrets: await collectTelemetryRedactionSecrets({ + project, + projectDir: args.projectDir, + connectionId: args.connectionId, + includeLlm: false, + includeEmbeddings: false, + env: process.env, + }), + }); + io.stderr.write(`${error instanceof Error ? error.message : String(error)}\n`); + return 1; + } +} diff --git a/packages/cli/test/mongo-query.test.ts b/packages/cli/test/mongo-query.test.ts new file mode 100644 index 000000000..be1265ca2 --- /dev/null +++ b/packages/cli/test/mongo-query.test.ts @@ -0,0 +1,69 @@ +import { describe, expect, it, vi } from 'vitest'; +import { runKtxMongoQuery } from '../src/mongo-query.js'; +import { KtxMongoDbScanConnector } from '../src/connectors/mongodb/connector.js'; +import type { KtxCliIo } from '../src/cli-runtime.js'; +import type { KtxLocalProject } from '../src/context/project/project.js'; + +function captureIo(): { io: KtxCliIo; out: () => string; err: () => string } { + let out = ''; + let err = ''; + const io = { + stdout: { write: (s: string) => { out += s; return true; }, isTTY: false }, + stderr: { write: (s: string) => { err += s; return true; } }, + } as unknown as KtxCliIo; + return { io, out: () => out, err: () => err }; +} + +function mongoConnector(): KtxMongoDbScanConnector { + return new KtxMongoDbScanConnector({ + connectionId: 'mongo', + connection: { driver: 'mongodb', url: 'mongodb://localhost:27017/app', databases: ['app'] }, + clientFactory: { + create: () => ({ + listCollections: vi.fn(async () => []), + estimatedDocumentCount: vi.fn(async () => 0), + find: vi.fn(async () => []), + aggregate: vi.fn(async () => [{ _id: 'a1', city: 'Indianapolis' }, { _id: 'a2', city: 'Indianapolis' }]), + ping: vi.fn(async () => undefined), + close: vi.fn(async () => undefined), + }), + }, + }); +} + +const project = { + projectDir: '/tmp/ktx', + config: { connections: { mongo: { driver: 'mongodb', url: 'mongodb://localhost:27017/app', databases: ['app'] } } }, +} as unknown as KtxLocalProject; + +const baseArgs = { + projectDir: '/tmp/ktx', + connectionId: 'mongo', + collection: 'business', + pipeline: [{ $match: { city: 'Indianapolis' } }], + limit: 100, + cliVersion: '0.0.0-test', +}; + +describe('runKtxMongoQuery', () => { + it('prints rows in plain mode and returns exit code 0', async () => { + const { io, out } = captureIo(); + const code = await runKtxMongoQuery({ ...baseArgs, output: 'plain' }, io, { + loadProject: async () => project, + createScanConnector: (async () => mongoConnector()) as never, + }); + expect(code).toBe(0); + expect(out()).toContain('_id\tcity'); + expect(out()).toContain('a1\tIndianapolis'); + }); + + it('returns exit code 1 and writes the error message for an unconfigured connection', async () => { + const { io, err } = captureIo(); + const code = await runKtxMongoQuery({ ...baseArgs, connectionId: 'nope' }, io, { + loadProject: async () => project, + createScanConnector: (async () => mongoConnector()) as never, + }); + expect(code).toBe(1); + expect(err().length).toBeGreaterThan(0); + }); +}); From 6563fddf87d1e3bbebf030a76c3638cd6954a697 Mon Sep 17 00:00:00 2001 From: Kevin Messiaen <114553769+kevinmessiaen@users.noreply.github.com> Date: Mon, 6 Jul 2026 20:12:52 +0700 Subject: [PATCH 12/17] feat(cli): register ktx mongo-query command --- packages/cli/src/cli-program.ts | 2 + packages/cli/src/cli-runtime.ts | 2 + .../cli/src/commands/mongo-query-commands.ts | 84 +++++++++++++++++++ .../commands/mongo-query-commands.test.ts | 36 ++++++++ 4 files changed, 124 insertions(+) create mode 100644 packages/cli/src/commands/mongo-query-commands.ts create mode 100644 packages/cli/test/commands/mongo-query-commands.test.ts diff --git a/packages/cli/src/cli-program.ts b/packages/cli/src/cli-program.ts index c30711c85..6fc7affdb 100644 --- a/packages/cli/src/cli-program.ts +++ b/packages/cli/src/cli-program.ts @@ -8,6 +8,7 @@ import { registerConnectionCommands } from './commands/connection-commands.js'; import { registerIngestCommands } from './commands/ingest-commands.js'; import { registerWikiCommands } from './commands/knowledge-commands.js'; import { registerMcpCommands } from './commands/mcp-commands.js'; +import { registerMongoQueryCommands } from './commands/mongo-query-commands.js'; import { registerSetupCommands } from './commands/setup-commands.js'; import { registerSlCommands } from './commands/sl-commands.js'; import { registerSqlCommands } from './commands/sql-commands.js'; @@ -505,6 +506,7 @@ export function buildKtxProgram(options: BuildKtxProgramOptions): Command { registerWikiCommands(program, context); registerSlCommands(program, context); registerSqlCommands(program, context); + registerMongoQueryCommands(program, context); registerStatusCommands(program, context); registerMcpCommands(program, context); registerAdminCommands(program, context); diff --git a/packages/cli/src/cli-runtime.ts b/packages/cli/src/cli-runtime.ts index 89c7c11d2..fc59293b2 100644 --- a/packages/cli/src/cli-runtime.ts +++ b/packages/cli/src/cli-runtime.ts @@ -4,6 +4,7 @@ import type { KtxConnectionArgs } from './connection.js'; import type { KtxAdminReindexArgs } from './admin-reindex.js'; import type { KtxDoctorArgs } from './doctor.js'; import type { KtxKnowledgeArgs } from './knowledge.js'; +import type { KtxMongoQueryArgs } from './mongo-query.js'; import type { KtxPublicIngestArgs } from './public-ingest.js'; import type { KtxRuntimeArgs } from './runtime.js'; import type { KtxSetupArgs } from './setup.js'; @@ -39,6 +40,7 @@ export interface KtxCliDeps { knowledge?: (args: KtxKnowledgeArgs, io: KtxCliIo) => Promise; sl?: (args: KtxSlArgs, io: KtxCliIo) => Promise; sql?: (args: KtxSqlArgs, io: KtxCliIo) => Promise; + mongoQuery?: (args: KtxMongoQueryArgs, io: KtxCliIo) => Promise; mcp?: { startDaemon?: typeof import('./managed-mcp-daemon.js').startKtxMcpDaemon; stopDaemon?: typeof import('./managed-mcp-daemon.js').stopKtxMcpDaemon; diff --git a/packages/cli/src/commands/mongo-query-commands.ts b/packages/cli/src/commands/mongo-query-commands.ts new file mode 100644 index 000000000..c22405c7e --- /dev/null +++ b/packages/cli/src/commands/mongo-query-commands.ts @@ -0,0 +1,84 @@ +import { type Command, InvalidArgumentError, Option } from '@commander-js/extra-typings'; +import { type KtxCliCommandContext, resolveCommandProjectDir } from '../cli-program.js'; +import { profileMark } from '../startup-profile.js'; + +profileMark('module:commands/mongo-query-commands'); + +const DEFAULT_LIMIT = 1000; +const LIMIT_CAP = 10_000; + +function parseLimitOption(value: string): number { + const parsed = Number(value); + if (!Number.isInteger(parsed) || parsed < 1 || parsed > LIMIT_CAP) { + throw new InvalidArgumentError(`must be an integer between 1 and ${LIMIT_CAP}`); + } + return parsed; +} + +/** @internal exported only for unit testing */ +export function parsePipelineArgument(raw: string): Record[] { + let parsed: unknown; + try { + parsed = JSON.parse(raw); + } catch { + throw new InvalidArgumentError('must be a JSON aggregation pipeline array, e.g. \'[{"$match":{"city":"NY"}}]\''); + } + if ( + !Array.isArray(parsed) || + !parsed.every((stage) => typeof stage === 'object' && stage !== null && !Array.isArray(stage)) + ) { + throw new InvalidArgumentError('must be a JSON array of pipeline-stage objects'); + } + return parsed as Record[]; +} + +export function registerMongoQueryCommands(program: Command, context: KtxCliCommandContext): void { + program + .command('mongo-query') + .description('Fetch rows from a MongoDB connection by running an aggregation pipeline') + .argument('', 'MongoDB aggregation pipeline as a JSON array, e.g. \'[{"$match":{"city":"NY"}}]\'', parsePipelineArgument) + .requiredOption('-c, --connection ', 'ktx connection id') + .requiredOption('--collection ', 'Collection to query') + .option('--database ', "Database name (defaults to the connection's first configured database)") + .option('--limit ', 'Maximum documents to return', parseLimitOption, DEFAULT_LIMIT) + .addOption( + new Option('--output ', 'Output mode: pretty (default), plain (TSV), or json').choices([ + 'pretty', + 'plain', + 'json', + ]), + ) + .option('--json', 'Shortcut for --output=json (overrides --output)', false) + .action( + async ( + pipeline: Record[], + options: { + connection: string; + collection: string; + database?: string; + limit: number; + output?: 'pretty' | 'plain' | 'json'; + json?: boolean; + }, + command, + ) => { + const runner = context.deps.mongoQuery ?? (await import('../mongo-query.js')).runKtxMongoQuery; + context.setExitCode( + await runner( + { + projectDir: resolveCommandProjectDir(command), + connectionId: options.connection, + collection: options.collection, + ...(options.database ? { database: options.database } : {}), + pipeline, + limit: options.limit, + output: options.output, + json: options.json === true, + cliVersion: context.packageInfo.version, + }, + context.io, + ), + ); + }, + ); +} diff --git a/packages/cli/test/commands/mongo-query-commands.test.ts b/packages/cli/test/commands/mongo-query-commands.test.ts new file mode 100644 index 000000000..61199754d --- /dev/null +++ b/packages/cli/test/commands/mongo-query-commands.test.ts @@ -0,0 +1,36 @@ +import { InvalidArgumentError } from '@commander-js/extra-typings'; +import { describe, expect, it } from 'vitest'; +import { parsePipelineArgument } from '../../src/commands/mongo-query-commands.js'; + +describe('parsePipelineArgument', () => { + it('parses a valid JSON array of pipeline-stage objects', () => { + expect(parsePipelineArgument('[{"$match":{"city":"NY"}},{"$limit":10}]')).toEqual([ + { $match: { city: 'NY' } }, + { $limit: 10 }, + ]); + }); + + it('parses an empty pipeline array', () => { + expect(parsePipelineArgument('[]')).toEqual([]); + }); + + it('throws InvalidArgumentError on invalid JSON', () => { + expect(() => parsePipelineArgument('{not json')).toThrow(InvalidArgumentError); + }); + + it('throws InvalidArgumentError when the value is not an array', () => { + expect(() => parsePipelineArgument('{}')).toThrow(InvalidArgumentError); + }); + + it('throws InvalidArgumentError when the array contains a non-object stage', () => { + expect(() => parsePipelineArgument('[{"$match":{}}, "not-a-stage"]')).toThrow(InvalidArgumentError); + }); + + it('throws InvalidArgumentError when the array contains a nested array stage', () => { + expect(() => parsePipelineArgument('[[1,2,3]]')).toThrow(InvalidArgumentError); + }); + + it('throws InvalidArgumentError when the array contains null', () => { + expect(() => parsePipelineArgument('[null]')).toThrow(InvalidArgumentError); + }); +}); From 10de18d63fa80b9502816c4166408f5e20d20e07 Mon Sep 17 00:00:00 2001 From: Kevin Messiaen <114553769+kevinmessiaen@users.noreply.github.com> Date: Mon, 6 Jul 2026 20:16:28 +0700 Subject: [PATCH 13/17] refactor(cli): simplify optional database passthrough in mongo-query command --- packages/cli/src/commands/mongo-query-commands.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/cli/src/commands/mongo-query-commands.ts b/packages/cli/src/commands/mongo-query-commands.ts index c22405c7e..30a6bc211 100644 --- a/packages/cli/src/commands/mongo-query-commands.ts +++ b/packages/cli/src/commands/mongo-query-commands.ts @@ -69,7 +69,7 @@ export function registerMongoQueryCommands(program: Command, context: KtxCliComm projectDir: resolveCommandProjectDir(command), connectionId: options.connection, collection: options.collection, - ...(options.database ? { database: options.database } : {}), + database: options.database, pipeline, limit: options.limit, output: options.output, From afb8a2d2346b46c20be901cc616d0288ac50605e Mon Sep 17 00:00:00 2001 From: Kevin Messiaen <114553769+kevinmessiaen@users.noreply.github.com> Date: Mon, 6 Jul 2026 20:20:46 +0700 Subject: [PATCH 14/17] docs: document ktx mongo-query command --- .../docs/cli-reference/ktx-mongo-query.mdx | 103 ++++++++++++++++++ .../content/docs/cli-reference/meta.json | 1 + .../docs/integrations/primary-sources.mdx | 8 ++ 3 files changed, 112 insertions(+) create mode 100644 docs-site/content/docs/cli-reference/ktx-mongo-query.mdx diff --git a/docs-site/content/docs/cli-reference/ktx-mongo-query.mdx b/docs-site/content/docs/cli-reference/ktx-mongo-query.mdx new file mode 100644 index 000000000..86f2378d6 --- /dev/null +++ b/docs-site/content/docs/cli-reference/ktx-mongo-query.mdx @@ -0,0 +1,103 @@ +--- +title: "ktx mongo-query" +description: "Fetch rows from a MongoDB connection by running an aggregation pipeline." +--- + +Run a MongoDB aggregation pipeline against a `mongodb` connection in your +**ktx** project. MongoDB is not a SQL source, so `ktx sql` refuses a mongodb +connection — `ktx mongo-query` is the row-fetch path for MongoDB, and the +`mongo_query` MCP tool is its agent-facing equivalent. + +## Command signature + +Use `ktx mongo-query` with a required connection id, a required collection +name, and a positional aggregation pipeline given as a JSON array of stage +objects. + +```bash +ktx mongo-query '' --connection --collection [options] +``` + +## Options + +| Flag | Description | Default | +|------|-------------|---------| +| `-c`, `--connection ` | **ktx** database connection id. Required. | - | +| `--collection ` | Collection to query. Required. | - | +| `--database ` | Database name. | Connection's first configured database | +| `--limit ` | Maximum documents to return. Must be between `1` and `10000`. | `1000` | +| `--output ` | Output mode: `pretty`, `plain` (TSV), or `json`. | `pretty` | +| `--json` | Shortcut for `--output=json` (overrides `--output`). | `false` | + +## Examples + +Quote the pipeline JSON in shell scripts and when it contains spaces or +punctuation. + +```bash +# Match and project fields from the business collection +ktx mongo-query '[{"$match":{"city":"Indianapolis"}},{"$project":{"business_id":1,"city":1}}]' \ + --connection mongo --collection business --limit 500 + +# Query a non-default database +ktx mongo-query '[{"$match":{"status":"open"}}]' \ + --connection mongo \ + --collection orders \ + --database analytics + +# Print JSON for agents or scripts +ktx mongo-query '[{"$match":{"status":"open"}}]' \ + --connection mongo \ + --collection orders \ + --json + +# Print TSV rows +ktx mongo-query '[{"$limit":10}]' \ + -c mongo \ + --collection orders \ + --output plain +``` + +## Output + +Pretty output prints aligned columns and a final row count. + +```text +business_id city +----------- ------------ +1001 Indianapolis +1002 Indianapolis + +2 rows +``` + +Plain output prints a TSV header row followed by TSV data rows. + +```text +business_id city +1001 Indianapolis +1002 Indianapolis +``` + +JSON output preserves connection id, headers, rows, and row count. + +```json +{ + "connectionId": "mongo", + "headers": ["business_id", "city"], + "rows": [ + [1001, "Indianapolis"], + [1002, "Indianapolis"] + ], + "rowCount": 2 +} +``` + +## Common errors + +| Error | Cause | Recovery | +|-------|-------|----------| +| `must be a JSON aggregation pipeline array` | The pipeline argument is not valid JSON. | Pass a JSON array of pipeline-stage objects. | +| `must be a JSON array of pipeline-stage objects` | The parsed JSON is not an array of objects. | Wrap each stage as an object inside a JSON array, e.g. `[{"$match":{...}}]`. | +| `Connection "" is not configured in ktx.yaml` | The connection id is wrong or missing from the project. | Run `ktx connection list` and retry with an exact id. | +| `driver "" is not a MongoDB connection` | The connection id resolves to a non-mongodb driver. | Use a `mongodb` connection id, or use `ktx sql` for SQL-queryable drivers. | diff --git a/docs-site/content/docs/cli-reference/meta.json b/docs-site/content/docs/cli-reference/meta.json index 2902f2c67..5165c539d 100644 --- a/docs-site/content/docs/cli-reference/meta.json +++ b/docs-site/content/docs/cli-reference/meta.json @@ -8,6 +8,7 @@ "ktx-ingest", "ktx-sl", "ktx-sql", + "ktx-mongo-query", "ktx-wiki", "ktx-status", "ktx-mcp", diff --git a/docs-site/content/docs/integrations/primary-sources.mdx b/docs-site/content/docs/integrations/primary-sources.mdx index b553a3a7e..4617a86c7 100644 --- a/docs-site/content/docs/integrations/primary-sources.mdx +++ b/docs-site/content/docs/integrations/primary-sources.mdx @@ -735,6 +735,7 @@ nullability from how often the field is present: | Table sampling | Yes | Reads the most recent documents | | Nested analysis | Yes | Sub-documents and arrays modeled as opaque `json` | | Read-only SQL (`ktx sql`) | No | MongoDB is not a SQL source | +| Row queries (`ktx mongo-query`) | Yes | Runs a read-only aggregation pipeline against a collection | ### Dialect notes @@ -746,6 +747,13 @@ nullability from how often the field is present: - `sample_size` trades inference coverage for speed; raise it for collections with highly variable documents +### Querying rows + +MongoDB is not SQL-queryable, so `ktx sql` refuses a mongodb connection. Use +[`ktx mongo-query`](/docs/cli-reference/ktx-mongo-query) to fetch rows by +running a MongoDB aggregation pipeline, or the `mongo_query` MCP tool for the +same capability from an agent client. + ## Common errors | Error or symptom | Likely cause | Recovery | From a236ec9d9150dbb8b9d05e2b6d6795ccadb46bed Mon Sep 17 00:00:00 2001 From: Kevin Messiaen <114553769+kevinmessiaen@users.noreply.github.com> Date: Mon, 6 Jul 2026 20:29:04 +0700 Subject: [PATCH 15/17] feat(cli): export runKtxMongoQuery as public API to match runKtxSql --- packages/cli/src/index.ts | 1 + 1 file changed, 1 insertion(+) diff --git a/packages/cli/src/index.ts b/packages/cli/src/index.ts index 236bdf699..740d8763d 100644 --- a/packages/cli/src/index.ts +++ b/packages/cli/src/index.ts @@ -34,6 +34,7 @@ export type { export { runKtxSetupSourcesStep } from './setup-sources.js'; export { runKtxRuntime, type KtxRuntimeArgs, type KtxRuntimeDeps } from './runtime.js'; export { runKtxSql, type KtxSqlArgs, type KtxSqlDeps } from './sql.js'; +export { runKtxMongoQuery, type KtxMongoQueryArgs, type KtxMongoQueryCliDeps } from './mongo-query.js'; export { allocateDaemonPort, readManagedPythonDaemonStatus, From 0e2e6c7d90269a2d8118a10d706fade74f3a6713 Mon Sep 17 00:00:00 2001 From: Kevin Messiaen <114553769+kevinmessiaen@users.noreply.github.com> Date: Sat, 11 Jul 2026 21:19:55 +0700 Subject: [PATCH 16/17] refactor(mongo-query): consolidate query types and drop unsafe cast Route the Mongo read path through one canonical request/result shape and remove the `as unknown as` narrowing flagged in review. - Add optional executeQuery to KtxScanConnector and narrow by capability in the runner, replacing the double cast with a driver + method guard. - Move KtxMongoQueryInput/KtxMongoQueryResult into context/scan/types.ts; delete the duplicate runner request type, the inline MCP-port input, and the KtxMongoQueryResponse alias. KtxMongoQueryArgs now extends the base. - Inline the pass-through executeMongoQuery wrapper. - Correct the --output default in the mongo-query and sql CLI docs: it is TTY/CI/KTX_OUTPUT-derived, not unconditionally pretty. - Add tests: database override, empty result, parseLimitOption bounds, bigint/plural-row/empty-header rendering, and mongo_query_completed telemetry fields on both outcomes. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01Ga2tvQ2YCEphmY8HM5Guzq --- .../docs/cli-reference/ktx-mongo-query.mdx | 2 +- .../content/docs/cli-reference/ktx-sql.mdx | 2 +- .../cli/src/commands/mongo-query-commands.ts | 3 +- .../cli/src/connectors/mongodb/connector.ts | 22 +++--------- .../src/context/mcp/local-project-ports.ts | 12 ++----- .../cli/src/context/mcp/mongo-query-runner.ts | 20 +++-------- packages/cli/src/context/mcp/types.ts | 9 ++--- packages/cli/src/context/scan/types.ts | 15 ++++++++ packages/cli/src/mongo-query.ts | 8 ++--- .../commands/mongo-query-commands.test.ts | 25 ++++++++++++- .../test/connectors/mongodb/connector.test.ts | 19 ++++++++++ packages/cli/test/context/mcp/server.test.ts | 4 +-- packages/cli/test/io/result-table.test.ts | 28 +++++++++++++-- packages/cli/test/mongo-query.test.ts | 35 ++++++++++++++++++- 14 files changed, 139 insertions(+), 65 deletions(-) diff --git a/docs-site/content/docs/cli-reference/ktx-mongo-query.mdx b/docs-site/content/docs/cli-reference/ktx-mongo-query.mdx index 86f2378d6..a8aaf48bd 100644 --- a/docs-site/content/docs/cli-reference/ktx-mongo-query.mdx +++ b/docs-site/content/docs/cli-reference/ktx-mongo-query.mdx @@ -26,7 +26,7 @@ ktx mongo-query '' --connection --collection [options | `--collection ` | Collection to query. Required. | - | | `--database ` | Database name. | Connection's first configured database | | `--limit ` | Maximum documents to return. Must be between `1` and `10000`. | `1000` | -| `--output ` | Output mode: `pretty`, `plain` (TSV), or `json`. | `pretty` | +| `--output ` | Output mode: `pretty`, `plain` (TSV), or `json`. | `pretty` on a terminal, `plain` when piped or in CI (override with `KTX_OUTPUT`) | | `--json` | Shortcut for `--output=json` (overrides `--output`). | `false` | ## Examples diff --git a/docs-site/content/docs/cli-reference/ktx-sql.mdx b/docs-site/content/docs/cli-reference/ktx-sql.mdx index d78864e2a..aa0ef959e 100644 --- a/docs-site/content/docs/cli-reference/ktx-sql.mdx +++ b/docs-site/content/docs/cli-reference/ktx-sql.mdx @@ -24,7 +24,7 @@ JSON. |------|-------------|---------| | `-c`, `--connection ` | **ktx** database connection id. Required. | - | | `--max-rows ` | Maximum rows to return. Must be between `1` and `10000`. | `1000` | -| `--output ` | Output mode: `pretty`, `plain` (TSV), or `json`. | `pretty` | +| `--output ` | Output mode: `pretty`, `plain` (TSV), or `json`. | `pretty` on a terminal, `plain` when piped or in CI (override with `KTX_OUTPUT`) | | `--json` | Shortcut for `--output=json` (overrides `--output`). | `false` | ## Examples diff --git a/packages/cli/src/commands/mongo-query-commands.ts b/packages/cli/src/commands/mongo-query-commands.ts index 30a6bc211..206c5df9a 100644 --- a/packages/cli/src/commands/mongo-query-commands.ts +++ b/packages/cli/src/commands/mongo-query-commands.ts @@ -7,7 +7,8 @@ profileMark('module:commands/mongo-query-commands'); const DEFAULT_LIMIT = 1000; const LIMIT_CAP = 10_000; -function parseLimitOption(value: string): number { +/** @internal exported only for unit testing */ +export function parseLimitOption(value: string): number { const parsed = Number(value); if (!Number.isInteger(parsed) || parsed < 1 || parsed > LIMIT_CAP) { throw new InvalidArgumentError(`must be an integer between 1 and ${LIMIT_CAP}`); diff --git a/packages/cli/src/connectors/mongodb/connector.ts b/packages/cli/src/connectors/mongodb/connector.ts index d6d185c82..6d45ed697 100644 --- a/packages/cli/src/connectors/mongodb/connector.ts +++ b/packages/cli/src/connectors/mongodb/connector.ts @@ -6,6 +6,8 @@ import { type KtxColumnSampleInput, type KtxColumnSampleResult, type KtxConnectorTestResult, + type KtxMongoQueryInput, + type KtxMongoQueryResult, type KtxScanConnector, type KtxScanContext, type KtxScanInput, @@ -39,20 +41,6 @@ export interface KtxMongoListedCollection { type?: string; } -export interface KtxMongoQueryInput { - connectionId: string; - collection: string; - database?: string; - pipeline: Record[]; - limit: number; -} - -export interface KtxMongoQueryResult { - headers: string[]; - rows: unknown[][]; - rowCount: number; -} - interface KtxMongoFindOptions { sort: Record; limit: number; @@ -67,7 +55,7 @@ export interface KtxMongoClient { aggregate( databaseName: string, collectionName: string, - pipeline: Record[], + pipeline: readonly Record[], options: { limit: number }, ): Promise; ping(databaseName: string): Promise; @@ -129,7 +117,7 @@ class DefaultMongoClient implements KtxMongoClient { async aggregate( databaseName: string, collectionName: string, - pipeline: Record[], + pipeline: readonly Record[], options: { limit: number }, ): Promise { const client = await this.connectedClient(); @@ -223,7 +211,7 @@ function unionDocumentKeys(documents: readonly KtxMongoDocument[]): string[] { } // MongoDB aggregation write stages ($out/$merge) persist results; mongo_query is read-only. -function assertReadOnlyPipeline(pipeline: Record[]): void { +function assertReadOnlyPipeline(pipeline: readonly Record[]): void { for (const stage of pipeline) { if ('$out' in stage || '$merge' in stage) { throw new Error('mongo_query pipelines must be read-only; $out and $merge stages are not allowed.'); diff --git a/packages/cli/src/context/mcp/local-project-ports.ts b/packages/cli/src/context/mcp/local-project-ports.ts index 43901685a..d1cc28e47 100644 --- a/packages/cli/src/context/mcp/local-project-ports.ts +++ b/packages/cli/src/context/mcp/local-project-ports.ts @@ -15,7 +15,6 @@ import { KtxDaemonComputeError, type KtxSemanticLayerComputePort } from '../../c import type { KtxLocalProject } from '../../context/project/project.js'; import { createKtxEntityDetailsService } from '../../context/scan/entity-details.js'; import type { LocalScanMcpOptions } from '../../context/scan/local-scan.js'; -import type { KtxScanConnector } from '../../context/scan/types.js'; import { createKtxDiscoverDataService } from '../../context/search/discover.js'; import { sqlAnalysisDialectForDriver } from '../../context/sql-analysis/dialect.js'; import type { SqlAnalysisPort } from '../../context/sql-analysis/ports.js'; @@ -26,7 +25,7 @@ import { assertSafeConnectionId } from '../../context/sl/source-files.js'; import { assertConfiguredConnectionId } from '../../context/connections/configured-connections.js'; import { readLocalKnowledgePage, searchLocalKnowledgePages } from '../wiki/local-knowledge.js'; import { runMongoQuery } from './mongo-query-runner.js'; -import type { KtxMcpContextPorts, KtxMcpProgressCallback, KtxMongoQueryResponse, KtxSqlExecutionResponse } from './types.js'; +import type { KtxMcpContextPorts, KtxMcpProgressCallback, KtxSqlExecutionResponse } from './types.js'; interface CreateLocalProjectMcpContextPortsOptions { semanticLayerCompute?: KtxSemanticLayerComputePort; @@ -62,13 +61,6 @@ function projectHasMongoConnection(project: KtxLocalProject): boolean { ); } -async function executeMongoQuery( - createConnector: (connectionId: string) => Promise | KtxScanConnector, - input: { connectionId: string; collection: string; database?: string; pipeline: Record[]; limit: number }, -): Promise { - return runMongoQuery(createConnector, input); -} - async function executeValidatedReadOnlySql( project: KtxLocalProject, options: CreateLocalProjectMcpContextPortsOptions, @@ -259,7 +251,7 @@ export function createLocalProjectMcpContextPorts( ports.mongoQuery = { async execute(input) { try { - return await executeMongoQuery(mongoCreateConnector, input); + return await runMongoQuery(mongoCreateConnector, input); } catch (error) { throwClassifiedQueryError(error); } diff --git a/packages/cli/src/context/mcp/mongo-query-runner.ts b/packages/cli/src/context/mcp/mongo-query-runner.ts index fdd0b0820..65136cd86 100644 --- a/packages/cli/src/context/mcp/mongo-query-runner.ts +++ b/packages/cli/src/context/mcp/mongo-query-runner.ts @@ -1,34 +1,22 @@ -import type { KtxMongoDbScanConnector, KtxMongoQueryResult } from '../../connectors/mongodb/connector.js'; import { KtxExpectedError } from '../../errors.js'; import { assertSafeConnectionId } from '../sl/source-files.js'; -import type { KtxScanConnector } from '../scan/types.js'; - -export interface KtxMongoQueryRequest { - connectionId: string; - collection: string; - database?: string; - pipeline: Record[]; - limit: number; -} +import type { KtxMongoQueryInput, KtxMongoQueryResult, KtxScanConnector } from '../scan/types.js'; /** Single Mongo-read execution seam: resolve the connector, guard the driver, run the pipeline. */ export async function runMongoQuery( createConnector: (connectionId: string) => Promise | KtxScanConnector, - input: KtxMongoQueryRequest, + input: KtxMongoQueryInput, ): Promise { const connectionId = assertSafeConnectionId(input.connectionId); let connector: KtxScanConnector | null = null; try { connector = await createConnector(connectionId); - if (connector.driver !== 'mongodb') { + if (connector.driver !== 'mongodb' || typeof connector.executeQuery !== 'function') { throw new KtxExpectedError( `Connection "${connectionId}" driver "${connector.driver}" is not a MongoDB connection; mongo_query serves mongodb connections only.`, ); } - return await (connector as unknown as KtxMongoDbScanConnector).executeQuery( - { connectionId, collection: input.collection, database: input.database, pipeline: input.pipeline, limit: input.limit }, - { runId: 'mongo-query' }, - ); + return await connector.executeQuery({ ...input, connectionId }, { runId: 'mongo-query' }); } finally { await connector?.cleanup?.(); } diff --git a/packages/cli/src/context/mcp/types.ts b/packages/cli/src/context/mcp/types.ts index 4ef12272a..f05c49b47 100644 --- a/packages/cli/src/context/mcp/types.ts +++ b/packages/cli/src/context/mcp/types.ts @@ -5,7 +5,7 @@ import type { KtxEntityDetailsInput, KtxEntityDetailsResponse } from '../scan/en import type { KtxDiscoverDataInput, KtxDiscoverDataResponse } from '../../context/search/discover.js'; import type { KtxDictionarySearchInput, KtxDictionarySearchResponse } from '../../context/sl/dictionary-search.js'; import type { SemanticLayerQueryInput } from '../../context/sl/types.js'; -import type { KtxMongoQueryResult } from '../../connectors/mongodb/connector.js'; +import type { KtxMongoQueryInput, KtxMongoQueryResult } from '../scan/types.js'; import type { WikiSearchLaneSummary, WikiSearchMatchReason } from '../../context/wiki/types.js'; interface KtxMcpTextContent { @@ -181,14 +181,9 @@ export interface KtxSqlExecutionMcpPort { ): Promise; } -export type KtxMongoQueryResponse = KtxMongoQueryResult; - /** @internal */ export interface KtxMongoQueryMcpPort { - execute( - input: { connectionId: string; collection: string; database?: string; pipeline: Record[]; limit: number }, - options?: { onProgress?: KtxMcpProgressCallback }, - ): Promise; + execute(input: KtxMongoQueryInput, options?: { onProgress?: KtxMcpProgressCallback }): Promise; } /** @internal */ diff --git a/packages/cli/src/context/scan/types.ts b/packages/cli/src/context/scan/types.ts index bf72558c4..3d39307b5 100644 --- a/packages/cli/src/context/scan/types.ts +++ b/packages/cli/src/context/scan/types.ts @@ -293,6 +293,20 @@ export interface KtxReadOnlyQueryInput { maxRows?: number; } +export interface KtxMongoQueryInput { + connectionId: string; + collection: string; + database?: string; + pipeline: readonly Record[]; + limit: number; +} + +export interface KtxMongoQueryResult { + headers: string[]; + rows: unknown[][]; + rowCount: number; +} + export interface KtxQueryResult { headers: string[]; headerTypes?: string[]; @@ -346,6 +360,7 @@ export interface KtxScanConnector { sampleTable?(input: KtxTableSampleInput, ctx: KtxScanContext): Promise; columnStats?(input: KtxColumnStatsInput, ctx: KtxScanContext): Promise; executeReadOnly?(input: KtxReadOnlyQueryInput, ctx: KtxScanContext): Promise; + executeQuery?(input: KtxMongoQueryInput, ctx: KtxScanContext): Promise; cleanup?(): Promise; } diff --git a/packages/cli/src/mongo-query.ts b/packages/cli/src/mongo-query.ts index f529df889..17b1e33ea 100644 --- a/packages/cli/src/mongo-query.ts +++ b/packages/cli/src/mongo-query.ts @@ -1,6 +1,7 @@ import { runMongoQuery } from './context/mcp/mongo-query-runner.js'; import { resolveConfiguredConnection } from './context/connections/resolve-connection.js'; import { loadKtxProject, type KtxLocalProject } from './context/project/project.js'; +import type { KtxMongoQueryInput } from './context/scan/types.js'; import type { KtxCliIo } from './cli-runtime.js'; import { type KtxOutputMode, resolveOutputMode } from './io/mode.js'; import { type KtxResultTable, printResultTable } from './io/result-table.js'; @@ -13,13 +14,8 @@ import { scrubErrorClass } from './telemetry/scrubber.js'; profileMark('module:mongo-query'); -export type KtxMongoQueryArgs = { +export type KtxMongoQueryArgs = KtxMongoQueryInput & { projectDir: string; - connectionId: string; - collection: string; - database?: string; - pipeline: Record[]; - limit: number; output?: KtxOutputMode; json?: boolean; cliVersion: string; diff --git a/packages/cli/test/commands/mongo-query-commands.test.ts b/packages/cli/test/commands/mongo-query-commands.test.ts index 61199754d..c04d988d4 100644 --- a/packages/cli/test/commands/mongo-query-commands.test.ts +++ b/packages/cli/test/commands/mongo-query-commands.test.ts @@ -1,6 +1,6 @@ import { InvalidArgumentError } from '@commander-js/extra-typings'; import { describe, expect, it } from 'vitest'; -import { parsePipelineArgument } from '../../src/commands/mongo-query-commands.js'; +import { parseLimitOption, parsePipelineArgument } from '../../src/commands/mongo-query-commands.js'; describe('parsePipelineArgument', () => { it('parses a valid JSON array of pipeline-stage objects', () => { @@ -34,3 +34,26 @@ describe('parsePipelineArgument', () => { expect(() => parsePipelineArgument('[null]')).toThrow(InvalidArgumentError); }); }); + +describe('parseLimitOption', () => { + it('accepts the boundary values 1 and 10000', () => { + expect(parseLimitOption('1')).toBe(1); + expect(parseLimitOption('10000')).toBe(10000); + }); + + it('rejects a non-numeric value', () => { + expect(() => parseLimitOption('abc')).toThrow(InvalidArgumentError); + }); + + it('rejects a non-integer value', () => { + expect(() => parseLimitOption('1.5')).toThrow(InvalidArgumentError); + }); + + it('rejects a value below 1', () => { + expect(() => parseLimitOption('0')).toThrow(InvalidArgumentError); + }); + + it('rejects a value above the cap', () => { + expect(() => parseLimitOption('10001')).toThrow(InvalidArgumentError); + }); +}); diff --git a/packages/cli/test/connectors/mongodb/connector.test.ts b/packages/cli/test/connectors/mongodb/connector.test.ts index 7089ee218..e46d4bc2b 100644 --- a/packages/cli/test/connectors/mongodb/connector.test.ts +++ b/packages/cli/test/connectors/mongodb/connector.test.ts @@ -295,4 +295,23 @@ describe('KtxMongoDbScanConnector.executeQuery', () => { ), ).rejects.toThrow(/must be read-only/); }); + + it('runs against the requested database when input.database is given', async () => { + const { factory, client } = fakeClientFactory(); + await connector(baseConnection, factory).executeQuery( + { connectionId: 'mongo-prod', collection: 'orders', database: 'analytics', pipeline: [], limit: 5 }, + { runId: 't' }, + ); + expect(client.aggregate).toHaveBeenCalledWith('analytics', 'orders', [], { limit: 5 }); + }); + + it('returns an empty result with no headers when the pipeline matches no documents', async () => { + const { factory, client } = fakeClientFactory(); + (client.aggregate as ReturnType).mockResolvedValueOnce([]); + const result = await connector(baseConnection, factory).executeQuery( + { connectionId: 'mongo-prod', collection: 'users', pipeline: [{ $match: { age: { $gte: 999 } } }], limit: 50 }, + { runId: 't' }, + ); + expect(result).toEqual({ headers: [], rows: [], rowCount: 0 }); + }); }); diff --git a/packages/cli/test/context/mcp/server.test.ts b/packages/cli/test/context/mcp/server.test.ts index 69091aab9..43f9b3204 100644 --- a/packages/cli/test/context/mcp/server.test.ts +++ b/packages/cli/test/context/mcp/server.test.ts @@ -22,12 +22,12 @@ import type { KtxMcpContextPorts, KtxMcpToolHandlerContext, KtxMongoQueryMcpPort, - KtxMongoQueryResponse, KtxSemanticLayerMcpPort, KtxSqlExecutionMcpPort, KtxSqlExecutionResponse, MemoryIngestPort, } from '../../../src/context/mcp/types.js'; +import type { KtxMongoQueryResult } from '../../../src/context/scan/types.js'; const reportExceptionMock = vi.hoisted(() => vi.fn(async () => {})); @@ -499,7 +499,7 @@ describe('createKtxMcpServer', () => { it('registers mongo_query when the host provides a MongoDB query port', async () => { const fake = makeFakeServer(); - const response: KtxMongoQueryResponse = { + const response: KtxMongoQueryResult = { headers: ['_id', 'city'], rows: [ ['a1', 'Indianapolis'], diff --git a/packages/cli/test/io/result-table.test.ts b/packages/cli/test/io/result-table.test.ts index 39508d370..9129265ef 100644 --- a/packages/cli/test/io/result-table.test.ts +++ b/packages/cli/test/io/result-table.test.ts @@ -18,6 +18,7 @@ describe('formatValue', () => { expect(formatValue('x')).toBe('x'); expect(formatValue(42)).toBe('42'); expect(formatValue(true)).toBe('true'); + expect(formatValue(10n)).toBe('10'); expect(formatValue({ a: 1 })).toBe(JSON.stringify({ a: 1 })); }); }); @@ -37,10 +38,33 @@ describe('printResultTable', () => { expect(out()).toBe('_id\tcity\na1\tNY\n'); }); - it('pretty mode prints a header rule and a row count', () => { + it('pretty mode prints a header rule and a singular row count', () => { const { io, out } = captureIo(); printResultTable(table, 'pretty', io); expect(out()).toContain('_id'); - expect(out()).toContain('1 row'); + expect(out()).toContain('1 row\n'); + }); + + it('pretty mode aligns multiple rows and pluralizes the row count', () => { + const { io, out } = captureIo(); + const multi = { + connectionId: 'mongo', + headers: ['id', 'city'], + rows: [ + ['1', 'NY'], + ['1000', 'Indianapolis'], + ], + rowCount: 2, + }; + printResultTable(multi, 'pretty', io); + // Column widened to the longest cell ("1000"), so "1" is right-padded to 4 chars. + expect(out()).toContain('1 NY'); + expect(out()).toContain('2 rows\n'); + }); + + it('pretty mode omits the header rule when there are no headers', () => { + const { io, out } = captureIo(); + printResultTable({ connectionId: 'mongo', headers: [], rows: [], rowCount: 0 }, 'pretty', io); + expect(out()).toBe('\n0 rows\n'); }); }); diff --git a/packages/cli/test/mongo-query.test.ts b/packages/cli/test/mongo-query.test.ts index be1265ca2..5b7d17bbc 100644 --- a/packages/cli/test/mongo-query.test.ts +++ b/packages/cli/test/mongo-query.test.ts @@ -1,4 +1,4 @@ -import { describe, expect, it, vi } from 'vitest'; +import { afterEach, describe, expect, it, vi } from 'vitest'; import { runKtxMongoQuery } from '../src/mongo-query.js'; import { KtxMongoDbScanConnector } from '../src/connectors/mongodb/connector.js'; import type { KtxCliIo } from '../src/cli-runtime.js'; @@ -67,3 +67,36 @@ describe('runKtxMongoQuery', () => { expect(err().length).toBeGreaterThan(0); }); }); + +describe('runKtxMongoQuery telemetry', () => { + afterEach(() => { + vi.unstubAllEnvs(); + }); + + it('emits mongo_query_completed with outcome ok and the pipeline stage count on success', async () => { + vi.stubEnv('KTX_TELEMETRY_DEBUG', '1'); + vi.stubEnv('CI', ''); + const { io, err } = captureIo(); + const code = await runKtxMongoQuery({ ...baseArgs, output: 'plain' }, io, { + loadProject: async () => project, + createScanConnector: (async () => mongoConnector()) as never, + }); + expect(code).toBe(0); + expect(err()).toContain('"event":"mongo_query_completed"'); + expect(err()).toContain('"outcome":"ok"'); + expect(err()).toContain('"stageCount":1'); + }); + + it('emits mongo_query_completed with outcome error on failure', async () => { + vi.stubEnv('KTX_TELEMETRY_DEBUG', '1'); + vi.stubEnv('CI', ''); + const { io, err } = captureIo(); + const code = await runKtxMongoQuery({ ...baseArgs, connectionId: 'nope' }, io, { + loadProject: async () => project, + createScanConnector: (async () => mongoConnector()) as never, + }); + expect(code).toBe(1); + expect(err()).toContain('"event":"mongo_query_completed"'); + expect(err()).toContain('"outcome":"error"'); + }); +}); From bcfd562f18738feaa820fad657b76010d6b07ed5 Mon Sep 17 00:00:00 2001 From: Kevin Messiaen <114553769+kevinmessiaen@users.noreply.github.com> Date: Sat, 11 Jul 2026 21:26:10 +0700 Subject: [PATCH 17/17] refactor(mongo-query): forward the query input without re-listing fields Now that KtxMongoQueryArgs extends KtxMongoQueryInput and the runner takes that type, pass the args/input object straight through instead of rebuilding it field by field. assertSafeConnectionId returns its input verbatim, so the `{ ...input, connectionId }` spread was re-asserting an equal value. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01Ga2tvQ2YCEphmY8HM5Guzq --- packages/cli/src/context/mcp/mongo-query-runner.ts | 2 +- packages/cli/src/mongo-query.ts | 8 +------- 2 files changed, 2 insertions(+), 8 deletions(-) diff --git a/packages/cli/src/context/mcp/mongo-query-runner.ts b/packages/cli/src/context/mcp/mongo-query-runner.ts index 65136cd86..2d45dc543 100644 --- a/packages/cli/src/context/mcp/mongo-query-runner.ts +++ b/packages/cli/src/context/mcp/mongo-query-runner.ts @@ -16,7 +16,7 @@ export async function runMongoQuery( `Connection "${connectionId}" driver "${connector.driver}" is not a MongoDB connection; mongo_query serves mongodb connections only.`, ); } - return await connector.executeQuery({ ...input, connectionId }, { runId: 'mongo-query' }); + return await connector.executeQuery(input, { runId: 'mongo-query' }); } finally { await connector?.cleanup?.(); } diff --git a/packages/cli/src/mongo-query.ts b/packages/cli/src/mongo-query.ts index 17b1e33ea..bc0e6e8f1 100644 --- a/packages/cli/src/mongo-query.ts +++ b/packages/cli/src/mongo-query.ts @@ -42,13 +42,7 @@ export async function runKtxMongoQuery( demoConnection = isDemoConnection(args.connectionId, connection); const createScanConnector = deps.createScanConnector ?? createKtxCliScanConnector; - const result = await runMongoQuery((connectionId) => createScanConnector(project!, connectionId), { - connectionId: args.connectionId, - collection: args.collection, - database: args.database, - pipeline: args.pipeline, - limit: args.limit, - }); + const result = await runMongoQuery((connectionId) => createScanConnector(project!, connectionId), args); const output: KtxResultTable = { connectionId: args.connectionId,