refactor(sdk): config validation

packages/sdk/typescript/src/index.ts

@@ -33,10 +33,11 @@ export type {
   CreateQueryOptions,
   PermissionMode,
   PermissionCallback,
-  ExternalMcpServerConfig,
   TransportOptions,
 } from './types/config.js';
 
+export type { ExternalMcpServerConfig } from './types/queryOptionsSchema.js';
+
 export type { QueryOptions } from './query/createQuery.js';
 
 // Protocol types

packages/sdk/typescript/src/query/createQuery.ts

@@ -4,35 +4,16 @@
 
 import type { CLIUserMessage } from '../types/protocol.js';
 import { serializeJsonLine } from '../utils/jsonLines.js';
-import type {
-  CreateQueryOptions,
-  PermissionMode,
-  PermissionCallback,
-  ExternalMcpServerConfig,
-} from '../types/config.js';
+import type { CreateQueryOptions } from '../types/config.js';
 import { ProcessTransport } from '../transport/ProcessTransport.js';
 import { parseExecutableSpec } from '../utils/cliPath.js';
 import { Query } from './Query.js';
+import {
+  QueryOptionsSchema,
+  type QueryOptions,
+} from '../types/queryOptionsSchema.js';
 
-/**
- * Configuration options for creating a Query.
- */
-export type QueryOptions = {
-  cwd?: string;
-  model?: string;
-  pathToQwenExecutable?: string;
-  env?: Record<string, string>;
-  permissionMode?: PermissionMode;
-  canUseTool?: PermissionCallback;
-  mcpServers?: Record<string, ExternalMcpServerConfig>;
-  sdkMcpServers?: Record<
-    string,
-    { connect: (transport: unknown) => Promise<void> }
-  >;
-  abortController?: AbortController;
-  debug?: boolean;
-  stderr?: (message: string) => void;
-};
+export type { QueryOptions };
 
 /**
  * Create a Query instance for interacting with the Qwen CLI.
@@ -146,8 +127,8 @@ export const createQuery = query;
 /**
  * Validate query configuration options and normalize CLI executable details.
  *
- * Performs strict validation for each supported option, including
- * permission mode, callbacks, AbortController usage, and executable spec.
+ * Performs strict validation for each supported option using Zod schema,
+ * including permission mode, callbacks, AbortController usage, and executable spec.
  * Returns the parsed executable description so callers can retain
  * explicit runtime directives (e.g., `bun:/path/to/cli.js`) while still
  * benefiting from early validation and auto-detection fallbacks when the
@@ -156,32 +137,17 @@ export const createQuery = query;
 function validateOptions(
   options: QueryOptions,
 ): ReturnType<typeof parseExecutableSpec> {
-  let parsedExecutable: ReturnType<typeof parseExecutableSpec>;
-
-  // Validate permission mode if provided
-  if (options.permissionMode) {
-    const validModes = ['default', 'plan', 'auto-edit', 'yolo'];
-    if (!validModes.includes(options.permissionMode)) {
-      throw new Error(
-        `Invalid permissionMode: ${options.permissionMode}. Valid values are: ${validModes.join(', ')}`,
-      );
-    }
-  }
-
-  // Validate canUseTool is a function if provided
-  if (options.canUseTool && typeof options.canUseTool !== 'function') {
-    throw new Error('canUseTool must be a function');
-  }
-
-  // Validate abortController is AbortController if provided
-  if (
-    options.abortController &&
-    !(options.abortController instanceof AbortController)
-  ) {
-    throw new Error('abortController must be an AbortController instance');
+  // Validate options using Zod schema
+  const validationResult = QueryOptionsSchema.safeParse(options);
+  if (!validationResult.success) {
+    const errors = validationResult.error.errors
+      .map((err) => `${err.path.join('.')}: ${err.message}`)
+      .join('; ');
+    throw new Error(`Invalid QueryOptions: ${errors}`);
   }
 
   // Validate executable path early to provide clear error messages
+  let parsedExecutable: ReturnType<typeof parseExecutableSpec>;
   try {
     parsedExecutable = parseExecutableSpec(options.pathToQwenExecutable);
   } catch (error) {
@@ -189,7 +155,7 @@ function validateOptions(
     throw new Error(`Invalid pathToQwenExecutable: ${errorMessage}`);
   }
 
-  // Validate no MCP server name conflicts
+  // Validate no MCP server name conflicts (cross-field validation not easily expressible in Zod)
   if (options.mcpServers && options.sdkMcpServers) {
     const externalNames = Object.keys(options.mcpServers);
     const sdkNames = Object.keys(options.sdkMcpServers);

packages/sdk/typescript/src/types/config.ts

@@ -4,6 +4,7 @@
 
 import type { ToolDefinition as ToolDef } from './mcp.js';
 import type { PermissionMode } from './protocol.js';
+import type { ExternalMcpServerConfig } from './queryOptionsSchema.js';
 
 export type { ToolDef as ToolDefinition };
 export type { PermissionMode };
@@ -56,18 +57,6 @@ export type HookConfig = {
   [event: string]: HookMatcher[];
 };
 
-/**
- * External MCP server configuration (spawned by CLI)
- */
-export type ExternalMcpServerConfig = {
-  /** Command to execute (e.g., 'mcp-server-filesystem') */
-  command: string;
-  /** Command-line arguments */
-  args?: string[];
-  /** Environment variables */
-  env?: Record<string, string>;
-};
-
 /**
  * Options for creating a Query instance
  */

