refactor(mongo-query): dedupe result type and enforce read-only pipeline

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<string, unknown>[]): 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<KtxMongoQueryResult> {
     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,

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?.();
   }

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<KtxSqlExecutionResponse>;
 }
 
-export interface KtxMongoQueryResponse {
-  headers: string[];
-  rows: unknown[][];
-  rowCount: number;
-}
+export type KtxMongoQueryResponse = KtxMongoQueryResult;
 
 /** @internal */
 export interface KtxMongoQueryMcpPort {

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