abbyswag/qwen-code
1
0
Fork 0
mirror of https://github.com/QwenLM/qwen-code.git synced 2025-12-21 09:17:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: create draft framework for cli & sdk

Browse Source
This commit is contained in:
mingholy.lmh
2025-11-23 19:36:31 +08:00
parent 6729980b47
commit e1ffaec499
58 changed files with 8982 additions and 668 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
packages/cli/src/nonInteractive/control/ControlDispatcher.ts
View File

@@ -26,7 +26,7 @@
import type { IControlContext } from './ControlContext.js';
import type { IPendingRequestRegistry } from './controllers/baseController.js';
import { SystemController } from './controllers/systemController.js';
// import { PermissionController } from './controllers/permissionController.js';
import { PermissionController } from './controllers/permissionController.js';
// import { MCPController } from './controllers/mcpController.js';
// import { HookController } from './controllers/hookController.js';
import type {
@@ -64,7 +64,7 @@ export class ControlDispatcher implements IPendingRequestRegistry {
// Make controllers publicly accessible
readonly systemController: SystemController;
// readonly permissionController: PermissionController;
readonly permissionController: PermissionController;
// readonly mcpController: MCPController;
// readonly hookController: HookController;
@@ -83,11 +83,11 @@ export class ControlDispatcher implements IPendingRequestRegistry {
this,
'SystemController',
);
// this.permissionController = new PermissionController(
// context,
// this,
// 'PermissionController',
// );
this.permissionController = new PermissionController(
context,
this,
'PermissionController',
);
// this.mcpController = new MCPController(context, this, 'MCPController');
// this.hookController = new HookController(context, this, 'HookController');
@@ -230,7 +230,7 @@ export class ControlDispatcher implements IPendingRequestRegistry {
// Cleanup controllers (MCP controller will close all clients)
this.systemController.cleanup();
// this.permissionController.cleanup();
this.permissionController.cleanup();
// this.mcpController.cleanup();
// this.hookController.cleanup();
}
@@ -302,9 +302,9 @@ export class ControlDispatcher implements IPendingRequestRegistry {
case 'supported_commands':
return this.systemController;
// case 'can_use_tool':
// case 'set_permission_mode':
// return this.permissionController;
case 'can_use_tool':
case 'set_permission_mode':
return this.permissionController;
// case 'mcp_message':
// case 'mcp_server_status':

64
packages/cli/src/nonInteractive/control/ControlService.ts
View File

@@ -29,7 +29,7 @@
import type { IControlContext } from './ControlContext.js';
import type { ControlDispatcher } from './ControlDispatcher.js';
import type {
// PermissionServiceAPI,
PermissionServiceAPI,
SystemServiceAPI,
// McpServiceAPI,
// HookServiceAPI,
@@ -61,43 +61,31 @@ export class ControlService {
* Handles tool execution permissions, approval checks, and callbacks.
* Delegates to the shared PermissionController instance.
*/
// get permission(): PermissionServiceAPI {
// const controller = this.dispatcher.permissionController;
// return {
// /**
// * Check if a tool should be allowed based on current permission settings
// *
// * Evaluates permission mode and tool registry to determine if execution
// * should proceed. Can optionally modify tool arguments based on confirmation details.
// *
// * @param toolRequest - Tool call request information
// * @param confirmationDetails - Optional confirmation details for UI
// * @returns Permission decision with optional updated arguments
// */
// shouldAllowTool: controller.shouldAllowTool.bind(controller),
//
// /**
// * Build UI suggestions for tool confirmation dialogs
// *
// * Creates actionable permission suggestions based on tool confirmation details.
// *
// * @param confirmationDetails - Tool confirmation details
// * @returns Array of permission suggestions or null
// */
// buildPermissionSuggestions:
// controller.buildPermissionSuggestions.bind(controller),
//
// /**
// * Get callback for monitoring tool call status updates
// *
// * Returns callback function for integration with CoreToolScheduler.
// *
// * @returns Callback function for tool call updates
// */
// getToolCallUpdateCallback:
// controller.getToolCallUpdateCallback.bind(controller),
// };
// }
get permission(): PermissionServiceAPI {
const controller = this.dispatcher.permissionController;
return {
/**
* Build UI suggestions for tool confirmation dialogs
*
* Creates actionable permission suggestions based on tool confirmation details.
*
* @param confirmationDetails - Tool confirmation details
* @returns Array of permission suggestions or null
*/
buildPermissionSuggestions:
controller.buildPermissionSuggestions.bind(controller),
/**
* Get callback for monitoring tool call status updates
*
* Returns callback function for integration with CoreToolScheduler.
*
* @returns Callback function for tool call updates
*/
getToolCallUpdateCallback:
controller.getToolCallUpdateCallback.bind(controller),
};
}
/**
* System Domain API

4
packages/cli/src/nonInteractive/control/controllers/baseController.ts
View File

@@ -174,7 +174,5 @@ export abstract class BaseController {
/**
* Cleanup resources
*/
cleanup(): void {
// Subclasses can override to add cleanup logic
}
cleanup(): void {}
}

92
packages/cli/src/nonInteractive/control/controllers/permissionController.ts
View File

@@ -15,8 +15,10 @@
*/
import type {
ToolCallRequestInfo,
WaitingToolCall,
ToolExecuteConfirmationDetails,
ToolMcpConfirmationDetails,
ApprovalMode,
} from '@qwen-code/qwen-code-core';
import {
InputFormat,
@@ -206,6 +208,7 @@ export class PermissionController extends BaseController {
}
this.context.permissionMode = mode;
this.context.config.setApprovalMode(mode as ApprovalMode);
if (this.context.debugMode) {
console.error(
@@ -334,47 +337,6 @@ export class PermissionController extends BaseController {
}
}
/**
* Check if a tool should be executed based on current permission settings
*
* This is a convenience method for direct tool execution checks without
* going through the control request flow.
*/
async shouldAllowTool(
toolRequest: ToolCallRequestInfo,
confirmationDetails?: unknown,
): Promise<{
allowed: boolean;
message?: string;
updatedArgs?: Record<string, unknown>;
}> {
// Check permission mode
const modeResult = this.checkPermissionMode();
if (!modeResult.allowed) {
return {
allowed: false,
message: modeResult.message,
};
}
// Check tool registry
const registryResult = this.checkToolRegistry(toolRequest.name);
if (!registryResult.allowed) {
return {
allowed: false,
message: registryResult.message,
};
}
// If we have confirmation details, we could potentially modify args
// This is a hook for future enhancement
if (confirmationDetails) {
// Future: handle argument modifications based on confirmation details
}
return { allowed: true };
}
/**
* Get callback for monitoring tool calls and handling outgoing permission requests
* This is passed to executeToolCall to hook into CoreToolScheduler updates
@@ -430,17 +392,14 @@ export class PermissionController extends BaseController {
toolCall.confirmationDetails,
);
const response = await this.sendControlRequest(
{
subtype: 'can_use_tool',
tool_name: toolCall.request.name,
tool_use_id: toolCall.request.callId,
input: toolCall.request.args,
permission_suggestions: permissionSuggestions,
blocked_path: null,
} as CLIControlPermissionRequest,
30000,
);
const response = await this.sendControlRequest({
subtype: 'can_use_tool',
tool_name: toolCall.request.name,
tool_use_id: toolCall.request.callId,
input: toolCall.request.args,
permission_suggestions: permissionSuggestions,
blocked_path: null,
} as CLIControlPermissionRequest);
if (response.subtype !== 'success') {
await toolCall.confirmationDetails.onConfirm(
@@ -462,8 +421,15 @@ export class PermissionController extends BaseController {
ToolConfirmationOutcome.ProceedOnce,
);
} else {
// Extract cancel message from response if available
const cancelMessage =
typeof payload['message'] === 'string'
? payload['message']
: undefined;
await toolCall.confirmationDetails.onConfirm(
ToolConfirmationOutcome.Cancel,
cancelMessage ? { cancelMessage } : undefined,
);
}
} catch (error) {
@@ -473,9 +439,23 @@ export class PermissionController extends BaseController {
error,
);
}
await toolCall.confirmationDetails.onConfirm(
ToolConfirmationOutcome.Cancel,
);
// On error, use default cancel message
// Only pass payload for exec and mcp types that support it
const confirmationType = toolCall.confirmationDetails.type;
if (confirmationType === 'exec' || confirmationType === 'mcp') {
const execOrMcpDetails = toolCall.confirmationDetails as
| ToolExecuteConfirmationDetails
| ToolMcpConfirmationDetails;
await execOrMcpDetails.onConfirm(
ToolConfirmationOutcome.Cancel,
undefined,
);
} else {
// For other types, don't pass payload (backward compatible)
await toolCall.confirmationDetails.onConfirm(
ToolConfirmationOutcome.Cancel,
);
}
} finally {
this.pendingOutgoingRequests.delete(toolCall.request.callId);
}

62
packages/cli/src/nonInteractive/control/controllers/systemController.ts
View File

@@ -55,12 +55,68 @@ export class SystemController extends BaseController {
payload: CLIControlInitializeRequest,
): Promise<Record<string, unknown>> {
// Register SDK MCP servers if provided
if (payload.sdkMcpServers && Array.isArray(payload.sdkMcpServers)) {
for (const serverName of payload.sdkMcpServers) {
if (payload.sdkMcpServers && typeof payload.sdkMcpServers === 'object') {
for (const serverName of Object.keys(payload.sdkMcpServers)) {
this.context.sdkMcpServers.add(serverName);
}
// Add SDK MCP servers to config
try {
this.context.config.addMcpServers(payload.sdkMcpServers);
if (this.context.debugMode) {
console.error(
`[SystemController] Added ${Object.keys(payload.sdkMcpServers).length} SDK MCP servers to config`,
);
}
} catch (error) {
if (this.context.debugMode) {
console.error(
'[SystemController] Failed to add SDK MCP servers:',
error,
);
}
}
}
// Add MCP servers to config if provided
if (payload.mcpServers && typeof payload.mcpServers === 'object') {
try {
this.context.config.addMcpServers(payload.mcpServers);
if (this.context.debugMode) {
console.error(
`[SystemController] Added ${Object.keys(payload.mcpServers).length} MCP servers to config`,
);
}
} catch (error) {
if (this.context.debugMode) {
console.error('[SystemController] Failed to add MCP servers:', error);
}
}
}
// Add session subagents to config if provided
if (payload.agents && Array.isArray(payload.agents)) {
try {
this.context.config.addSessionSubagents(payload.agents);
if (this.context.debugMode) {
console.error(
`[SystemController] Added ${payload.agents.length} session subagents to config`,
);
}
} catch (error) {
if (this.context.debugMode) {
console.error(
'[SystemController] Failed to add session subagents:',
error,
);
}
}
}
// Set SDK mode to true after handling initialize
this.context.config.setSdkMode(true);
// Build capabilities for response
const capabilities = this.buildControlCapabilities();
@@ -86,7 +142,7 @@ export class SystemController extends BaseController {
buildControlCapabilities(): Record<string, unknown> {
const capabilities: Record<string, unknown> = {
can_handle_can_use_tool: true,
can_handle_hook_callback: true,
can_handle_hook_callback: false,
can_set_permission_mode:
typeof this.context.config.setApprovalMode === 'function',
can_set_model: typeof this.context.config.setModel === 'function',

24
packages/cli/src/nonInteractive/control/types/serviceAPIs.ts
View File

@@ -13,10 +13,7 @@
*/
import type { Client } from '@modelcontextprotocol/sdk/client/index.js';
import type {
ToolCallRequestInfo,
MCPServerConfig,
} from '@qwen-code/qwen-code-core';
import type { MCPServerConfig } from '@qwen-code/qwen-code-core';
import type { PermissionSuggestion } from '../../types.js';
/**
@@ -26,25 +23,6 @@ import type { PermissionSuggestion } from '../../types.js';
* permission suggestions, and tool call monitoring callbacks.
*/
export interface PermissionServiceAPI {
/**
* Check if a tool should be allowed based on current permission settings
*
* Evaluates permission mode and tool registry to determine if execution
* should proceed. Can optionally modify tool arguments based on confirmation details.
*
* @param toolRequest - Tool call request information containing name, args, and call ID
* @param confirmationDetails - Optional confirmation details for UI-driven approvals
* @returns Promise resolving to permission decision with optional updated arguments
*/
shouldAllowTool(
toolRequest: ToolCallRequestInfo,
confirmationDetails?: unknown,
): Promise<{
allowed: boolean;
message?: string;
updatedArgs?: Record<string, unknown>;
}>;
/**
* Build UI suggestions for tool confirmation dialogs
*

61
packages/cli/src/nonInteractive/io/BaseJsonOutputAdapter.ts
View File

@@ -939,9 +939,25 @@ export abstract class BaseJsonOutputAdapter {
this.emitMessageImpl(message);
}
/**
* Checks if responseParts contain any functionResponse with an error.
* This handles cancelled responses and other error cases where the error
* is embedded in responseParts rather than the top-level error field.
* @param responseParts - Array of Part objects
* @returns Error message if found, undefined otherwise
*/
private checkResponsePartsForError(
responseParts: Part[] | undefined,
): string | undefined {
// Use the shared helper function defined at file level
return checkResponsePartsForError(responseParts);
}
/**
* Emits a tool result message.
* Collects execution denied tool calls for inclusion in result messages.
* Handles both explicit errors (response.error) and errors embedded in
* responseParts (e.g., cancelled responses).
* @param request - Tool call request info
* @param response - Tool call response info
* @param parentToolUseId - Parent tool use ID (null for main agent)
@@ -951,6 +967,14 @@ export abstract class BaseJsonOutputAdapter {
response: ToolCallResponseInfo,
parentToolUseId: string | null = null,
): void {
// Check for errors in responseParts (e.g., cancelled responses)
const responsePartsError = this.checkResponsePartsForError(
response.responseParts,
);
// Determine if this is an error response
const hasError = Boolean(response.error) || Boolean(responsePartsError);
// Track permission denials (execution denied errors)
if (
response.error &&
@@ -967,7 +991,7 @@ export abstract class BaseJsonOutputAdapter {
const block: ToolResultBlock = {
type: 'tool_result',
tool_use_id: request.callId,
is_error: Boolean(response.error),
is_error: hasError,
};
const content = toolResultContent(response);
if (content !== undefined) {
@@ -1173,11 +1197,41 @@ export function partsToString(parts: Part[]): string {
.join('');
}
/**
* Checks if responseParts contain any functionResponse with an error.
* Helper function for extracting error messages from responseParts.
* @param responseParts - Array of Part objects
* @returns Error message if found, undefined otherwise
*/
function checkResponsePartsForError(
responseParts: Part[] | undefined,
): string | undefined {
if (!responseParts || responseParts.length === 0) {
return undefined;
}
for (const part of responseParts) {
if (
'functionResponse' in part &&
part.functionResponse?.response &&
typeof part.functionResponse.response === 'object' &&
'error' in part.functionResponse.response &&
part.functionResponse.response['error']
) {
const error = part.functionResponse.response['error'];
return typeof error === 'string' ? error : String(error);
}
}
return undefined;
}
/**
* Extracts content from tool response.
* Uses functionResponsePartsToString to properly handle functionResponse parts,
* which correctly extracts output content from functionResponse objects rather
* than simply concatenating text or JSON.stringify.
* Also handles errors embedded in responseParts (e.g., cancelled responses).
*
* @param response - Tool call response
* @returns String content or undefined
@@ -1188,6 +1242,11 @@ export function toolResultContent(
if (response.error) {
return response.error.message;
}
// Check for errors in responseParts (e.g., cancelled responses)
const responsePartsError = checkResponsePartsForError(response.responseParts);
if (responsePartsError) {
return responsePartsError;
}
if (
typeof response.resultDisplay === 'string' &&
response.resultDisplay.trim().length > 0

2
packages/cli/src/nonInteractive/session.test.ts
View File

@@ -134,7 +134,7 @@ function createControlCancel(requestId: string): ControlCancelRequest {
};
}
describe('runNonInteractiveStreamJson', () => {
describe('runNonInteractiveStreamJson (refactored)', () => {
let config: Config;
let mockInputReader: {
read: () => AsyncGenerator<

656
packages/cli/src/nonInteractive/session.ts
View File

@@ -4,17 +4,6 @@
* SPDX-License-Identifier: Apache-2.0
*/
/**
* Stream JSON Runner with Session State Machine
*
* Handles stream-json input/output format with:
* - Initialize handshake
* - Message routing (control vs user messages)
* - FIFO user message queue
* - Sequential message processing
* - Graceful shutdown
*/
import type { Config } from '@qwen-code/qwen-code-core';
import { StreamJsonInputReader } from './io/StreamJsonInputReader.js';
import { StreamJsonOutputAdapter } from './io/StreamJsonOutputAdapter.js';
@@ -42,48 +31,7 @@ import { createMinimalSettings } from '../config/settings.js';
import { runNonInteractive } from '../nonInteractiveCli.js';
import { ConsolePatcher } from '../ui/utils/ConsolePatcher.js';
const SESSION_STATE = {
INITIALIZING: 'initializing',
IDLE: 'idle',
PROCESSING_QUERY: 'processing_query',
SHUTTING_DOWN: 'shutting_down',
} as const;
type SessionState = (typeof SESSION_STATE)[keyof typeof SESSION_STATE];
/**
* Message type classification for routing
*/
type MessageType =
| 'control_request'
| 'control_response'
| 'control_cancel'
| 'user'
| 'assistant'
| 'system'
| 'result'
| 'stream_event'
| 'unknown';
/**
* Routed message with classification
*/
interface RoutedMessage {
type: MessageType;
message:
| CLIMessage
| CLIControlRequest
| CLIControlResponse
| ControlCancelRequest;
}
/**
* Session Manager
*
* Manages the session lifecycle and message processing state machine.
*/
class SessionManager {
private state: SessionState = SESSION_STATE.INITIALIZING;
class Session {
private userMessageQueue: CLIUserMessage[] = [];
private abortController: AbortController;
private config: Config;
@@ -98,6 +46,9 @@ class SessionManager {
private debugMode: boolean;
private shutdownHandler: (() => void) | null = null;
private initialPrompt: CLIUserMessage | null = null;
private processingPromise: Promise<void> | null = null;
private isShuttingDown: boolean = false;
private configInitialized: boolean = false;
constructor(config: Config, initialPrompt?: CLIUserMessage) {
this.config = config;
@@ -112,146 +63,31 @@ class SessionManager {
config.getIncludePartialMessages(),
);
// Setup signal handlers for graceful shutdown
this.setupSignalHandlers();
}
/**
* Get next prompt ID
*/
private getNextPromptId(): string {
this.promptIdCounter++;
return `${this.sessionId}########${this.promptIdCounter}`;
}
/**
* Route a message to the appropriate handler based on its type
*
* Classifies incoming messages and routes them to appropriate handlers.
*/
private route(
message:
| CLIMessage
| CLIControlRequest
| CLIControlResponse
| ControlCancelRequest,
): RoutedMessage {
// Check control messages first
if (isControlRequest(message)) {
return { type: 'control_request', message };
}
if (isControlResponse(message)) {
return { type: 'control_response', message };
}
if (isControlCancel(message)) {
return { type: 'control_cancel', message };
private async ensureConfigInitialized(): Promise<void> {
if (this.configInitialized) {
return;
}
// Check data messages
if (isCLIUserMessage(message)) {
return { type: 'user', message };
}
if (isCLIAssistantMessage(message)) {
return { type: 'assistant', message };
}
if (isCLISystemMessage(message)) {
return { type: 'system', message };
}
if (isCLIResultMessage(message)) {
return { type: 'result', message };
}
if (isCLIPartialAssistantMessage(message)) {
return { type: 'stream_event', message };
}
// Unknown message type
if (this.debugMode) {
console.error(
'[SessionManager] Unknown message type:',
JSON.stringify(message, null, 2),
);
}
return { type: 'unknown', message };
}
/**
* Process a single message with unified logic for both initial prompt and stream messages.
*
* Handles:
* - Abort check
* - First message detection and handling
* - Normal message processing
* - Shutdown state checks
*
* @param message - Message to process
* @returns true if the calling code should exit (break/return), false to continue
*/
private async processSingleMessage(
message:
| CLIMessage
| CLIControlRequest
| CLIControlResponse
| ControlCancelRequest,
): Promise<boolean> {
// Check for abort
if (this.abortController.signal.aborted) {
return true;
console.error('[Session] Initializing config');
}
// Handle first message if control system not yet initialized
if (this.controlSystemEnabled === null) {
const handled = await this.handleFirstMessage(message);
if (handled) {
// If handled, check if we should shutdown
return this.state === SESSION_STATE.SHUTTING_DOWN;
}
// If not handled, fall through to normal processing
}
// Process message normally
await this.processMessage(message);
// Check for shutdown after processing
return this.state === SESSION_STATE.SHUTTING_DOWN;
}
/**
* Main entry point - run the session
*/
async run(): Promise<void> {
try {
if (this.debugMode) {
console.error('[SessionManager] Starting session', this.sessionId);
}
// Process initial prompt if provided
if (this.initialPrompt !== null) {
const shouldExit = await this.processSingleMessage(this.initialPrompt);
if (shouldExit) {
await this.shutdown();
return;
}
}
// Process messages from stream
for await (const message of this.inputReader.read()) {
const shouldExit = await this.processSingleMessage(message);
if (shouldExit) {
break;
}
}
// Stream closed, shutdown
await this.shutdown();
await this.config.initialize();
this.configInitialized = true;
} catch (error) {
if (this.debugMode) {
console.error('[SessionManager] Error:', error);
console.error('[Session] Failed to initialize config:', error);
}
await this.shutdown();
throw error;
} finally {
// Ensure signal handlers are always cleaned up even if shutdown wasn't called
this.cleanupSignalHandlers();
}
}
@@ -259,14 +95,6 @@ class SessionManager {
if (this.controlContext && this.dispatcher && this.controlService) {
return;
}
// The control system follows a strict three-layer architecture:
// 1. ControlContext (shared session state)
// 2. ControlDispatcher (protocol routing SDK ↔ CLI)
// 3. ControlService (programmatic API for CLI runtime)
//
// Application code MUST interact with the control plane exclusively through
// ControlService. ControlDispatcher is reserved for protocol-level message
// routing and should never be used directly outside of this file.
this.controlContext = new ControlContext({
config: this.config,
streamJson: this.outputAdapter,
@@ -299,25 +127,32 @@ class SessionManager {
| CLIControlResponse
| ControlCancelRequest,
): Promise<boolean> {
const routed = this.route(message);
if (routed.type === 'control_request') {
const request = routed.message as CLIControlRequest;
if (isControlRequest(message)) {
const request = message as CLIControlRequest;
this.controlSystemEnabled = true;
this.ensureControlSystem();
if (request.request.subtype === 'initialize') {
// Dispatch the initialize request first
await this.dispatcher?.dispatch(request);
this.state = SESSION_STATE.IDLE;
// After handling initialize control request, initialize the config
// This is the SDK mode where config initialization is deferred
await this.ensureConfigInitialized();
return true;
}
return false;
if (this.debugMode) {
console.error(
'[Session] Ignoring non-initialize control request during initialization',
);
}
return true;
}
if (routed.type === 'user') {
if (isCLIUserMessage(message)) {
this.controlSystemEnabled = false;
this.state = SESSION_STATE.PROCESSING_QUERY;
this.userMessageQueue.push(routed.message as CLIUserMessage);
await this.processUserMessageQueue();
// For non-SDK mode (direct user message), initialize config if not already done
await this.ensureConfigInitialized();
this.enqueueUserMessage(message as CLIUserMessage);
return true;
}
@@ -325,241 +160,50 @@ class SessionManager {
return false;
}
/**
* Process a single message from the stream
*/
private async processMessage(
message:
| CLIMessage
| CLIControlRequest
| CLIControlResponse
| ControlCancelRequest,
private async handleControlRequest(
request: CLIControlRequest,
): Promise<void> {
const routed = this.route(message);
if (this.debugMode) {
console.error(
`[SessionManager] State: ${this.state}, Message type: ${routed.type}`,
);
}
switch (this.state) {
case SESSION_STATE.INITIALIZING:
await this.handleInitializingState(routed);
break;
case SESSION_STATE.IDLE:
await this.handleIdleState(routed);
break;
case SESSION_STATE.PROCESSING_QUERY:
await this.handleProcessingState(routed);
break;
case SESSION_STATE.SHUTTING_DOWN:
// Ignore all messages during shutdown
break;
default: {
// Exhaustive check
const _exhaustiveCheck: never = this.state;
if (this.debugMode) {
console.error('[SessionManager] Unknown state:', _exhaustiveCheck);
}
break;
}
}
}
/**
* Handle messages in initializing state
*/
private async handleInitializingState(routed: RoutedMessage): Promise<void> {
if (routed.type === 'control_request') {
const request = routed.message as CLIControlRequest;
const dispatcher = this.getDispatcher();
if (!dispatcher) {
if (this.debugMode) {
console.error(
'[SessionManager] Control request received before control system initialization',
);
}
return;
}
if (request.request.subtype === 'initialize') {
await dispatcher.dispatch(request);
this.state = SESSION_STATE.IDLE;
if (this.debugMode) {
console.error('[SessionManager] Initialized, transitioning to idle');
}
} else {
if (this.debugMode) {
console.error(
'[SessionManager] Ignoring non-initialize control request during initialization',
);
}
}
} else {
if (this.debugMode) {
console.error(
'[SessionManager] Ignoring non-control message during initialization',
);
}
}
}
/**
* Handle messages in idle state
*/
private async handleIdleState(routed: RoutedMessage): Promise<void> {
const dispatcher = this.getDispatcher();
if (routed.type === 'control_request') {
if (!dispatcher) {
if (this.debugMode) {
console.error('[SessionManager] Ignoring control request (disabled)');
}
return;
}
const request = routed.message as CLIControlRequest;
await dispatcher.dispatch(request);
// Stay in idle state
} else if (routed.type === 'control_response') {
if (!dispatcher) {
return;
}
const response = routed.message as CLIControlResponse;
dispatcher.handleControlResponse(response);
// Stay in idle state
} else if (routed.type === 'control_cancel') {
if (!dispatcher) {
return;
}
const cancelRequest = routed.message as ControlCancelRequest;
dispatcher.handleCancel(cancelRequest.request_id);
} else if (routed.type === 'user') {
const userMessage = routed.message as CLIUserMessage;
this.userMessageQueue.push(userMessage);
// Start processing queue
await this.processUserMessageQueue();
} else {
if (!dispatcher) {
if (this.debugMode) {
console.error(
'[SessionManager] Ignoring message type in idle state:',
routed.type,
);
console.error('[Session] Control system not enabled');
}
}
}
/**
* Handle messages in processing state
*/
private async handleProcessingState(routed: RoutedMessage): Promise<void> {
const dispatcher = this.getDispatcher();
if (routed.type === 'control_request') {
if (!dispatcher) {
if (this.debugMode) {
console.error(
'[SessionManager] Control request ignored during processing (disabled)',
);
}
return;
}
const request = routed.message as CLIControlRequest;
await dispatcher.dispatch(request);
// Continue processing
} else if (routed.type === 'control_response') {
if (!dispatcher) {
return;
}
const response = routed.message as CLIControlResponse;
dispatcher.handleControlResponse(response);
// Continue processing
} else if (routed.type === 'user') {
// Enqueue for later
const userMessage = routed.message as CLIUserMessage;
this.userMessageQueue.push(userMessage);
if (this.debugMode) {
console.error(
'[SessionManager] Enqueued user message during processing',
);
}
} else {
if (this.debugMode) {
console.error(
'[SessionManager] Ignoring message type during processing:',
routed.type,
);
}
}
}
/**
* Process user message queue (FIFO)
*/
private async processUserMessageQueue(): Promise<void> {
while (
this.userMessageQueue.length > 0 &&
!this.abortController.signal.aborted
) {
this.state = SESSION_STATE.PROCESSING_QUERY;
const userMessage = this.userMessageQueue.shift()!;
try {
await this.processUserMessage(userMessage);
} catch (error) {
if (this.debugMode) {
console.error(
'[SessionManager] Error processing user message:',
error,
);
}
// Send error result
this.emitErrorResult(error);
}
}
// If control system is disabled (single-query mode) and queue is empty,
// automatically shutdown instead of returning to idle
if (
!this.abortController.signal.aborted &&
this.state === SESSION_STATE.PROCESSING_QUERY &&
this.controlSystemEnabled === false &&
this.userMessageQueue.length === 0
) {
if (this.debugMode) {
console.error(
'[SessionManager] Single-query mode: queue processed, shutting down',
);
}
this.state = SESSION_STATE.SHUTTING_DOWN;
return;
}
// Return to idle after processing queue (for multi-query mode with control system)
if (
!this.abortController.signal.aborted &&
this.state === SESSION_STATE.PROCESSING_QUERY
) {
this.state = SESSION_STATE.IDLE;
if (this.debugMode) {
console.error('[SessionManager] Queue processed, returning to idle');
}
}
await dispatcher.dispatch(request);
}
private handleControlResponse(response: CLIControlResponse): void {
const dispatcher = this.getDispatcher();
if (!dispatcher) {
return;
}
dispatcher.handleControlResponse(response);
}
private handleControlCancel(cancelRequest: ControlCancelRequest): void {
const dispatcher = this.getDispatcher();
if (!dispatcher) {
return;
}
dispatcher.handleCancel(cancelRequest.request_id);
}
/**
* Process a single user message
*/
private async processUserMessage(userMessage: CLIUserMessage): Promise<void> {
const input = extractUserMessageText(userMessage);
if (!input) {
if (this.debugMode) {
console.error('[SessionManager] No text content in user message');
console.error('[Session] No text content in user message');
}
return;
}
// Ensure config is initialized before processing user messages
await this.ensureConfigInitialized();
const promptId = this.getNextPromptId();
try {
@@ -575,16 +219,56 @@ class SessionManager {
},
);
} catch (error) {
// Error already handled by runNonInteractive via adapter.emitResult
if (this.debugMode) {
console.error('[SessionManager] Query execution error:', error);
console.error('[Session] Query execution error:', error);
}
}
}
/**
* Send tool results as user message
*/
private async processUserMessageQueue(): Promise<void> {
if (this.isShuttingDown || this.abortController.signal.aborted) {
return;
}
while (
this.userMessageQueue.length > 0 &&
!this.isShuttingDown &&
!this.abortController.signal.aborted
) {
const userMessage = this.userMessageQueue.shift()!;
try {
await this.processUserMessage(userMessage);
} catch (error) {
if (this.debugMode) {
console.error('[Session] Error processing user message:', error);
}
this.emitErrorResult(error);
}
}
}
private enqueueUserMessage(userMessage: CLIUserMessage): void {
this.userMessageQueue.push(userMessage);
this.ensureProcessingStarted();
}
private ensureProcessingStarted(): void {
if (this.processingPromise) {
return;
}
this.processingPromise = this.processUserMessageQueue().finally(() => {
this.processingPromise = null;
if (
this.userMessageQueue.length > 0 &&
!this.isShuttingDown &&
!this.abortController.signal.aborted
) {
this.ensureProcessingStarted();
}
});
}
private emitErrorResult(
error: unknown,
numTurns: number = 0,
@@ -602,52 +286,51 @@ class SessionManager {
});
}
/**
* Handle interrupt control request
*/
private handleInterrupt(): void {
if (this.debugMode) {
console.error('[SessionManager] Interrupt requested');
}
// Abort current query if processing
if (this.state === SESSION_STATE.PROCESSING_QUERY) {
this.abortController.abort();
this.abortController = new AbortController(); // Create new controller for next query
console.error('[Session] Interrupt requested');
}
this.abortController.abort();
this.abortController = new AbortController();
}
/**
* Setup signal handlers for graceful shutdown
*/
private setupSignalHandlers(): void {
this.shutdownHandler = () => {
if (this.debugMode) {
console.error('[SessionManager] Shutdown signal received');
console.error('[Session] Shutdown signal received');
}
this.isShuttingDown = true;
this.abortController.abort();
this.state = SESSION_STATE.SHUTTING_DOWN;
};
process.on('SIGINT', this.shutdownHandler);
process.on('SIGTERM', this.shutdownHandler);
}
/**
* Shutdown session and cleanup resources
*/
private async shutdown(): Promise<void> {
if (this.debugMode) {
console.error('[SessionManager] Shutting down');
console.error('[Session] Shutting down');
}
this.isShuttingDown = true;
if (this.processingPromise) {
try {
await this.processingPromise;
} catch (error) {
if (this.debugMode) {
console.error(
'[Session] Error waiting for processing to complete:',
error,
);
}
}
}
this.state = SESSION_STATE.SHUTTING_DOWN;
this.dispatcher?.shutdown();
this.cleanupSignalHandlers();
}
/**
* Remove signal handlers to prevent memory leaks
*/
private cleanupSignalHandlers(): void {
if (this.shutdownHandler) {
process.removeListener('SIGINT', this.shutdownHandler);
@@ -655,6 +338,94 @@ class SessionManager {
this.shutdownHandler = null;
}
}
async run(): Promise<void> {
try {
if (this.debugMode) {
console.error('[Session] Starting session', this.sessionId);
}
if (this.initialPrompt !== null) {
const handled = await this.handleFirstMessage(this.initialPrompt);
if (handled && this.isShuttingDown) {
await this.shutdown();
return;
}
}
try {
for await (const message of this.inputReader.read()) {
if (this.abortController.signal.aborted) {
break;
}
if (this.controlSystemEnabled === null) {
const handled = await this.handleFirstMessage(message);
if (handled) {
if (this.isShuttingDown) {
break;
}
continue;
}
}
if (isControlRequest(message)) {
await this.handleControlRequest(message as CLIControlRequest);
} else if (isControlResponse(message)) {
this.handleControlResponse(message as CLIControlResponse);
} else if (isControlCancel(message)) {
this.handleControlCancel(message as ControlCancelRequest);
} else if (isCLIUserMessage(message)) {
this.enqueueUserMessage(message as CLIUserMessage);
} else if (this.debugMode) {
if (
!isCLIAssistantMessage(message) &&
!isCLISystemMessage(message) &&
!isCLIResultMessage(message) &&
!isCLIPartialAssistantMessage(message)
) {
console.error(
'[Session] Unknown message type:',
JSON.stringify(message, null, 2),
);
}
}
if (this.isShuttingDown) {
break;
}
}
} catch (streamError) {
if (this.debugMode) {
console.error('[Session] Stream reading error:', streamError);
}
throw streamError;
}
while (this.processingPromise) {
if (this.debugMode) {
console.error('[Session] Waiting for final processing to complete');
}
try {
await this.processingPromise;
} catch (error) {
if (this.debugMode) {
console.error('[Session] Error in final processing:', error);
}
}
}
await this.shutdown();
} catch (error) {
if (this.debugMode) {
console.error('[Session] Error:', error);
}
await this.shutdown();
throw error;
} finally {
this.cleanupSignalHandlers();
}
}
}
function extractUserMessageText(message: CLIUserMessage): string | null {
@@ -682,12 +453,6 @@ function extractUserMessageText(message: CLIUserMessage): string | null {
return null;
}
/**
* Entry point for stream-json mode
*
* @param config - Configuration object
* @param input - Optional initial prompt input to process before reading from stream
*/
export async function runNonInteractiveStreamJson(
config: Config,
input: string,
@@ -698,7 +463,6 @@ export async function runNonInteractiveStreamJson(
consolePatcher.patch();
try {
// Create initial user message from prompt input if provided
let initialPrompt: CLIUserMessage | undefined = undefined;
if (input && input.trim().length > 0) {
const sessionId = config.getSessionId();
@@ -713,7 +477,7 @@ export async function runNonInteractiveStreamJson(
};
}
const manager = new SessionManager(config, initialPrompt);
const manager = new Session(config, initialPrompt);
await manager.run();
} finally {
consolePatcher.cleanup();

14
packages/cli/src/nonInteractive/types.ts
View File

@@ -1,4 +1,8 @@
/* eslint-disable @typescript-eslint/no-explicit-any */
import type {
MCPServerConfig,
SubagentConfig,
} from '@qwen-code/qwen-code-core';
/**
* Annotation for attaching metadata to content blocks
@@ -295,10 +299,18 @@ export interface CLIControlPermissionRequest {
blocked_path: string | null;
}
export enum AuthProviderType {
DYNAMIC_DISCOVERY = 'dynamic_discovery',
GOOGLE_CREDENTIALS = 'google_credentials',
SERVICE_ACCOUNT_IMPERSONATION = 'service_account_impersonation',
}
export interface CLIControlInitializeRequest {
subtype: 'initialize';
hooks?: HookRegistration[] | null;
sdkMcpServers?: string[];
sdkMcpServers?: Record<string, MCPServerConfig>;
mcpServers?: Record<string, MCPServerConfig>;
agents?: SubagentConfig[];
}
export interface CLIControlSetPermissionModeRequest {