packages/sdk/typescript/src/types/queryOptionsSchema.ts

@@ -0,0 +1,60 @@
+/**
+ * Zod schemas for QueryOptions validation
+ */
+
+import { z } from 'zod';
+import type { PermissionCallback } from './config.js';
+
+/**
+ * Schema for external MCP server configuration
+ */
+export const ExternalMcpServerConfigSchema = z.object({
+  command: z.string().min(1, 'Command must be a non-empty string'),
+  args: z.array(z.string()).optional(),
+  env: z.record(z.string(), z.string()).optional(),
+});
+
+/**
+ * Schema for SDK-embedded MCP server configuration
+ */
+export const SdkMcpServerConfigSchema = z.object({
+  connect: z.custom<(transport: unknown) => Promise<void>>(
+    (val) => typeof val === 'function',
+    { message: 'connect must be a function' },
+  ),
+});
+
+/**
+ * Schema for QueryOptions
+ */
+export const QueryOptionsSchema = z
+  .object({
+    cwd: z.string().optional(),
+    model: z.string().optional(),
+    pathToQwenExecutable: z.string().optional(),
+    env: z.record(z.string(), z.string()).optional(),
+    permissionMode: z.enum(['default', 'plan', 'auto-edit', 'yolo']).optional(),
+    canUseTool: z
+      .custom<PermissionCallback>((val) => typeof val === 'function', {
+        message: 'canUseTool must be a function',
+      })
+      .optional(),
+    mcpServers: z.record(z.string(), ExternalMcpServerConfigSchema).optional(),
+    sdkMcpServers: z.record(z.string(), SdkMcpServerConfigSchema).optional(),
+    abortController: z.instanceof(AbortController).optional(),
+    debug: z.boolean().optional(),
+    stderr: z
+      .custom<
+        (message: string) => void
+      >((val) => typeof val === 'function', { message: 'stderr must be a function' })
+      .optional(),
+  })
+  .strict();
+
+/**
+ * Inferred TypeScript types from schemas
+ */
+export type ExternalMcpServerConfig = z.infer<
+  typeof ExternalMcpServerConfigSchema
+>;
+export type QueryOptions = z.infer<typeof QueryOptionsSchema>;

packages/sdk/typescript/test/e2e/control.test.ts

