feat(cli): register ktx mongo-query command

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);

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<number>;
   sl?: (args: KtxSlArgs, io: KtxCliIo) => Promise<number>;
   sql?: (args: KtxSqlArgs, io: KtxCliIo) => Promise<number>;
+  mongoQuery?: (args: KtxMongoQueryArgs, io: KtxCliIo) => Promise<number>;
   mcp?: {
     startDaemon?: typeof import('./managed-mcp-daemon.js').startKtxMcpDaemon;
     stopDaemon?: typeof import('./managed-mcp-daemon.js').stopKtxMcpDaemon;

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<string, unknown>[] {
+  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<string, unknown>[];
+}
+
+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('<pipeline>', 'MongoDB aggregation pipeline as a JSON array, e.g. \'[{"$match":{"city":"NY"}}]\'', parsePipelineArgument)
+    .requiredOption('-c, --connection <id>', 'ktx connection id')
+    .requiredOption('--collection <name>', 'Collection to query')
+    .option('--database <name>', "Database name (defaults to the connection's first configured database)")
+    .option('--limit <n>', 'Maximum documents to return', parseLimitOption, DEFAULT_LIMIT)
+    .addOption(
+      new Option('--output <mode>', '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<string, unknown>[],
+        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,
+          ),
+        );
+      },
+    );
+}

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);
+  });
+});