@@ -0,0 +1,269 @@
+import { describe, it, expect } from 'vitest';
+import { query } from '../../src/index.js';
+import {
+  isCLIAssistantMessage,
+  isCLIResultMessage,
+  isCLISystemMessage,
+  type CLIUserMessage,
+} from '../../src/types/protocol.js';
+
+const TEST_CLI_PATH =
+  '/Users/mingholy/Work/Projects/qwen-code/packages/cli/index.ts';
+const TEST_TIMEOUT = 160000;
+
+const SHARED_TEST_OPTIONS = {
+  pathToQwenExecutable: TEST_CLI_PATH,
+  permissionMode: 'yolo' as const,
+};
+
+/**
+ * Factory function that creates a streaming input with a control point.
+ * After the first message is yielded, the generator waits for a resume signal,
+ * allowing the test code to call query instance methods like setModel or setPermissionMode.
+ *
+ * @param firstMessage - The first user message to send
+ * @param secondMessage - The second user message to send after control operations
+ * @returns Object containing the async generator and a resume function
+ */
+function createStreamingInputWithControlPoint(
+  firstMessage: string,
+  secondMessage: string,
+): {
+  generator: AsyncIterable<CLIUserMessage>;
+  resume: () => void;
+} {
+  let resumeResolve: (() => void) | null = null;
+  const resumePromise = new Promise<void>((resolve) => {
+    resumeResolve = resolve;
+  });
+
+  const generator = (async function* () {
+    const sessionId = crypto.randomUUID();
+
+    yield {
+      type: 'user',
+      session_id: sessionId,
+      message: {
+        role: 'user',
+        content: firstMessage,
+      },
+      parent_tool_use_id: null,
+    } as CLIUserMessage;
+
+    await new Promise((resolve) => setTimeout(resolve, 200));
+
+    await resumePromise;
+
+    await new Promise((resolve) => setTimeout(resolve, 200));
+
+    yield {
+      type: 'user',
+      session_id: sessionId,
+      message: {
+        role: 'user',
+        content: secondMessage,
+      },
+      parent_tool_use_id: null,
+    } as CLIUserMessage;
+  })();
+
+  const resume = () => {
+    if (resumeResolve) {
+      resumeResolve();
+    }
+  };
+
+  return { generator, resume };
+}
+
+describe('Control Request/Response (E2E)', () => {
+  describe('System Controller Scope', () => {
+    it(
+      'should set model via control request during streaming input',
+      async () => {
+        const { generator, resume } = createStreamingInputWithControlPoint(
+          'Tell me the model name.',
+          'Tell me the model name now again.',
+        );
+
+        const q = query({
+          prompt: generator,
+          options: {
+            ...SHARED_TEST_OPTIONS,
+            model: 'qwen3-max',
+            debug: false,
+          },
+        });
+
+        try {
+          const resolvers: {
+            first?: () => void;
+            second?: () => void;
+          } = {};
+          const firstResponsePromise = new Promise<void>((resolve) => {
+            resolvers.first = resolve;
+          });
+          const secondResponsePromise = new Promise<void>((resolve) => {
+            resolvers.second = resolve;
+          });
+
+          let firstResponseReceived = false;
+          let secondResponseReceived = false;
+          const systemMessages: Array<{ model?: string }> = [];
+
+          // Consume messages in a single loop
+          (async () => {
+            for await (const message of q) {
+              console.log(JSON.stringify(message));
+              if (isCLISystemMessage(message)) {
+                systemMessages.push({ model: message.model });
+              }
+              if (isCLIAssistantMessage(message)) {
+                if (!firstResponseReceived) {
+                  firstResponseReceived = true;
+                  resolvers.first?.();
+                } else if (!secondResponseReceived) {
+                  secondResponseReceived = true;
+                  resolvers.second?.();
+                }
+              }
+            }
+          })();
+
+          // Wait for first response
+          await Promise.race([
+            firstResponsePromise,
+            new Promise((_, reject) =>
+              setTimeout(
+                () => reject(new Error('Timeout waiting for first response')),
+                TEST_TIMEOUT,
+              ),
+            ),
+          ]);
+
+          expect(firstResponseReceived).toBe(true);
+
+          // Perform control operation: set model
+          await q.setModel('qwen3-vl-plus');
+
+          // Resume the input stream
+          resume();
+
+          // Wait for second response
+          await Promise.race([
+            secondResponsePromise,
+            new Promise((_, reject) =>
+              setTimeout(
+                () => reject(new Error('Timeout waiting for second response')),
+                TEST_TIMEOUT,
+              ),
+            ),
+          ]);
+
+          expect(secondResponseReceived).toBe(true);
+
+          // Verify system messages - model should change from qwen3-max to qwen3-vl-plus
+          expect(systemMessages.length).toBeGreaterThanOrEqual(2);
+          expect(systemMessages[0].model).toBe('qwen3-max');
+          expect(systemMessages[1].model).toBe('qwen3-vl-plus');
+        } finally {
+          await q.close();
+        }
+      },
+      TEST_TIMEOUT,
+    );
+  });
+
+  describe('Permission Controller Scope', () => {
+    it(
+      'should set permission mode via control request during streaming input',
+      async () => {
+        const { generator, resume } = createStreamingInputWithControlPoint(
+          'List files in the current directory',
+          'Now read the package.json file',
+        );
+
+        const q = query({
+          prompt: generator,
+          options: {
+            pathToQwenExecutable: TEST_CLI_PATH,
+            permissionMode: 'default',
+            debug: false,
+          },
+        });
+
+        try {
+          const resolvers: {
+            first?: () => void;
+            second?: () => void;
+          } = {};
+          const firstResponsePromise = new Promise<void>((resolve) => {
+            resolvers.first = resolve;
+          });
+          const secondResponsePromise = new Promise<void>((resolve) => {
+            resolvers.second = resolve;
+          });
+
+          let firstResponseReceived = false;
+          let permissionModeChanged = false;
+          let secondResponseReceived = false;
+
+          // Consume messages in a single loop
+          (async () => {
+            for await (const message of q) {
+              if (
+                isCLIAssistantMessage(message) ||
+                isCLIResultMessage(message)
+              ) {
+                if (!firstResponseReceived) {
+                  firstResponseReceived = true;
+                  resolvers.first?.();
+                } else if (!secondResponseReceived) {
+                  secondResponseReceived = true;
+                  resolvers.second?.();
+                }
+              }
+            }
+          })();
+
+          // Wait for first response
+          await Promise.race([
+            firstResponsePromise,
+            new Promise((_, reject) =>
+              setTimeout(
+                () => reject(new Error('Timeout waiting for first response')),
+                TEST_TIMEOUT,
+              ),
+            ),
+          ]);
+
+          expect(firstResponseReceived).toBe(true);
+
+          // Perform control operation: set permission mode
+          await q.setPermissionMode('yolo');
+          permissionModeChanged = true;
+
+          // Resume the input stream
+          resume();
+
+          // Wait for second response
+          await Promise.race([
+            secondResponsePromise,
+            new Promise((_, reject) =>
+              setTimeout(
+                () => reject(new Error('Timeout waiting for second response')),
+                TEST_TIMEOUT,
+              ),
+            ),
+          ]);
+
+          expect(permissionModeChanged).toBe(true);
+          expect(secondResponseReceived).toBe(true);
+        } finally {
+          await q.close();
+        }
+      },
+      TEST_TIMEOUT,
+    );
+  });
+});