forked from
oppi.li/claude-code
source dump of claude code
1// biome-ignore-all assist/source/organizeImports: ANT-ONLY import markers must not be reordered
2import { feature } from 'bun:bundle'
3import { readFile, stat } from 'fs/promises'
4import { dirname } from 'path'
5import {
6 downloadUserSettings,
7 redownloadUserSettings,
8} from 'src/services/settingsSync/index.js'
9import { waitForRemoteManagedSettingsToLoad } from 'src/services/remoteManagedSettings/index.js'
10import { StructuredIO } from 'src/cli/structuredIO.js'
11import { RemoteIO } from 'src/cli/remoteIO.js'
12import {
13 type Command,
14 formatDescriptionWithSource,
15 getCommandName,
16} from 'src/commands.js'
17import { createStreamlinedTransformer } from 'src/utils/streamlinedTransform.js'
18import { installStreamJsonStdoutGuard } from 'src/utils/streamJsonStdoutGuard.js'
19import type { ToolPermissionContext } from 'src/Tool.js'
20import type { ThinkingConfig } from 'src/utils/thinking.js'
21import { assembleToolPool, filterToolsByDenyRules } from 'src/tools.js'
22import uniqBy from 'lodash-es/uniqBy.js'
23import { uniq } from 'src/utils/array.js'
24import { mergeAndFilterTools } from 'src/utils/toolPool.js'
25import {
26 logEvent,
27 type AnalyticsMetadata_I_VERIFIED_THIS_IS_NOT_CODE_OR_FILEPATHS,
28} from 'src/services/analytics/index.js'
29import { getFeatureValue_CACHED_MAY_BE_STALE } from 'src/services/analytics/growthbook.js'
30import { logForDebugging } from 'src/utils/debug.js'
31import {
32 logForDiagnosticsNoPII,
33 withDiagnosticsTiming,
34} from 'src/utils/diagLogs.js'
35import { toolMatchesName, type Tool, type Tools } from 'src/Tool.js'
36import {
37 type AgentDefinition,
38 isBuiltInAgent,
39 parseAgentsFromJson,
40} from 'src/tools/AgentTool/loadAgentsDir.js'
41import type { Message, NormalizedUserMessage } from 'src/types/message.js'
42import type { QueuedCommand } from 'src/types/textInputTypes.js'
43import {
44 dequeue,
45 dequeueAllMatching,
46 enqueue,
47 hasCommandsInQueue,
48 peek,
49 subscribeToCommandQueue,
50 getCommandsByMaxPriority,
51} from 'src/utils/messageQueueManager.js'
52import { notifyCommandLifecycle } from 'src/utils/commandLifecycle.js'
53import {
54 getSessionState,
55 notifySessionStateChanged,
56 notifySessionMetadataChanged,
57 setPermissionModeChangedListener,
58 type RequiresActionDetails,
59 type SessionExternalMetadata,
60} from 'src/utils/sessionState.js'
61import { externalMetadataToAppState } from 'src/state/onChangeAppState.js'
62import { getInMemoryErrors, logError, logMCPDebug } from 'src/utils/log.js'
63import {
64 writeToStdout,
65 registerProcessOutputErrorHandlers,
66} from 'src/utils/process.js'
67import type { Stream } from 'src/utils/stream.js'
68import { EMPTY_USAGE } from 'src/services/api/logging.js'
69import {
70 loadConversationForResume,
71 type TurnInterruptionState,
72} from 'src/utils/conversationRecovery.js'
73import type {
74 MCPServerConnection,
75 McpSdkServerConfig,
76 ScopedMcpServerConfig,
77} from 'src/services/mcp/types.js'
78import {
79 ChannelMessageNotificationSchema,
80 gateChannelServer,
81 wrapChannelMessage,
82 findChannelEntry,
83} from 'src/services/mcp/channelNotification.js'
84import {
85 isChannelAllowlisted,
86 isChannelsEnabled,
87} from 'src/services/mcp/channelAllowlist.js'
88import { parsePluginIdentifier } from 'src/utils/plugins/pluginIdentifier.js'
89import { validateUuid } from 'src/utils/uuid.js'
90import { fromArray } from 'src/utils/generators.js'
91import { ask } from 'src/QueryEngine.js'
92import type { PermissionPromptTool } from 'src/utils/queryHelpers.js'
93import {
94 createFileStateCacheWithSizeLimit,
95 mergeFileStateCaches,
96 READ_FILE_STATE_CACHE_SIZE,
97} from 'src/utils/fileStateCache.js'
98import { expandPath } from 'src/utils/path.js'
99import { extractReadFilesFromMessages } from 'src/utils/queryHelpers.js'
100import { registerHookEventHandler } from 'src/utils/hooks/hookEvents.js'
101import { executeFilePersistence } from 'src/utils/filePersistence/filePersistence.js'
102import { finalizePendingAsyncHooks } from 'src/utils/hooks/AsyncHookRegistry.js'
103import {
104 gracefulShutdown,
105 gracefulShutdownSync,
106 isShuttingDown,
107} from 'src/utils/gracefulShutdown.js'
108import { registerCleanup } from 'src/utils/cleanupRegistry.js'
109import { createIdleTimeoutManager } from 'src/utils/idleTimeout.js'
110import type {
111 SDKStatus,
112 ModelInfo,
113 SDKMessage,
114 SDKUserMessage,
115 SDKUserMessageReplay,
116 PermissionResult,
117 McpServerConfigForProcessTransport,
118 McpServerStatus,
119 RewindFilesResult,
120} from 'src/entrypoints/agentSdkTypes.js'
121import type {
122 StdoutMessage,
123 SDKControlInitializeRequest,
124 SDKControlInitializeResponse,
125 SDKControlRequest,
126 SDKControlResponse,
127 SDKControlMcpSetServersResponse,
128 SDKControlReloadPluginsResponse,
129} from 'src/entrypoints/sdk/controlTypes.js'
130import type { PermissionMode } from '@anthropic-ai/claude-agent-sdk'
131import type { PermissionMode as InternalPermissionMode } from 'src/types/permissions.js'
132import { cwd } from 'process'
133import { getCwd } from 'src/utils/cwd.js'
134import omit from 'lodash-es/omit.js'
135import reject from 'lodash-es/reject.js'
136import { isPolicyAllowed } from 'src/services/policyLimits/index.js'
137import type { ReplBridgeHandle } from 'src/bridge/replBridge.js'
138import { getRemoteSessionUrl } from 'src/constants/product.js'
139import { buildBridgeConnectUrl } from 'src/bridge/bridgeStatusUtil.js'
140import { extractInboundMessageFields } from 'src/bridge/inboundMessages.js'
141import { resolveAndPrepend } from 'src/bridge/inboundAttachments.js'
142import type { CanUseToolFn } from 'src/hooks/useCanUseTool.js'
143import { hasPermissionsToUseTool } from 'src/utils/permissions/permissions.js'
144import { safeParseJSON } from 'src/utils/json.js'
145import {
146 outputSchema as permissionToolOutputSchema,
147 permissionPromptToolResultToPermissionDecision,
148} from 'src/utils/permissions/PermissionPromptToolResultSchema.js'
149import { createAbortController } from 'src/utils/abortController.js'
150import { createCombinedAbortSignal } from 'src/utils/combinedAbortSignal.js'
151import { generateSessionTitle } from 'src/utils/sessionTitle.js'
152import { buildSideQuestionFallbackParams } from 'src/utils/queryContext.js'
153import { runSideQuestion } from 'src/utils/sideQuestion.js'
154import {
155 processSessionStartHooks,
156 processSetupHooks,
157 takeInitialUserMessage,
158} from 'src/utils/sessionStart.js'
159import {
160 DEFAULT_OUTPUT_STYLE_NAME,
161 getAllOutputStyles,
162} from 'src/constants/outputStyles.js'
163import { TEAMMATE_MESSAGE_TAG, TICK_TAG } from 'src/constants/xml.js'
164import {
165 getSettings_DEPRECATED,
166 getSettingsWithSources,
167} from 'src/utils/settings/settings.js'
168import { settingsChangeDetector } from 'src/utils/settings/changeDetector.js'
169import { applySettingsChange } from 'src/utils/settings/applySettingsChange.js'
170import {
171 isFastModeAvailable,
172 isFastModeEnabled,
173 isFastModeSupportedByModel,
174 getFastModeState,
175} from 'src/utils/fastMode.js'
176import {
177 isAutoModeGateEnabled,
178 getAutoModeUnavailableNotification,
179 getAutoModeUnavailableReason,
180 isBypassPermissionsModeDisabled,
181 transitionPermissionMode,
182} from 'src/utils/permissions/permissionSetup.js'
183import {
184 tryGenerateSuggestion,
185 logSuggestionOutcome,
186 logSuggestionSuppressed,
187 type PromptVariant,
188} from 'src/services/PromptSuggestion/promptSuggestion.js'
189import { getLastCacheSafeParams } from 'src/utils/forkedAgent.js'
190import { getAccountInformation } from 'src/utils/auth.js'
191import { OAuthService } from 'src/services/oauth/index.js'
192import { installOAuthTokens } from 'src/cli/handlers/auth.js'
193import { getAPIProvider } from 'src/utils/model/providers.js'
194import type { HookCallbackMatcher } from 'src/types/hooks.js'
195import { AwsAuthStatusManager } from 'src/utils/awsAuthStatusManager.js'
196import type { HookEvent } from 'src/entrypoints/agentSdkTypes.js'
197import {
198 registerHookCallbacks,
199 setInitJsonSchema,
200 getInitJsonSchema,
201 setSdkAgentProgressSummariesEnabled,
202} from 'src/bootstrap/state.js'
203import { createSyntheticOutputTool } from 'src/tools/SyntheticOutputTool/SyntheticOutputTool.js'
204import { parseSessionIdentifier } from 'src/utils/sessionUrl.js'
205import {
206 hydrateRemoteSession,
207 hydrateFromCCRv2InternalEvents,
208 resetSessionFilePointer,
209 doesMessageExistInSession,
210 findUnresolvedToolUse,
211 recordAttributionSnapshot,
212 saveAgentSetting,
213 saveMode,
214 saveAiGeneratedTitle,
215 restoreSessionMetadata,
216} from 'src/utils/sessionStorage.js'
217import { incrementPromptCount } from 'src/utils/commitAttribution.js'
218import {
219 setupSdkMcpClients,
220 connectToServer,
221 clearServerCache,
222 fetchToolsForClient,
223 areMcpConfigsEqual,
224 reconnectMcpServerImpl,
225} from 'src/services/mcp/client.js'
226import {
227 filterMcpServersByPolicy,
228 getMcpConfigByName,
229 isMcpServerDisabled,
230 setMcpServerEnabled,
231} from 'src/services/mcp/config.js'
232import {
233 performMCPOAuthFlow,
234 revokeServerTokens,
235} from 'src/services/mcp/auth.js'
236import {
237 runElicitationHooks,
238 runElicitationResultHooks,
239} from 'src/services/mcp/elicitationHandler.js'
240import { executeNotificationHooks } from 'src/utils/hooks.js'
241import {
242 ElicitRequestSchema,
243 ElicitationCompleteNotificationSchema,
244} from '@modelcontextprotocol/sdk/types.js'
245import { getMcpPrefix } from 'src/services/mcp/mcpStringUtils.js'
246import {
247 commandBelongsToServer,
248 filterToolsByServer,
249} from 'src/services/mcp/utils.js'
250import { setupVscodeSdkMcp } from 'src/services/mcp/vscodeSdkMcp.js'
251import { getAllMcpConfigs } from 'src/services/mcp/config.js'
252import {
253 isQualifiedForGrove,
254 checkGroveForNonInteractive,
255} from 'src/services/api/grove.js'
256import {
257 toInternalMessages,
258 toSDKRateLimitInfo,
259} from 'src/utils/messages/mappers.js'
260import { createModelSwitchBreadcrumbs } from 'src/utils/messages.js'
261import { collectContextData } from 'src/commands/context/context-noninteractive.js'
262import { LOCAL_COMMAND_STDOUT_TAG } from 'src/constants/xml.js'
263import {
264 statusListeners,
265 type ClaudeAILimits,
266} from 'src/services/claudeAiLimits.js'
267import {
268 getDefaultMainLoopModel,
269 getMainLoopModel,
270 modelDisplayString,
271 parseUserSpecifiedModel,
272} from 'src/utils/model/model.js'
273import { getModelOptions } from 'src/utils/model/modelOptions.js'
274import {
275 modelSupportsEffort,
276 modelSupportsMaxEffort,
277 EFFORT_LEVELS,
278 resolveAppliedEffort,
279} from 'src/utils/effort.js'
280import { modelSupportsAdaptiveThinking } from 'src/utils/thinking.js'
281import { modelSupportsAutoMode } from 'src/utils/betas.js'
282import { ensureModelStringsInitialized } from 'src/utils/model/modelStrings.js'
283import {
284 getSessionId,
285 setMainLoopModelOverride,
286 setMainThreadAgentType,
287 switchSession,
288 isSessionPersistenceDisabled,
289 getIsRemoteMode,
290 getFlagSettingsInline,
291 setFlagSettingsInline,
292 getMainThreadAgentType,
293 getAllowedChannels,
294 setAllowedChannels,
295 type ChannelEntry,
296} from 'src/bootstrap/state.js'
297import { runWithWorkload, WORKLOAD_CRON } from 'src/utils/workloadContext.js'
298import type { UUID } from 'crypto'
299import { randomUUID } from 'crypto'
300import type { ContentBlockParam } from '@anthropic-ai/sdk/resources/messages.mjs'
301import type { AppState } from 'src/state/AppStateStore.js'
302import {
303 fileHistoryRewind,
304 fileHistoryCanRestore,
305 fileHistoryEnabled,
306 fileHistoryGetDiffStats,
307} from 'src/utils/fileHistory.js'
308import {
309 restoreAgentFromSession,
310 restoreSessionStateFromLog,
311} from 'src/utils/sessionRestore.js'
312import { SandboxManager } from 'src/utils/sandbox/sandbox-adapter.js'
313import {
314 headlessProfilerStartTurn,
315 headlessProfilerCheckpoint,
316 logHeadlessProfilerTurn,
317} from 'src/utils/headlessProfiler.js'
318import {
319 startQueryProfile,
320 logQueryProfileReport,
321} from 'src/utils/queryProfiler.js'
322import { asSessionId } from 'src/types/ids.js'
323import { jsonStringify } from '../utils/slowOperations.js'
324import { skillChangeDetector } from '../utils/skills/skillChangeDetector.js'
325import { getCommands, clearCommandsCache } from '../commands.js'
326import {
327 isBareMode,
328 isEnvTruthy,
329 isEnvDefinedFalsy,
330} from '../utils/envUtils.js'
331import { installPluginsForHeadless } from '../utils/plugins/headlessPluginInstall.js'
332import { refreshActivePlugins } from '../utils/plugins/refresh.js'
333import { loadAllPluginsCacheOnly } from '../utils/plugins/pluginLoader.js'
334import {
335 isTeamLead,
336 hasActiveInProcessTeammates,
337 hasWorkingInProcessTeammates,
338 waitForTeammatesToBecomeIdle,
339} from '../utils/teammate.js'
340import {
341 readUnreadMessages,
342 markMessagesAsRead,
343 isShutdownApproved,
344} from '../utils/teammateMailbox.js'
345import { removeTeammateFromTeamFile } from '../utils/swarm/teamHelpers.js'
346import { unassignTeammateTasks } from '../utils/tasks.js'
347import { getRunningTasks } from '../utils/task/framework.js'
348import { isBackgroundTask } from '../tasks/types.js'
349import { stopTask } from '../tasks/stopTask.js'
350import { drainSdkEvents } from '../utils/sdkEventQueue.js'
351import { initializeGrowthBook } from '../services/analytics/growthbook.js'
352import { errorMessage, toError } from '../utils/errors.js'
353import { sleep } from '../utils/sleep.js'
354import { isExtractModeActive } from '../memdir/paths.js'
355
356// Dead code elimination: conditional imports
357/* eslint-disable @typescript-eslint/no-require-imports */
358const coordinatorModeModule = feature('COORDINATOR_MODE')
359 ? (require('../coordinator/coordinatorMode.js') as typeof import('../coordinator/coordinatorMode.js'))
360 : null
361const proactiveModule =
362 feature('PROACTIVE') || feature('KAIROS')
363 ? (require('../proactive/index.js') as typeof import('../proactive/index.js'))
364 : null
365const cronSchedulerModule = feature('AGENT_TRIGGERS')
366 ? (require('../utils/cronScheduler.js') as typeof import('../utils/cronScheduler.js'))
367 : null
368const cronJitterConfigModule = feature('AGENT_TRIGGERS')
369 ? (require('../utils/cronJitterConfig.js') as typeof import('../utils/cronJitterConfig.js'))
370 : null
371const cronGate = feature('AGENT_TRIGGERS')
372 ? (require('../tools/ScheduleCronTool/prompt.js') as typeof import('../tools/ScheduleCronTool/prompt.js'))
373 : null
374const extractMemoriesModule = feature('EXTRACT_MEMORIES')
375 ? (require('../services/extractMemories/extractMemories.js') as typeof import('../services/extractMemories/extractMemories.js'))
376 : null
377/* eslint-enable @typescript-eslint/no-require-imports */
378
379const SHUTDOWN_TEAM_PROMPT = `<system-reminder>
380You are running in non-interactive mode and cannot return a response to the user until your team is shut down.
381
382You MUST shut down your team before preparing your final response:
3831. Use requestShutdown to ask each team member to shut down gracefully
3842. Wait for shutdown approvals
3853. Use the cleanup operation to clean up the team
3864. Only then provide your final response to the user
387
388The user cannot receive your response until the team is completely shut down.
389</system-reminder>
390
391Shut down your team and prepare your final response for the user.`
392
393// Track message UUIDs received during the current session runtime
394const MAX_RECEIVED_UUIDS = 10_000
395const receivedMessageUuids = new Set<UUID>()
396const receivedMessageUuidsOrder: UUID[] = []
397
398function trackReceivedMessageUuid(uuid: UUID): boolean {
399 if (receivedMessageUuids.has(uuid)) {
400 return false // duplicate
401 }
402 receivedMessageUuids.add(uuid)
403 receivedMessageUuidsOrder.push(uuid)
404 // Evict oldest entries when at capacity
405 if (receivedMessageUuidsOrder.length > MAX_RECEIVED_UUIDS) {
406 const toEvict = receivedMessageUuidsOrder.splice(
407 0,
408 receivedMessageUuidsOrder.length - MAX_RECEIVED_UUIDS,
409 )
410 for (const old of toEvict) {
411 receivedMessageUuids.delete(old)
412 }
413 }
414 return true // new UUID
415}
416
417type PromptValue = string | ContentBlockParam[]
418
419function toBlocks(v: PromptValue): ContentBlockParam[] {
420 return typeof v === 'string' ? [{ type: 'text', text: v }] : v
421}
422
423/**
424 * Join prompt values from multiple queued commands into one. Strings are
425 * newline-joined; if any value is a block array, all values are normalized
426 * to blocks and concatenated.
427 */
428export function joinPromptValues(values: PromptValue[]): PromptValue {
429 if (values.length === 1) return values[0]!
430 if (values.every(v => typeof v === 'string')) {
431 return values.join('\n')
432 }
433 return values.flatMap(toBlocks)
434}
435
436/**
437 * Whether `next` can be batched into the same ask() call as `head`. Only
438 * prompt-mode commands batch, and only when the workload tag matches (so the
439 * combined turn is attributed correctly) and the isMeta flag matches (so a
440 * proactive tick can't merge into a user prompt and lose its hidden-in-
441 * transcript marking when the head is spread over the merged command).
442 */
443export function canBatchWith(
444 head: QueuedCommand,
445 next: QueuedCommand | undefined,
446): boolean {
447 return (
448 next !== undefined &&
449 next.mode === 'prompt' &&
450 next.workload === head.workload &&
451 next.isMeta === head.isMeta
452 )
453}
454
455export async function runHeadless(
456 inputPrompt: string | AsyncIterable<string>,
457 getAppState: () => AppState,
458 setAppState: (f: (prev: AppState) => AppState) => void,
459 commands: Command[],
460 tools: Tools,
461 sdkMcpConfigs: Record<string, McpSdkServerConfig>,
462 agents: AgentDefinition[],
463 options: {
464 continue: boolean | undefined
465 resume: string | boolean | undefined
466 resumeSessionAt: string | undefined
467 verbose: boolean | undefined
468 outputFormat: string | undefined
469 jsonSchema: Record<string, unknown> | undefined
470 permissionPromptToolName: string | undefined
471 allowedTools: string[] | undefined
472 thinkingConfig: ThinkingConfig | undefined
473 maxTurns: number | undefined
474 maxBudgetUsd: number | undefined
475 taskBudget: { total: number } | undefined
476 systemPrompt: string | undefined
477 appendSystemPrompt: string | undefined
478 userSpecifiedModel: string | undefined
479 fallbackModel: string | undefined
480 teleport: string | true | null | undefined
481 sdkUrl: string | undefined
482 replayUserMessages: boolean | undefined
483 includePartialMessages: boolean | undefined
484 forkSession: boolean | undefined
485 rewindFiles: string | undefined
486 enableAuthStatus: boolean | undefined
487 agent: string | undefined
488 workload: string | undefined
489 setupTrigger?: 'init' | 'maintenance' | undefined
490 sessionStartHooksPromise?: ReturnType<typeof processSessionStartHooks>
491 setSDKStatus?: (status: SDKStatus) => void
492 },
493): Promise<void> {
494 if (
495 process.env.USER_TYPE === 'ant' &&
496 isEnvTruthy(process.env.CLAUDE_CODE_EXIT_AFTER_FIRST_RENDER)
497 ) {
498 process.stderr.write(
499 `\nStartup time: ${Math.round(process.uptime() * 1000)}ms\n`,
500 )
501 // eslint-disable-next-line custom-rules/no-process-exit
502 process.exit(0)
503 }
504
505 // Fire user settings download now so it overlaps with the MCP/tool setup
506 // below. Managed settings already started in main.tsx preAction; this gives
507 // user settings a similar head start. The cached promise is joined in
508 // installPluginsAndApplyMcpInBackground before plugin install reads
509 // enabledPlugins.
510 if (
511 feature('DOWNLOAD_USER_SETTINGS') &&
512 (isEnvTruthy(process.env.CLAUDE_CODE_REMOTE) || getIsRemoteMode())
513 ) {
514 void downloadUserSettings()
515 }
516
517 // In headless mode there is no React tree, so the useSettingsChange hook
518 // never runs. Subscribe directly so that settings changes (including
519 // managed-settings / policy updates) are fully applied.
520 settingsChangeDetector.subscribe(source => {
521 applySettingsChange(source, setAppState)
522
523 // In headless mode, also sync the denormalized fastMode field from
524 // settings. The TUI manages fastMode via the UI so it skips this.
525 if (isFastModeEnabled()) {
526 setAppState(prev => {
527 const s = prev.settings as Record<string, unknown>
528 const fastMode = s.fastMode === true && !s.fastModePerSessionOptIn
529 return { ...prev, fastMode }
530 })
531 }
532 })
533
534 // Proactive activation is now handled in main.tsx before getTools() so
535 // SleepTool passes isEnabled() filtering. This fallback covers the case
536 // where CLAUDE_CODE_PROACTIVE is set but main.tsx's check didn't fire
537 // (e.g. env was injected by the SDK transport after argv parsing).
538 if (
539 (feature('PROACTIVE') || feature('KAIROS')) &&
540 proactiveModule &&
541 !proactiveModule.isProactiveActive() &&
542 isEnvTruthy(process.env.CLAUDE_CODE_PROACTIVE)
543 ) {
544 proactiveModule.activateProactive('command')
545 }
546
547 // Periodically force a full GC to keep memory usage in check
548 if (typeof Bun !== 'undefined') {
549 const gcTimer = setInterval(Bun.gc, 1000)
550 gcTimer.unref()
551 }
552
553 // Start headless profiler for first turn
554 headlessProfilerStartTurn()
555 headlessProfilerCheckpoint('runHeadless_entry')
556
557 // Check Grove requirements for non-interactive consumer subscribers
558 if (await isQualifiedForGrove()) {
559 await checkGroveForNonInteractive()
560 }
561 headlessProfilerCheckpoint('after_grove_check')
562
563 // Initialize GrowthBook so feature flags take effect in headless mode.
564 // Without this, the disk cache is empty and all flags fall back to defaults.
565 void initializeGrowthBook()
566
567 if (options.resumeSessionAt && !options.resume) {
568 process.stderr.write(`Error: --resume-session-at requires --resume\n`)
569 gracefulShutdownSync(1)
570 return
571 }
572
573 if (options.rewindFiles && !options.resume) {
574 process.stderr.write(`Error: --rewind-files requires --resume\n`)
575 gracefulShutdownSync(1)
576 return
577 }
578
579 if (options.rewindFiles && inputPrompt) {
580 process.stderr.write(
581 `Error: --rewind-files is a standalone operation and cannot be used with a prompt\n`,
582 )
583 gracefulShutdownSync(1)
584 return
585 }
586
587 const structuredIO = getStructuredIO(inputPrompt, options)
588
589 // When emitting NDJSON for SDK clients, any stray write to stdout (debug
590 // prints, dependency console.log, library banners) breaks the client's
591 // line-by-line JSON parser. Install a guard that diverts non-JSON lines to
592 // stderr so the stream stays clean. Must run before the first
593 // structuredIO.write below.
594 if (options.outputFormat === 'stream-json') {
595 installStreamJsonStdoutGuard()
596 }
597
598 // #34044: if user explicitly set sandbox.enabled=true but deps are missing,
599 // isSandboxingEnabled() returns false silently. Surface the reason so users
600 // know their security config isn't being enforced.
601 const sandboxUnavailableReason = SandboxManager.getSandboxUnavailableReason()
602 if (sandboxUnavailableReason) {
603 if (SandboxManager.isSandboxRequired()) {
604 process.stderr.write(
605 `\nError: sandbox required but unavailable: ${sandboxUnavailableReason}\n` +
606 ` sandbox.failIfUnavailable is set — refusing to start without a working sandbox.\n\n`,
607 )
608 gracefulShutdownSync(1)
609 return
610 }
611 process.stderr.write(
612 `\n⚠ Sandbox disabled: ${sandboxUnavailableReason}\n` +
613 ` Commands will run WITHOUT sandboxing. Network and filesystem restrictions will NOT be enforced.\n\n`,
614 )
615 } else if (SandboxManager.isSandboxingEnabled()) {
616 // Initialize sandbox with a callback that forwards network permission
617 // requests to the SDK host via the can_use_tool control_request protocol.
618 // This must happen after structuredIO is created so we can send requests.
619 try {
620 await SandboxManager.initialize(structuredIO.createSandboxAskCallback())
621 } catch (err) {
622 process.stderr.write(`\n❌ Sandbox Error: ${errorMessage(err)}\n`)
623 gracefulShutdownSync(1, 'other')
624 return
625 }
626 }
627
628 if (options.outputFormat === 'stream-json' && options.verbose) {
629 registerHookEventHandler(event => {
630 const message: StdoutMessage = (() => {
631 switch (event.type) {
632 case 'started':
633 return {
634 type: 'system' as const,
635 subtype: 'hook_started' as const,
636 hook_id: event.hookId,
637 hook_name: event.hookName,
638 hook_event: event.hookEvent,
639 uuid: randomUUID(),
640 session_id: getSessionId(),
641 }
642 case 'progress':
643 return {
644 type: 'system' as const,
645 subtype: 'hook_progress' as const,
646 hook_id: event.hookId,
647 hook_name: event.hookName,
648 hook_event: event.hookEvent,
649 stdout: event.stdout,
650 stderr: event.stderr,
651 output: event.output,
652 uuid: randomUUID(),
653 session_id: getSessionId(),
654 }
655 case 'response':
656 return {
657 type: 'system' as const,
658 subtype: 'hook_response' as const,
659 hook_id: event.hookId,
660 hook_name: event.hookName,
661 hook_event: event.hookEvent,
662 output: event.output,
663 stdout: event.stdout,
664 stderr: event.stderr,
665 exit_code: event.exitCode,
666 outcome: event.outcome,
667 uuid: randomUUID(),
668 session_id: getSessionId(),
669 }
670 }
671 })()
672 void structuredIO.write(message)
673 })
674 }
675
676 if (options.setupTrigger) {
677 await processSetupHooks(options.setupTrigger)
678 }
679
680 headlessProfilerCheckpoint('before_loadInitialMessages')
681 const appState = getAppState()
682 const {
683 messages: initialMessages,
684 turnInterruptionState,
685 agentSetting: resumedAgentSetting,
686 } = await loadInitialMessages(setAppState, {
687 continue: options.continue,
688 teleport: options.teleport,
689 resume: options.resume,
690 resumeSessionAt: options.resumeSessionAt,
691 forkSession: options.forkSession,
692 outputFormat: options.outputFormat,
693 sessionStartHooksPromise: options.sessionStartHooksPromise,
694 restoredWorkerState: structuredIO.restoredWorkerState,
695 })
696
697 // SessionStart hooks can emit initialUserMessage — the first user turn for
698 // headless orchestrator sessions where stdin is empty and additionalContext
699 // alone (an attachment, not a turn) would leave the REPL with nothing to
700 // respond to. The hook promise is awaited inside loadInitialMessages, so the
701 // module-level pending value is set by the time we get here.
702 const hookInitialUserMessage = takeInitialUserMessage()
703 if (hookInitialUserMessage) {
704 structuredIO.prependUserMessage(hookInitialUserMessage)
705 }
706
707 // Restore agent setting from the resumed session (if not overridden by current --agent flag
708 // or settings-based agent, which would already have set mainThreadAgentType in main.tsx)
709 if (!options.agent && !getMainThreadAgentType() && resumedAgentSetting) {
710 const { agentDefinition: restoredAgent } = restoreAgentFromSession(
711 resumedAgentSetting,
712 undefined,
713 { activeAgents: agents, allAgents: agents },
714 )
715 if (restoredAgent) {
716 setAppState(prev => ({ ...prev, agent: restoredAgent.agentType }))
717 // Apply the agent's system prompt for non-built-in agents (mirrors main.tsx initial --agent path)
718 if (!options.systemPrompt && !isBuiltInAgent(restoredAgent)) {
719 const agentSystemPrompt = restoredAgent.getSystemPrompt()
720 if (agentSystemPrompt) {
721 options.systemPrompt = agentSystemPrompt
722 }
723 }
724 // Re-persist agent setting so future resumes maintain the agent
725 saveAgentSetting(restoredAgent.agentType)
726 }
727 }
728
729 // gracefulShutdownSync schedules an async shutdown and sets process.exitCode.
730 // If a loadInitialMessages error path triggered it, bail early to avoid
731 // unnecessary work while the process winds down.
732 if (initialMessages.length === 0 && process.exitCode !== undefined) {
733 return
734 }
735
736 // Handle --rewind-files: restore filesystem and exit immediately
737 if (options.rewindFiles) {
738 // File history snapshots are only created for user messages,
739 // so we require the target to be a user message
740 const targetMessage = initialMessages.find(
741 m => m.uuid === options.rewindFiles,
742 )
743
744 if (!targetMessage || targetMessage.type !== 'user') {
745 process.stderr.write(
746 `Error: --rewind-files requires a user message UUID, but ${options.rewindFiles} is not a user message in this session\n`,
747 )
748 gracefulShutdownSync(1)
749 return
750 }
751
752 const currentAppState = getAppState()
753 const result = await handleRewindFiles(
754 options.rewindFiles as UUID,
755 currentAppState,
756 setAppState,
757 false,
758 )
759 if (!result.canRewind) {
760 process.stderr.write(`Error: ${result.error || 'Unexpected error'}\n`)
761 gracefulShutdownSync(1)
762 return
763 }
764
765 // Rewind complete - exit successfully
766 process.stdout.write(
767 `Files rewound to state at message ${options.rewindFiles}\n`,
768 )
769 gracefulShutdownSync(0)
770 return
771 }
772
773 // Check if we need input prompt - skip if we're resuming with a valid session ID/JSONL file or using SDK URL
774 const hasValidResumeSessionId =
775 typeof options.resume === 'string' &&
776 (Boolean(validateUuid(options.resume)) || options.resume.endsWith('.jsonl'))
777 const isUsingSdkUrl = Boolean(options.sdkUrl)
778
779 if (!inputPrompt && !hasValidResumeSessionId && !isUsingSdkUrl) {
780 process.stderr.write(
781 `Error: Input must be provided either through stdin or as a prompt argument when using --print\n`,
782 )
783 gracefulShutdownSync(1)
784 return
785 }
786
787 if (options.outputFormat === 'stream-json' && !options.verbose) {
788 process.stderr.write(
789 'Error: When using --print, --output-format=stream-json requires --verbose\n',
790 )
791 gracefulShutdownSync(1)
792 return
793 }
794
795 // Filter out MCP tools that are in the deny list
796 const allowedMcpTools = filterToolsByDenyRules(
797 appState.mcp.tools,
798 appState.toolPermissionContext,
799 )
800 let filteredTools = [...tools, ...allowedMcpTools]
801
802 // When using SDK URL, always use stdio permission prompting to delegate to the SDK
803 const effectivePermissionPromptToolName = options.sdkUrl
804 ? 'stdio'
805 : options.permissionPromptToolName
806
807 // Callback for when a permission prompt is shown
808 const onPermissionPrompt = (details: RequiresActionDetails) => {
809 if (feature('COMMIT_ATTRIBUTION')) {
810 setAppState(prev => ({
811 ...prev,
812 attribution: {
813 ...prev.attribution,
814 permissionPromptCount: prev.attribution.permissionPromptCount + 1,
815 },
816 }))
817 }
818 notifySessionStateChanged('requires_action', details)
819 }
820
821 const canUseTool = getCanUseToolFn(
822 effectivePermissionPromptToolName,
823 structuredIO,
824 () => getAppState().mcp.tools,
825 onPermissionPrompt,
826 )
827 if (options.permissionPromptToolName) {
828 // Remove the permission prompt tool from the list of available tools.
829 filteredTools = filteredTools.filter(
830 tool => !toolMatchesName(tool, options.permissionPromptToolName!),
831 )
832 }
833
834 // Install errors handlers to gracefully handle broken pipes (e.g., when parent process dies)
835 registerProcessOutputErrorHandlers()
836
837 headlessProfilerCheckpoint('after_loadInitialMessages')
838
839 // Ensure model strings are initialized before generating model options.
840 // For Bedrock users, this waits for the profile fetch to get correct region strings.
841 await ensureModelStringsInitialized()
842 headlessProfilerCheckpoint('after_modelStrings')
843
844 // UDS inbox store registration is deferred until after `run` is defined
845 // so we can pass `run` as the onEnqueue callback (see below).
846
847 // Only `json` + `verbose` needs the full array (jsonStringify(messages) below).
848 // For stream-json (SDK/CCR) and default text output, only the last message is
849 // read for the exit code / final result. Avoid accumulating every message in
850 // memory for the entire session.
851 const needsFullArray = options.outputFormat === 'json' && options.verbose
852 const messages: SDKMessage[] = []
853 let lastMessage: SDKMessage | undefined
854 // Streamlined mode transforms messages when CLAUDE_CODE_STREAMLINED_OUTPUT=true and using stream-json
855 // Build flag gates this out of external builds; env var is the runtime opt-in for ant builds
856 const transformToStreamlined =
857 feature('STREAMLINED_OUTPUT') &&
858 isEnvTruthy(process.env.CLAUDE_CODE_STREAMLINED_OUTPUT) &&
859 options.outputFormat === 'stream-json'
860 ? createStreamlinedTransformer()
861 : null
862
863 headlessProfilerCheckpoint('before_runHeadlessStreaming')
864 for await (const message of runHeadlessStreaming(
865 structuredIO,
866 appState.mcp.clients,
867 [...commands, ...appState.mcp.commands],
868 filteredTools,
869 initialMessages,
870 canUseTool,
871 sdkMcpConfigs,
872 getAppState,
873 setAppState,
874 agents,
875 options,
876 turnInterruptionState,
877 )) {
878 if (transformToStreamlined) {
879 // Streamlined mode: transform messages and stream immediately
880 const transformed = transformToStreamlined(message)
881 if (transformed) {
882 await structuredIO.write(transformed)
883 }
884 } else if (options.outputFormat === 'stream-json' && options.verbose) {
885 await structuredIO.write(message)
886 }
887 // Should not be getting control messages or stream events in non-stream mode.
888 // Also filter out streamlined types since they're only produced by the transformer.
889 // SDK-only system events are excluded so lastMessage stays at the result
890 // (session_state_changed(idle) and any late task_notification drain after
891 // result in the finally block).
892 if (
893 message.type !== 'control_response' &&
894 message.type !== 'control_request' &&
895 message.type !== 'control_cancel_request' &&
896 !(
897 message.type === 'system' &&
898 (message.subtype === 'session_state_changed' ||
899 message.subtype === 'task_notification' ||
900 message.subtype === 'task_started' ||
901 message.subtype === 'task_progress' ||
902 message.subtype === 'post_turn_summary')
903 ) &&
904 message.type !== 'stream_event' &&
905 message.type !== 'keep_alive' &&
906 message.type !== 'streamlined_text' &&
907 message.type !== 'streamlined_tool_use_summary' &&
908 message.type !== 'prompt_suggestion'
909 ) {
910 if (needsFullArray) {
911 messages.push(message)
912 }
913 lastMessage = message
914 }
915 }
916
917 switch (options.outputFormat) {
918 case 'json':
919 if (!lastMessage || lastMessage.type !== 'result') {
920 throw new Error('No messages returned')
921 }
922 if (options.verbose) {
923 writeToStdout(jsonStringify(messages) + '\n')
924 break
925 }
926 writeToStdout(jsonStringify(lastMessage) + '\n')
927 break
928 case 'stream-json':
929 // already logged above
930 break
931 default:
932 if (!lastMessage || lastMessage.type !== 'result') {
933 throw new Error('No messages returned')
934 }
935 switch (lastMessage.subtype) {
936 case 'success':
937 writeToStdout(
938 lastMessage.result.endsWith('\n')
939 ? lastMessage.result
940 : lastMessage.result + '\n',
941 )
942 break
943 case 'error_during_execution':
944 writeToStdout(`Execution error`)
945 break
946 case 'error_max_turns':
947 writeToStdout(`Error: Reached max turns (${options.maxTurns})`)
948 break
949 case 'error_max_budget_usd':
950 writeToStdout(`Error: Exceeded USD budget (${options.maxBudgetUsd})`)
951 break
952 case 'error_max_structured_output_retries':
953 writeToStdout(
954 `Error: Failed to provide valid structured output after maximum retries`,
955 )
956 }
957 }
958
959 // Log headless latency metrics for the final turn
960 logHeadlessProfilerTurn()
961
962 // Drain any in-flight memory extraction before shutdown. The response is
963 // already flushed above, so this adds no user-visible latency — it just
964 // delays process exit so gracefulShutdownSync's 5s failsafe doesn't kill
965 // the forked agent mid-flight. Gated by isExtractModeActive so the
966 // tengu_slate_thimble flag controls non-interactive extraction end-to-end.
967 if (feature('EXTRACT_MEMORIES') && isExtractModeActive()) {
968 await extractMemoriesModule!.drainPendingExtraction()
969 }
970
971 gracefulShutdownSync(
972 lastMessage?.type === 'result' && lastMessage?.is_error ? 1 : 0,
973 )
974}
975
976function runHeadlessStreaming(
977 structuredIO: StructuredIO,
978 mcpClients: MCPServerConnection[],
979 commands: Command[],
980 tools: Tools,
981 initialMessages: Message[],
982 canUseTool: CanUseToolFn,
983 sdkMcpConfigs: Record<string, McpSdkServerConfig>,
984 getAppState: () => AppState,
985 setAppState: (f: (prev: AppState) => AppState) => void,
986 agents: AgentDefinition[],
987 options: {
988 verbose: boolean | undefined
989 jsonSchema: Record<string, unknown> | undefined
990 permissionPromptToolName: string | undefined
991 allowedTools: string[] | undefined
992 thinkingConfig: ThinkingConfig | undefined
993 maxTurns: number | undefined
994 maxBudgetUsd: number | undefined
995 taskBudget: { total: number } | undefined
996 systemPrompt: string | undefined
997 appendSystemPrompt: string | undefined
998 userSpecifiedModel: string | undefined
999 fallbackModel: string | undefined
1000 replayUserMessages?: boolean | undefined
1001 includePartialMessages?: boolean | undefined
1002 enableAuthStatus?: boolean | undefined
1003 agent?: string | undefined
1004 setSDKStatus?: (status: SDKStatus) => void
1005 promptSuggestions?: boolean | undefined
1006 workload?: string | undefined
1007 },
1008 turnInterruptionState?: TurnInterruptionState,
1009): AsyncIterable<StdoutMessage> {
1010 let running = false
1011 let runPhase:
1012 | 'draining_commands'
1013 | 'waiting_for_agents'
1014 | 'finally_flush'
1015 | 'finally_post_flush'
1016 | undefined
1017 let inputClosed = false
1018 let shutdownPromptInjected = false
1019 let heldBackResult: StdoutMessage | null = null
1020 let abortController: AbortController | undefined
1021 // Same queue sendRequest() enqueues to — one FIFO for everything.
1022 const output = structuredIO.outbound
1023
1024 // Ctrl+C in -p mode: abort the in-flight query, then shut down gracefully.
1025 // gracefulShutdown persists session state and flushes analytics, with a
1026 // failsafe timer that force-exits if cleanup hangs.
1027 const sigintHandler = () => {
1028 logForDiagnosticsNoPII('info', 'shutdown_signal', { signal: 'SIGINT' })
1029 if (abortController && !abortController.signal.aborted) {
1030 abortController.abort()
1031 }
1032 void gracefulShutdown(0)
1033 }
1034 process.on('SIGINT', sigintHandler)
1035
1036 // Dump run()'s state at SIGTERM so a stuck session's healthsweep can name
1037 // the do/while(waitingForAgents) poll without reading the transcript.
1038 registerCleanup(async () => {
1039 const bg: Record<string, number> = {}
1040 for (const t of getRunningTasks(getAppState())) {
1041 if (isBackgroundTask(t)) bg[t.type] = (bg[t.type] ?? 0) + 1
1042 }
1043 logForDiagnosticsNoPII('info', 'run_state_at_shutdown', {
1044 run_active: running,
1045 run_phase: runPhase,
1046 worker_status: getSessionState(),
1047 internal_events_pending: structuredIO.internalEventsPending,
1048 bg_tasks: bg,
1049 })
1050 })
1051
1052 // Wire the central onChangeAppState mode-diff hook to the SDK output stream.
1053 // This fires whenever ANY code path mutates toolPermissionContext.mode —
1054 // Shift+Tab, ExitPlanMode dialog, /plan slash command, rewind, bridge
1055 // set_permission_mode, the query loop, stop_task — rather than the two
1056 // paths that previously went through a bespoke wrapper.
1057 // The wrapper's body was fully redundant (it enqueued here AND called
1058 // notifySessionMetadataChanged, both of which onChangeAppState now covers);
1059 // keeping it would double-emit status messages.
1060 setPermissionModeChangedListener(newMode => {
1061 // Only emit for SDK-exposed modes.
1062 if (
1063 newMode === 'default' ||
1064 newMode === 'acceptEdits' ||
1065 newMode === 'bypassPermissions' ||
1066 newMode === 'plan' ||
1067 newMode === (feature('TRANSCRIPT_CLASSIFIER') && 'auto') ||
1068 newMode === 'dontAsk'
1069 ) {
1070 output.enqueue({
1071 type: 'system',
1072 subtype: 'status',
1073 status: null,
1074 permissionMode: newMode as PermissionMode,
1075 uuid: randomUUID(),
1076 session_id: getSessionId(),
1077 })
1078 }
1079 })
1080
1081 // Prompt suggestion tracking (push model)
1082 const suggestionState: {
1083 abortController: AbortController | null
1084 inflightPromise: Promise<void> | null
1085 lastEmitted: {
1086 text: string
1087 emittedAt: number
1088 promptId: PromptVariant
1089 generationRequestId: string | null
1090 } | null
1091 pendingSuggestion: {
1092 type: 'prompt_suggestion'
1093 suggestion: string
1094 uuid: UUID
1095 session_id: string
1096 } | null
1097 pendingLastEmittedEntry: {
1098 text: string
1099 promptId: PromptVariant
1100 generationRequestId: string | null
1101 } | null
1102 } = {
1103 abortController: null,
1104 inflightPromise: null,
1105 lastEmitted: null,
1106 pendingSuggestion: null,
1107 pendingLastEmittedEntry: null,
1108 }
1109
1110 // Set up AWS auth status listener if enabled
1111 let unsubscribeAuthStatus: (() => void) | undefined
1112 if (options.enableAuthStatus) {
1113 const authStatusManager = AwsAuthStatusManager.getInstance()
1114 unsubscribeAuthStatus = authStatusManager.subscribe(status => {
1115 output.enqueue({
1116 type: 'auth_status',
1117 isAuthenticating: status.isAuthenticating,
1118 output: status.output,
1119 error: status.error,
1120 uuid: randomUUID(),
1121 session_id: getSessionId(),
1122 })
1123 })
1124 }
1125
1126 // Set up rate limit status listener to emit SDKRateLimitEvent for all status changes.
1127 // Emitting for all statuses (including 'allowed') ensures consumers can clear warnings
1128 // when rate limits reset. The upstream emitStatusChange already deduplicates via isEqual.
1129 const rateLimitListener = (limits: ClaudeAILimits) => {
1130 const rateLimitInfo = toSDKRateLimitInfo(limits)
1131 if (rateLimitInfo) {
1132 output.enqueue({
1133 type: 'rate_limit_event',
1134 rate_limit_info: rateLimitInfo,
1135 uuid: randomUUID(),
1136 session_id: getSessionId(),
1137 })
1138 }
1139 }
1140 statusListeners.add(rateLimitListener)
1141
1142 // Messages for internal tracking, directly mutated by ask(). These messages
1143 // include Assistant, User, Attachment, and Progress messages.
1144 // TODO: Clean up this code to avoid passing around a mutable array.
1145 const mutableMessages: Message[] = initialMessages
1146
1147 // Seed the readFileState cache from the transcript (content the model saw,
1148 // with message timestamps) so getChangedFiles can detect external edits.
1149 // This cache instance must persist across ask() calls, since the edit tool
1150 // relies on this as a global state.
1151 let readFileState = extractReadFilesFromMessages(
1152 initialMessages,
1153 cwd(),
1154 READ_FILE_STATE_CACHE_SIZE,
1155 )
1156
1157 // Client-supplied readFileState seeds (via seed_read_state control request).
1158 // The stdin IIFE runs concurrently with ask() — a seed arriving mid-turn
1159 // would be lost to ask()'s clone-then-replace (QueryEngine.ts finally block)
1160 // if written directly into readFileState. Instead, seeds land here, merge
1161 // into getReadFileCache's view (readFileState-wins-ties: seeds fill gaps),
1162 // and are re-applied then CLEARED in setReadFileCache. One-shot: each seed
1163 // survives exactly one clone-replace cycle, then becomes a regular
1164 // readFileState entry subject to compact's clear like everything else.
1165 const pendingSeeds = createFileStateCacheWithSizeLimit(
1166 READ_FILE_STATE_CACHE_SIZE,
1167 )
1168
1169 // Auto-resume interrupted turns on restart so CC continues from where it
1170 // left off without requiring the SDK to re-send the prompt.
1171 const resumeInterruptedTurnEnv =
1172 process.env.CLAUDE_CODE_RESUME_INTERRUPTED_TURN
1173 if (
1174 turnInterruptionState &&
1175 turnInterruptionState.kind !== 'none' &&
1176 resumeInterruptedTurnEnv
1177 ) {
1178 logForDebugging(
1179 `[print.ts] Auto-resuming interrupted turn (kind: ${turnInterruptionState.kind})`,
1180 )
1181
1182 // Remove the interrupted message and its sentinel, then re-enqueue so
1183 // the model sees it exactly once. For mid-turn interruptions, the
1184 // deserialization layer transforms them into interrupted_prompt by
1185 // appending a synthetic "Continue from where you left off." message.
1186 removeInterruptedMessage(mutableMessages, turnInterruptionState.message)
1187 enqueue({
1188 mode: 'prompt',
1189 value: turnInterruptionState.message.message.content,
1190 uuid: randomUUID(),
1191 })
1192 }
1193
1194 const modelOptions = getModelOptions()
1195 const modelInfos = modelOptions.map(option => {
1196 const modelId = option.value === null ? 'default' : option.value
1197 const resolvedModel =
1198 modelId === 'default'
1199 ? getDefaultMainLoopModel()
1200 : parseUserSpecifiedModel(modelId)
1201 const hasEffort = modelSupportsEffort(resolvedModel)
1202 const hasAdaptiveThinking = modelSupportsAdaptiveThinking(resolvedModel)
1203 const hasFastMode = isFastModeSupportedByModel(option.value)
1204 const hasAutoMode = modelSupportsAutoMode(resolvedModel)
1205 return {
1206 value: modelId,
1207 displayName: option.label,
1208 description: option.description,
1209 ...(hasEffort && {
1210 supportsEffort: true,
1211 supportedEffortLevels: modelSupportsMaxEffort(resolvedModel)
1212 ? [...EFFORT_LEVELS]
1213 : EFFORT_LEVELS.filter(l => l !== 'max'),
1214 }),
1215 ...(hasAdaptiveThinking && { supportsAdaptiveThinking: true }),
1216 ...(hasFastMode && { supportsFastMode: true }),
1217 ...(hasAutoMode && { supportsAutoMode: true }),
1218 }
1219 })
1220 let activeUserSpecifiedModel = options.userSpecifiedModel
1221
1222 function injectModelSwitchBreadcrumbs(
1223 modelArg: string,
1224 resolvedModel: string,
1225 ): void {
1226 const breadcrumbs = createModelSwitchBreadcrumbs(
1227 modelArg,
1228 modelDisplayString(resolvedModel),
1229 )
1230 mutableMessages.push(...breadcrumbs)
1231 for (const crumb of breadcrumbs) {
1232 if (
1233 typeof crumb.message.content === 'string' &&
1234 crumb.message.content.includes(`<${LOCAL_COMMAND_STDOUT_TAG}>`)
1235 ) {
1236 output.enqueue({
1237 type: 'user',
1238 message: crumb.message,
1239 session_id: getSessionId(),
1240 parent_tool_use_id: null,
1241 uuid: crumb.uuid,
1242 timestamp: crumb.timestamp,
1243 isReplay: true,
1244 } satisfies SDKUserMessageReplay)
1245 }
1246 }
1247 }
1248
1249 // Cache SDK MCP clients to avoid reconnecting on each run
1250 let sdkClients: MCPServerConnection[] = []
1251 let sdkTools: Tools = []
1252
1253 // Track which MCP clients have had elicitation handlers registered
1254 const elicitationRegistered = new Set<string>()
1255
1256 /**
1257 * Register elicitation request/completion handlers on connected MCP clients
1258 * that haven't been registered yet. SDK MCP servers are excluded because they
1259 * route through SdkControlClientTransport. Hooks run first (matching REPL
1260 * behavior); if no hook responds, the request is forwarded to the SDK
1261 * consumer via the control protocol.
1262 */
1263 function registerElicitationHandlers(clients: MCPServerConnection[]): void {
1264 for (const connection of clients) {
1265 if (
1266 connection.type !== 'connected' ||
1267 elicitationRegistered.has(connection.name)
1268 ) {
1269 continue
1270 }
1271 // Skip SDK MCP servers — elicitation flows through SdkControlClientTransport
1272 if (connection.config.type === 'sdk') {
1273 continue
1274 }
1275 const serverName = connection.name
1276
1277 // Wrapped in try/catch because setRequestHandler throws if the client wasn't
1278 // created with elicitation capability declared (e.g., SDK-created clients).
1279 try {
1280 connection.client.setRequestHandler(
1281 ElicitRequestSchema,
1282 async (request, extra) => {
1283 logMCPDebug(
1284 serverName,
1285 `Elicitation request received in print mode: ${jsonStringify(request)}`,
1286 )
1287
1288 const mode = request.params.mode === 'url' ? 'url' : 'form'
1289
1290 logEvent('tengu_mcp_elicitation_shown', {
1291 mode: mode as AnalyticsMetadata_I_VERIFIED_THIS_IS_NOT_CODE_OR_FILEPATHS,
1292 })
1293
1294 // Run elicitation hooks first — they can provide a response programmatically
1295 const hookResponse = await runElicitationHooks(
1296 serverName,
1297 request.params,
1298 extra.signal,
1299 )
1300 if (hookResponse) {
1301 logMCPDebug(
1302 serverName,
1303 `Elicitation resolved by hook: ${jsonStringify(hookResponse)}`,
1304 )
1305 logEvent('tengu_mcp_elicitation_response', {
1306 mode: mode as AnalyticsMetadata_I_VERIFIED_THIS_IS_NOT_CODE_OR_FILEPATHS,
1307 action:
1308 hookResponse.action as AnalyticsMetadata_I_VERIFIED_THIS_IS_NOT_CODE_OR_FILEPATHS,
1309 })
1310 return hookResponse
1311 }
1312
1313 // Delegate to SDK consumer via control protocol
1314 const url =
1315 'url' in request.params
1316 ? (request.params.url as string)
1317 : undefined
1318 const requestedSchema =
1319 'requestedSchema' in request.params
1320 ? (request.params.requestedSchema as
1321 | Record<string, unknown>
1322 | undefined)
1323 : undefined
1324
1325 const elicitationId =
1326 'elicitationId' in request.params
1327 ? (request.params.elicitationId as string | undefined)
1328 : undefined
1329
1330 const rawResult = await structuredIO.handleElicitation(
1331 serverName,
1332 request.params.message,
1333 requestedSchema,
1334 extra.signal,
1335 mode,
1336 url,
1337 elicitationId,
1338 )
1339
1340 const result = await runElicitationResultHooks(
1341 serverName,
1342 rawResult,
1343 extra.signal,
1344 mode,
1345 elicitationId,
1346 )
1347
1348 logEvent('tengu_mcp_elicitation_response', {
1349 mode: mode as AnalyticsMetadata_I_VERIFIED_THIS_IS_NOT_CODE_OR_FILEPATHS,
1350 action:
1351 result.action as AnalyticsMetadata_I_VERIFIED_THIS_IS_NOT_CODE_OR_FILEPATHS,
1352 })
1353 return result
1354 },
1355 )
1356
1357 // Surface completion notifications to SDK consumers (URL mode)
1358 connection.client.setNotificationHandler(
1359 ElicitationCompleteNotificationSchema,
1360 notification => {
1361 const { elicitationId } = notification.params
1362 logMCPDebug(
1363 serverName,
1364 `Elicitation completion notification: ${elicitationId}`,
1365 )
1366 void executeNotificationHooks({
1367 message: `MCP server "${serverName}" confirmed elicitation ${elicitationId} complete`,
1368 notificationType: 'elicitation_complete',
1369 })
1370 output.enqueue({
1371 type: 'system',
1372 subtype: 'elicitation_complete',
1373 mcp_server_name: serverName,
1374 elicitation_id: elicitationId,
1375 uuid: randomUUID(),
1376 session_id: getSessionId(),
1377 })
1378 },
1379 )
1380
1381 elicitationRegistered.add(serverName)
1382 } catch {
1383 // setRequestHandler throws if the client wasn't created with
1384 // elicitation capability — skip silently
1385 }
1386 }
1387 }
1388
1389 async function updateSdkMcp() {
1390 // Check if SDK MCP servers need to be updated (new servers added or removed)
1391 const currentServerNames = new Set(Object.keys(sdkMcpConfigs))
1392 const connectedServerNames = new Set(sdkClients.map(c => c.name))
1393
1394 // Check if there are any differences (additions or removals)
1395 const hasNewServers = Array.from(currentServerNames).some(
1396 name => !connectedServerNames.has(name),
1397 )
1398 const hasRemovedServers = Array.from(connectedServerNames).some(
1399 name => !currentServerNames.has(name),
1400 )
1401 // Check if any SDK clients are pending and need to be upgraded
1402 const hasPendingSdkClients = sdkClients.some(c => c.type === 'pending')
1403 // Check if any SDK clients failed their handshake and need to be retried.
1404 // Without this, a client that lands in 'failed' (e.g. handshake timeout on
1405 // a WS reconnect race) stays failed forever — its name satisfies the
1406 // connectedServerNames diff but it contributes zero tools.
1407 const hasFailedSdkClients = sdkClients.some(c => c.type === 'failed')
1408
1409 const haveServersChanged =
1410 hasNewServers ||
1411 hasRemovedServers ||
1412 hasPendingSdkClients ||
1413 hasFailedSdkClients
1414
1415 if (haveServersChanged) {
1416 // Clean up removed servers
1417 for (const client of sdkClients) {
1418 if (!currentServerNames.has(client.name)) {
1419 if (client.type === 'connected') {
1420 await client.cleanup()
1421 }
1422 }
1423 }
1424
1425 // Re-initialize all SDK MCP servers with current config
1426 const sdkSetup = await setupSdkMcpClients(
1427 sdkMcpConfigs,
1428 (serverName, message) =>
1429 structuredIO.sendMcpMessage(serverName, message),
1430 )
1431 sdkClients = sdkSetup.clients
1432 sdkTools = sdkSetup.tools
1433
1434 // Store SDK MCP tools in appState so subagents can access them via
1435 // assembleToolPool. Only tools are stored here — SDK clients are already
1436 // merged separately in the query loop (allMcpClients) and mcp_status handler.
1437 // Use both old (connectedServerNames) and new (currentServerNames) to remove
1438 // stale SDK tools when servers are added or removed.
1439 const allSdkNames = uniq([...connectedServerNames, ...currentServerNames])
1440 setAppState(prev => ({
1441 ...prev,
1442 mcp: {
1443 ...prev.mcp,
1444 tools: [
1445 ...prev.mcp.tools.filter(
1446 t =>
1447 !allSdkNames.some(name =>
1448 t.name.startsWith(getMcpPrefix(name)),
1449 ),
1450 ),
1451 ...sdkTools,
1452 ],
1453 },
1454 }))
1455
1456 // Set up the special internal VSCode MCP server if necessary.
1457 setupVscodeSdkMcp(sdkClients)
1458 }
1459 }
1460
1461 void updateSdkMcp()
1462
1463 // State for dynamically added MCP servers (via mcp_set_servers control message)
1464 // These are separate from SDK MCP servers and support all transport types
1465 let dynamicMcpState: DynamicMcpState = {
1466 clients: [],
1467 tools: [],
1468 configs: {},
1469 }
1470
1471 // Shared tool assembly for ask() and the get_context_usage control request.
1472 // Closes over the mutable sdkTools/dynamicMcpState bindings so both call
1473 // sites see late-connecting servers.
1474 const buildAllTools = (appState: AppState): Tools => {
1475 const assembledTools = assembleToolPool(
1476 appState.toolPermissionContext,
1477 appState.mcp.tools,
1478 )
1479 let allTools = uniqBy(
1480 mergeAndFilterTools(
1481 [...tools, ...sdkTools, ...dynamicMcpState.tools],
1482 assembledTools,
1483 appState.toolPermissionContext.mode,
1484 ),
1485 'name',
1486 )
1487 if (options.permissionPromptToolName) {
1488 allTools = allTools.filter(
1489 tool => !toolMatchesName(tool, options.permissionPromptToolName!),
1490 )
1491 }
1492 const initJsonSchema = getInitJsonSchema()
1493 if (initJsonSchema && !options.jsonSchema) {
1494 const syntheticOutputResult = createSyntheticOutputTool(initJsonSchema)
1495 if ('tool' in syntheticOutputResult) {
1496 allTools = [...allTools, syntheticOutputResult.tool]
1497 }
1498 }
1499 return allTools
1500 }
1501
1502 // Bridge handle for remote-control (SDK control message).
1503 // Mirrors the REPL's useReplBridge hook: the handle is created when
1504 // `remote_control` is enabled and torn down when disabled.
1505 let bridgeHandle: ReplBridgeHandle | null = null
1506 // Cursor into mutableMessages — tracks how far we've forwarded.
1507 // Same index-based diff as useReplBridge's lastWrittenIndexRef.
1508 let bridgeLastForwardedIndex = 0
1509
1510 // Forward new messages from mutableMessages to the bridge.
1511 // Called incrementally during each turn (so claude.ai sees progress
1512 // and stays alive during permission waits) and again after the turn.
1513 //
1514 // writeMessages has its own UUID-based dedup (initialMessageUUIDs,
1515 // recentPostedUUIDs) — the index cursor here is a pre-filter to avoid
1516 // O(n) re-scanning of already-sent messages on every call.
1517 function forwardMessagesToBridge(): void {
1518 if (!bridgeHandle) return
1519 // Guard against mutableMessages shrinking (compaction truncates it).
1520 const startIndex = Math.min(
1521 bridgeLastForwardedIndex,
1522 mutableMessages.length,
1523 )
1524 const newMessages = mutableMessages
1525 .slice(startIndex)
1526 .filter(m => m.type === 'user' || m.type === 'assistant')
1527 bridgeLastForwardedIndex = mutableMessages.length
1528 if (newMessages.length > 0) {
1529 bridgeHandle.writeMessages(newMessages)
1530 }
1531 }
1532
1533 // Helper to apply MCP server changes - used by both mcp_set_servers control message
1534 // and background plugin installation.
1535 // NOTE: Nested function required - mutates closure state (sdkMcpConfigs, sdkClients, etc.)
1536 let mcpChangesPromise: Promise<{
1537 response: SDKControlMcpSetServersResponse
1538 sdkServersChanged: boolean
1539 }> = Promise.resolve({
1540 response: {
1541 added: [] as string[],
1542 removed: [] as string[],
1543 errors: {} as Record<string, string>,
1544 },
1545 sdkServersChanged: false,
1546 })
1547
1548 function applyMcpServerChanges(
1549 servers: Record<string, McpServerConfigForProcessTransport>,
1550 ): Promise<{
1551 response: SDKControlMcpSetServersResponse
1552 sdkServersChanged: boolean
1553 }> {
1554 // Serialize calls to prevent race conditions between concurrent callers
1555 // (background plugin install and mcp_set_servers control messages)
1556 const doWork = async (): Promise<{
1557 response: SDKControlMcpSetServersResponse
1558 sdkServersChanged: boolean
1559 }> => {
1560 const oldSdkClientNames = new Set(sdkClients.map(c => c.name))
1561
1562 const result = await handleMcpSetServers(
1563 servers,
1564 { configs: sdkMcpConfigs, clients: sdkClients, tools: sdkTools },
1565 dynamicMcpState,
1566 setAppState,
1567 )
1568
1569 // Update SDK state (need to mutate sdkMcpConfigs since it's shared)
1570 for (const key of Object.keys(sdkMcpConfigs)) {
1571 delete sdkMcpConfigs[key]
1572 }
1573 Object.assign(sdkMcpConfigs, result.newSdkState.configs)
1574 sdkClients = result.newSdkState.clients
1575 sdkTools = result.newSdkState.tools
1576 dynamicMcpState = result.newDynamicState
1577
1578 // Keep appState.mcp.tools in sync so subagents can see SDK MCP tools.
1579 // Use both old and new SDK client names to remove stale tools.
1580 if (result.sdkServersChanged) {
1581 const newSdkClientNames = new Set(sdkClients.map(c => c.name))
1582 const allSdkNames = uniq([...oldSdkClientNames, ...newSdkClientNames])
1583 setAppState(prev => ({
1584 ...prev,
1585 mcp: {
1586 ...prev.mcp,
1587 tools: [
1588 ...prev.mcp.tools.filter(
1589 t =>
1590 !allSdkNames.some(name =>
1591 t.name.startsWith(getMcpPrefix(name)),
1592 ),
1593 ),
1594 ...sdkTools,
1595 ],
1596 },
1597 }))
1598 }
1599
1600 return {
1601 response: result.response,
1602 sdkServersChanged: result.sdkServersChanged,
1603 }
1604 }
1605
1606 mcpChangesPromise = mcpChangesPromise.then(doWork, doWork)
1607 return mcpChangesPromise
1608 }
1609
1610 // Build McpServerStatus[] for control responses. Shared by mcp_status and
1611 // reload_plugins handlers. Reads closure state: sdkClients, dynamicMcpState.
1612 function buildMcpServerStatuses(): McpServerStatus[] {
1613 const currentAppState = getAppState()
1614 const currentMcpClients = currentAppState.mcp.clients
1615 const allMcpTools = uniqBy(
1616 [...currentAppState.mcp.tools, ...dynamicMcpState.tools],
1617 'name',
1618 )
1619 const existingNames = new Set([
1620 ...currentMcpClients.map(c => c.name),
1621 ...sdkClients.map(c => c.name),
1622 ])
1623 return [
1624 ...currentMcpClients,
1625 ...sdkClients,
1626 ...dynamicMcpState.clients.filter(c => !existingNames.has(c.name)),
1627 ].map(connection => {
1628 let config
1629 if (
1630 connection.config.type === 'sse' ||
1631 connection.config.type === 'http'
1632 ) {
1633 config = {
1634 type: connection.config.type,
1635 url: connection.config.url,
1636 headers: connection.config.headers,
1637 oauth: connection.config.oauth,
1638 }
1639 } else if (connection.config.type === 'claudeai-proxy') {
1640 config = {
1641 type: 'claudeai-proxy' as const,
1642 url: connection.config.url,
1643 id: connection.config.id,
1644 }
1645 } else if (
1646 connection.config.type === 'stdio' ||
1647 connection.config.type === undefined
1648 ) {
1649 config = {
1650 type: 'stdio' as const,
1651 command: connection.config.command,
1652 args: connection.config.args,
1653 }
1654 }
1655 const serverTools =
1656 connection.type === 'connected'
1657 ? filterToolsByServer(allMcpTools, connection.name).map(tool => ({
1658 name: tool.mcpInfo?.toolName ?? tool.name,
1659 annotations: {
1660 readOnly: tool.isReadOnly({}) || undefined,
1661 destructive: tool.isDestructive?.({}) || undefined,
1662 openWorld: tool.isOpenWorld?.({}) || undefined,
1663 },
1664 }))
1665 : undefined
1666 // Capabilities passthrough with allowlist pre-filter. The IDE reads
1667 // experimental['claude/channel'] to decide whether to show the
1668 // Enable-channel prompt — only echo it if channel_enable would
1669 // actually pass the allowlist. Not a security boundary (the
1670 // handler re-runs the full gate); just avoids dead buttons.
1671 let capabilities: { experimental?: Record<string, unknown> } | undefined
1672 if (
1673 (feature('KAIROS') || feature('KAIROS_CHANNELS')) &&
1674 connection.type === 'connected' &&
1675 connection.capabilities.experimental
1676 ) {
1677 const exp = { ...connection.capabilities.experimental }
1678 if (
1679 exp['claude/channel'] &&
1680 (!isChannelsEnabled() ||
1681 !isChannelAllowlisted(connection.config.pluginSource))
1682 ) {
1683 delete exp['claude/channel']
1684 }
1685 if (Object.keys(exp).length > 0) {
1686 capabilities = { experimental: exp }
1687 }
1688 }
1689 return {
1690 name: connection.name,
1691 status: connection.type,
1692 serverInfo:
1693 connection.type === 'connected' ? connection.serverInfo : undefined,
1694 error: connection.type === 'failed' ? connection.error : undefined,
1695 config,
1696 scope: connection.config.scope,
1697 tools: serverTools,
1698 capabilities,
1699 }
1700 })
1701 }
1702
1703 // NOTE: Nested function required - needs closure access to applyMcpServerChanges and updateSdkMcp
1704 async function installPluginsAndApplyMcpInBackground(): Promise<void> {
1705 try {
1706 // Join point for user settings (fired at runHeadless entry) and managed
1707 // settings (fired in main.tsx preAction). downloadUserSettings() caches
1708 // its promise so this awaits the same in-flight request.
1709 await Promise.all([
1710 feature('DOWNLOAD_USER_SETTINGS') &&
1711 (isEnvTruthy(process.env.CLAUDE_CODE_REMOTE) || getIsRemoteMode())
1712 ? withDiagnosticsTiming('headless_user_settings_download', () =>
1713 downloadUserSettings(),
1714 )
1715 : Promise.resolve(),
1716 withDiagnosticsTiming('headless_managed_settings_wait', () =>
1717 waitForRemoteManagedSettingsToLoad(),
1718 ),
1719 ])
1720
1721 const pluginsInstalled = await installPluginsForHeadless()
1722
1723 if (pluginsInstalled) {
1724 await applyPluginMcpDiff()
1725 }
1726 } catch (error) {
1727 logError(error)
1728 }
1729 }
1730
1731 // Background plugin installation for all headless users
1732 // Installs marketplaces from extraKnownMarketplaces and missing enabled plugins
1733 // CLAUDE_CODE_SYNC_PLUGIN_INSTALL=true: resolved in run() before the first
1734 // query so plugins are guaranteed available on the first ask().
1735 let pluginInstallPromise: Promise<void> | null = null
1736 // --bare / SIMPLE: skip plugin install. Scripted calls don't add plugins
1737 // mid-session; the next interactive run reconciles.
1738 if (!isBareMode()) {
1739 if (isEnvTruthy(process.env.CLAUDE_CODE_SYNC_PLUGIN_INSTALL)) {
1740 pluginInstallPromise = installPluginsAndApplyMcpInBackground()
1741 } else {
1742 void installPluginsAndApplyMcpInBackground()
1743 }
1744 }
1745
1746 // Idle timeout management
1747 const idleTimeout = createIdleTimeoutManager(() => !running)
1748
1749 // Mutable commands and agents for hot reloading
1750 let currentCommands = commands
1751 let currentAgents = agents
1752
1753 // Clear all plugin-related caches, reload commands/agents/hooks.
1754 // Called after CLAUDE_CODE_SYNC_PLUGIN_INSTALL completes (before first query)
1755 // and after non-sync background install finishes.
1756 // refreshActivePlugins calls clearAllCaches() which is required because
1757 // loadAllPlugins() may have run during main.tsx startup BEFORE managed
1758 // settings were fetched. Without clearing, getCommands() would rebuild
1759 // from a stale plugin list.
1760 async function refreshPluginState(): Promise<void> {
1761 // refreshActivePlugins handles the full cache sweep (clearAllCaches),
1762 // reloads all plugin component loaders, writes AppState.plugins +
1763 // AppState.agentDefinitions, registers hooks, and bumps mcp.pluginReconnectKey.
1764 const { agentDefinitions: freshAgentDefs } =
1765 await refreshActivePlugins(setAppState)
1766
1767 // Headless-specific: currentCommands/currentAgents are local mutable refs
1768 // captured by the query loop (REPL uses AppState instead). getCommands is
1769 // fresh because refreshActivePlugins cleared its cache.
1770 currentCommands = await getCommands(cwd())
1771
1772 // Preserve SDK-provided agents (--agents CLI flag or SDK initialize
1773 // control_request) — both inject via parseAgentsFromJson with
1774 // source='flagSettings'. loadMarkdownFilesForSubdir never assigns this
1775 // source, so it cleanly discriminates "injected, not disk-loadable".
1776 //
1777 // The previous filter used a negative set-diff (!freshAgentTypes.has(a))
1778 // which also matched plugin agents that were in the poisoned initial
1779 // currentAgents but correctly excluded from freshAgentDefs after managed
1780 // settings applied — leaking policy-blocked agents into the init message.
1781 // See gh-23085: isBridgeEnabled() at Commander-definition time poisoned
1782 // the settings cache before setEligibility(true) ran.
1783 const sdkAgents = currentAgents.filter(a => a.source === 'flagSettings')
1784 currentAgents = [...freshAgentDefs.allAgents, ...sdkAgents]
1785 }
1786
1787 // Re-diff MCP configs after plugin state changes. Filters to
1788 // process-transport-supported types and carries SDK-mode servers through
1789 // so applyMcpServerChanges' diff doesn't close their transports.
1790 // Nested: needs closure access to sdkMcpConfigs, applyMcpServerChanges,
1791 // updateSdkMcp.
1792 async function applyPluginMcpDiff(): Promise<void> {
1793 const { servers: newConfigs } = await getAllMcpConfigs()
1794 const supportedConfigs: Record<string, McpServerConfigForProcessTransport> =
1795 {}
1796 for (const [name, config] of Object.entries(newConfigs)) {
1797 const type = config.type
1798 if (
1799 type === undefined ||
1800 type === 'stdio' ||
1801 type === 'sse' ||
1802 type === 'http' ||
1803 type === 'sdk'
1804 ) {
1805 supportedConfigs[name] = config
1806 }
1807 }
1808 for (const [name, config] of Object.entries(sdkMcpConfigs)) {
1809 if (config.type === 'sdk' && !(name in supportedConfigs)) {
1810 supportedConfigs[name] = config
1811 }
1812 }
1813 const { response, sdkServersChanged } =
1814 await applyMcpServerChanges(supportedConfigs)
1815 if (sdkServersChanged) {
1816 void updateSdkMcp()
1817 }
1818 logForDebugging(
1819 `Headless MCP refresh: added=${response.added.length}, removed=${response.removed.length}`,
1820 )
1821 }
1822
1823 // Subscribe to skill changes for hot reloading
1824 const unsubscribeSkillChanges = skillChangeDetector.subscribe(() => {
1825 clearCommandsCache()
1826 void getCommands(cwd()).then(newCommands => {
1827 currentCommands = newCommands
1828 })
1829 })
1830
1831 // Proactive mode: schedule a tick to keep the model looping autonomously.
1832 // setTimeout(0) yields to the event loop so pending stdin messages
1833 // (interrupts, user messages) are processed before the tick fires.
1834 const scheduleProactiveTick =
1835 feature('PROACTIVE') || feature('KAIROS')
1836 ? () => {
1837 setTimeout(() => {
1838 if (
1839 !proactiveModule?.isProactiveActive() ||
1840 proactiveModule.isProactivePaused() ||
1841 inputClosed
1842 ) {
1843 return
1844 }
1845 const tickContent = `<${TICK_TAG}>${new Date().toLocaleTimeString()}</${TICK_TAG}>`
1846 enqueue({
1847 mode: 'prompt' as const,
1848 value: tickContent,
1849 uuid: randomUUID(),
1850 priority: 'later',
1851 isMeta: true,
1852 })
1853 void run()
1854 }, 0)
1855 }
1856 : undefined
1857
1858 // Abort the current operation when a 'now' priority message arrives.
1859 subscribeToCommandQueue(() => {
1860 if (abortController && getCommandsByMaxPriority('now').length > 0) {
1861 abortController.abort('interrupt')
1862 }
1863 })
1864
1865 const run = async () => {
1866 if (running) {
1867 return
1868 }
1869
1870 running = true
1871 runPhase = undefined
1872 notifySessionStateChanged('running')
1873 idleTimeout.stop()
1874
1875 headlessProfilerCheckpoint('run_entry')
1876 // TODO(custom-tool-refactor): Should move to the init message, like browser
1877
1878 await updateSdkMcp()
1879 headlessProfilerCheckpoint('after_updateSdkMcp')
1880
1881 // Resolve deferred plugin installation (CLAUDE_CODE_SYNC_PLUGIN_INSTALL).
1882 // The promise was started eagerly so installation overlaps with other init.
1883 // Awaiting here guarantees plugins are available before the first ask().
1884 // If CLAUDE_CODE_SYNC_PLUGIN_INSTALL_TIMEOUT_MS is set, races against that
1885 // deadline and proceeds without plugins on timeout (logging an error).
1886 if (pluginInstallPromise) {
1887 const timeoutMs = parseInt(
1888 process.env.CLAUDE_CODE_SYNC_PLUGIN_INSTALL_TIMEOUT_MS || '',
1889 10,
1890 )
1891 if (timeoutMs > 0) {
1892 const timeout = sleep(timeoutMs).then(() => 'timeout' as const)
1893 const result = await Promise.race([pluginInstallPromise, timeout])
1894 if (result === 'timeout') {
1895 logError(
1896 new Error(
1897 `CLAUDE_CODE_SYNC_PLUGIN_INSTALL: plugin installation timed out after ${timeoutMs}ms`,
1898 ),
1899 )
1900 logEvent('tengu_sync_plugin_install_timeout', {
1901 timeout_ms: timeoutMs,
1902 })
1903 }
1904 } else {
1905 await pluginInstallPromise
1906 }
1907 pluginInstallPromise = null
1908
1909 // Refresh commands, agents, and hooks now that plugins are installed
1910 await refreshPluginState()
1911
1912 // Set up hot-reload for plugin hooks now that the initial install is done.
1913 // In sync-install mode, setup.ts skips this to avoid racing with the install.
1914 const { setupPluginHookHotReload } = await import(
1915 '../utils/plugins/loadPluginHooks.js'
1916 )
1917 setupPluginHookHotReload()
1918 }
1919
1920 // Only main-thread commands (agentId===undefined) — subagent
1921 // notifications are drained by the subagent's mid-turn gate in query.ts.
1922 // Defined outside the try block so it's accessible in the post-finally
1923 // queue re-checks at the bottom of run().
1924 const isMainThread = (cmd: QueuedCommand) => cmd.agentId === undefined
1925
1926 try {
1927 let command: QueuedCommand | undefined
1928 let waitingForAgents = false
1929
1930 // Extract command processing into a named function for the do-while pattern.
1931 // Drains the queue, batching consecutive prompt-mode commands into one
1932 // ask() call so messages that queued up during a long turn coalesce
1933 // into a single follow-up turn instead of N separate turns.
1934 const drainCommandQueue = async () => {
1935 while ((command = dequeue(isMainThread))) {
1936 if (
1937 command.mode !== 'prompt' &&
1938 command.mode !== 'orphaned-permission' &&
1939 command.mode !== 'task-notification'
1940 ) {
1941 throw new Error(
1942 'only prompt commands are supported in streaming mode',
1943 )
1944 }
1945
1946 // Non-prompt commands (task-notification, orphaned-permission) carry
1947 // side effects or orphanedPermission state, so they process singly.
1948 // Prompt commands greedily collect followers with matching workload.
1949 const batch: QueuedCommand[] = [command]
1950 if (command.mode === 'prompt') {
1951 while (canBatchWith(command, peek(isMainThread))) {
1952 batch.push(dequeue(isMainThread)!)
1953 }
1954 if (batch.length > 1) {
1955 command = {
1956 ...command,
1957 value: joinPromptValues(batch.map(c => c.value)),
1958 uuid: batch.findLast(c => c.uuid)?.uuid ?? command.uuid,
1959 }
1960 }
1961 }
1962 const batchUuids = batch.map(c => c.uuid).filter(u => u !== undefined)
1963
1964 // QueryEngine will emit a replay for command.uuid (the last uuid in
1965 // the batch) via its messagesToAck path. Emit replays here for the
1966 // rest so consumers that track per-uuid delivery (clank's
1967 // asyncMessages footer, CCR) see an ack for every message they sent,
1968 // not just the one that survived the merge.
1969 if (options.replayUserMessages && batch.length > 1) {
1970 for (const c of batch) {
1971 if (c.uuid && c.uuid !== command.uuid) {
1972 output.enqueue({
1973 type: 'user',
1974 message: { role: 'user', content: c.value },
1975 session_id: getSessionId(),
1976 parent_tool_use_id: null,
1977 uuid: c.uuid,
1978 isReplay: true,
1979 } satisfies SDKUserMessageReplay)
1980 }
1981 }
1982 }
1983
1984 // Combine all MCP clients. appState.mcp is populated incrementally
1985 // per-server by main.tsx (mirrors useManageMCPConnections). Reading
1986 // fresh per-command means late-connecting servers are visible on the
1987 // next turn. registerElicitationHandlers is idempotent (tracking set).
1988 const appState = getAppState()
1989 const allMcpClients = [
1990 ...appState.mcp.clients,
1991 ...sdkClients,
1992 ...dynamicMcpState.clients,
1993 ]
1994 registerElicitationHandlers(allMcpClients)
1995 // Channel handlers for servers allowlisted via --channels at
1996 // construction time (or enableChannel() mid-session). Runs every
1997 // turn like registerElicitationHandlers — idempotent per-client
1998 // (setNotificationHandler replaces, not stacks) and no-ops for
1999 // non-allowlisted servers (one feature-flag check).
2000 for (const client of allMcpClients) {
2001 reregisterChannelHandlerAfterReconnect(client)
2002 }
2003
2004 const allTools = buildAllTools(appState)
2005
2006 for (const uuid of batchUuids) {
2007 notifyCommandLifecycle(uuid, 'started')
2008 }
2009
2010 // Task notifications arrive when background agents complete.
2011 // Emit an SDK system event for SDK consumers, then fall through
2012 // to ask() so the model sees the agent result and can act on it.
2013 // This matches TUI behavior where useQueueProcessor always feeds
2014 // notifications to the model regardless of coordinator mode.
2015 if (command.mode === 'task-notification') {
2016 const notificationText =
2017 typeof command.value === 'string' ? command.value : ''
2018 // Parse the XML-formatted notification
2019 const taskIdMatch = notificationText.match(
2020 /<task-id>([^<]+)<\/task-id>/,
2021 )
2022 const toolUseIdMatch = notificationText.match(
2023 /<tool-use-id>([^<]+)<\/tool-use-id>/,
2024 )
2025 const outputFileMatch = notificationText.match(
2026 /<output-file>([^<]+)<\/output-file>/,
2027 )
2028 const statusMatch = notificationText.match(
2029 /<status>([^<]+)<\/status>/,
2030 )
2031 const summaryMatch = notificationText.match(
2032 /<summary>([^<]+)<\/summary>/,
2033 )
2034
2035 const isValidStatus = (
2036 s: string | undefined,
2037 ): s is 'completed' | 'failed' | 'stopped' | 'killed' =>
2038 s === 'completed' ||
2039 s === 'failed' ||
2040 s === 'stopped' ||
2041 s === 'killed'
2042 const rawStatus = statusMatch?.[1]
2043 const status = isValidStatus(rawStatus)
2044 ? rawStatus === 'killed'
2045 ? 'stopped'
2046 : rawStatus
2047 : 'completed'
2048
2049 const usageMatch = notificationText.match(
2050 /<usage>([\s\S]*?)<\/usage>/,
2051 )
2052 const usageContent = usageMatch?.[1] ?? ''
2053 const totalTokensMatch = usageContent.match(
2054 /<total_tokens>(\d+)<\/total_tokens>/,
2055 )
2056 const toolUsesMatch = usageContent.match(
2057 /<tool_uses>(\d+)<\/tool_uses>/,
2058 )
2059 const durationMsMatch = usageContent.match(
2060 /<duration_ms>(\d+)<\/duration_ms>/,
2061 )
2062
2063 // Only emit a task_notification SDK event when a <status> tag is
2064 // present — that means this is a terminal notification (completed/
2065 // failed/stopped). Stream events from enqueueStreamEvent carry no
2066 // <status> (they're progress pings); emitting them here would
2067 // default to 'completed' and falsely close the task for SDK
2068 // consumers. Terminal bookends are now emitted directly via
2069 // emitTaskTerminatedSdk, so skipping statusless events is safe.
2070 if (statusMatch) {
2071 output.enqueue({
2072 type: 'system',
2073 subtype: 'task_notification',
2074 task_id: taskIdMatch?.[1] ?? '',
2075 tool_use_id: toolUseIdMatch?.[1],
2076 status,
2077 output_file: outputFileMatch?.[1] ?? '',
2078 summary: summaryMatch?.[1] ?? '',
2079 usage:
2080 totalTokensMatch && toolUsesMatch
2081 ? {
2082 total_tokens: parseInt(totalTokensMatch[1]!, 10),
2083 tool_uses: parseInt(toolUsesMatch[1]!, 10),
2084 duration_ms: durationMsMatch
2085 ? parseInt(durationMsMatch[1]!, 10)
2086 : 0,
2087 }
2088 : undefined,
2089 session_id: getSessionId(),
2090 uuid: randomUUID(),
2091 })
2092 }
2093 // No continue -- fall through to ask() so the model processes the result
2094 }
2095
2096 const input = command.value
2097
2098 if (structuredIO instanceof RemoteIO && command.mode === 'prompt') {
2099 logEvent('tengu_bridge_message_received', {
2100 is_repl: false,
2101 })
2102 }
2103
2104 // Abort any in-flight suggestion generation and track acceptance
2105 suggestionState.abortController?.abort()
2106 suggestionState.abortController = null
2107 suggestionState.pendingSuggestion = null
2108 suggestionState.pendingLastEmittedEntry = null
2109 if (suggestionState.lastEmitted) {
2110 if (command.mode === 'prompt') {
2111 // SDK user messages enqueue ContentBlockParam[], not a plain string
2112 const inputText =
2113 typeof input === 'string'
2114 ? input
2115 : (
2116 input.find(b => b.type === 'text') as
2117 | { type: 'text'; text: string }
2118 | undefined
2119 )?.text
2120 if (typeof inputText === 'string') {
2121 logSuggestionOutcome(
2122 suggestionState.lastEmitted.text,
2123 inputText,
2124 suggestionState.lastEmitted.emittedAt,
2125 suggestionState.lastEmitted.promptId,
2126 suggestionState.lastEmitted.generationRequestId,
2127 )
2128 }
2129 suggestionState.lastEmitted = null
2130 }
2131 }
2132
2133 abortController = createAbortController()
2134 const turnStartTime = feature('FILE_PERSISTENCE')
2135 ? Date.now()
2136 : undefined
2137
2138 headlessProfilerCheckpoint('before_ask')
2139 startQueryProfile()
2140 // Per-iteration ALS context so bg agents spawned inside ask()
2141 // inherit workload across their detached awaits. In-process cron
2142 // stamps cmd.workload; the SDK --workload flag is options.workload.
2143 // const-capture: TS loses `while ((command = dequeue()))` narrowing
2144 // inside the closure.
2145 const cmd = command
2146 await runWithWorkload(cmd.workload ?? options.workload, async () => {
2147 for await (const message of ask({
2148 commands: uniqBy(
2149 [...currentCommands, ...appState.mcp.commands],
2150 'name',
2151 ),
2152 prompt: input,
2153 promptUuid: cmd.uuid,
2154 isMeta: cmd.isMeta,
2155 cwd: cwd(),
2156 tools: allTools,
2157 verbose: options.verbose,
2158 mcpClients: allMcpClients,
2159 thinkingConfig: options.thinkingConfig,
2160 maxTurns: options.maxTurns,
2161 maxBudgetUsd: options.maxBudgetUsd,
2162 taskBudget: options.taskBudget,
2163 canUseTool,
2164 userSpecifiedModel: activeUserSpecifiedModel,
2165 fallbackModel: options.fallbackModel,
2166 jsonSchema: getInitJsonSchema() ?? options.jsonSchema,
2167 mutableMessages,
2168 getReadFileCache: () =>
2169 pendingSeeds.size === 0
2170 ? readFileState
2171 : mergeFileStateCaches(readFileState, pendingSeeds),
2172 setReadFileCache: cache => {
2173 readFileState = cache
2174 for (const [path, seed] of pendingSeeds.entries()) {
2175 const existing = readFileState.get(path)
2176 if (!existing || seed.timestamp > existing.timestamp) {
2177 readFileState.set(path, seed)
2178 }
2179 }
2180 pendingSeeds.clear()
2181 },
2182 customSystemPrompt: options.systemPrompt,
2183 appendSystemPrompt: options.appendSystemPrompt,
2184 getAppState,
2185 setAppState,
2186 abortController,
2187 replayUserMessages: options.replayUserMessages,
2188 includePartialMessages: options.includePartialMessages,
2189 handleElicitation: (serverName, params, elicitSignal) =>
2190 structuredIO.handleElicitation(
2191 serverName,
2192 params.message,
2193 undefined,
2194 elicitSignal,
2195 params.mode,
2196 params.url,
2197 'elicitationId' in params ? params.elicitationId : undefined,
2198 ),
2199 agents: currentAgents,
2200 orphanedPermission: cmd.orphanedPermission,
2201 setSDKStatus: status => {
2202 output.enqueue({
2203 type: 'system',
2204 subtype: 'status',
2205 status,
2206 session_id: getSessionId(),
2207 uuid: randomUUID(),
2208 })
2209 },
2210 })) {
2211 // Forward messages to bridge incrementally (mid-turn) so
2212 // claude.ai sees progress and the connection stays alive
2213 // while blocked on permission requests.
2214 forwardMessagesToBridge()
2215
2216 if (message.type === 'result') {
2217 // Flush pending SDK events so they appear before result on the stream.
2218 for (const event of drainSdkEvents()) {
2219 output.enqueue(event)
2220 }
2221
2222 // Hold-back: don't emit result while background agents are running
2223 const currentState = getAppState()
2224 if (
2225 getRunningTasks(currentState).some(
2226 t =>
2227 (t.type === 'local_agent' ||
2228 t.type === 'local_workflow') &&
2229 isBackgroundTask(t),
2230 )
2231 ) {
2232 heldBackResult = message
2233 } else {
2234 heldBackResult = null
2235 output.enqueue(message)
2236 }
2237 } else {
2238 // Flush SDK events (task_started, task_progress) so background
2239 // agent progress is streamed in real-time, not batched until result.
2240 for (const event of drainSdkEvents()) {
2241 output.enqueue(event)
2242 }
2243 output.enqueue(message)
2244 }
2245 }
2246 }) // end runWithWorkload
2247
2248 for (const uuid of batchUuids) {
2249 notifyCommandLifecycle(uuid, 'completed')
2250 }
2251
2252 // Forward messages to bridge after each turn
2253 forwardMessagesToBridge()
2254 bridgeHandle?.sendResult()
2255
2256 if (feature('FILE_PERSISTENCE') && turnStartTime !== undefined) {
2257 void executeFilePersistence(
2258 turnStartTime,
2259 abortController.signal,
2260 result => {
2261 output.enqueue({
2262 type: 'system' as const,
2263 subtype: 'files_persisted' as const,
2264 files: result.files,
2265 failed: result.failed,
2266 processed_at: new Date().toISOString(),
2267 uuid: randomUUID(),
2268 session_id: getSessionId(),
2269 })
2270 },
2271 )
2272 }
2273
2274 // Generate and emit prompt suggestion for SDK consumers
2275 if (
2276 options.promptSuggestions &&
2277 !isEnvDefinedFalsy(process.env.CLAUDE_CODE_ENABLE_PROMPT_SUGGESTION)
2278 ) {
2279 // TS narrows suggestionState to never in the while loop body;
2280 // cast via unknown to reset narrowing.
2281 const state = suggestionState as unknown as typeof suggestionState
2282 state.abortController?.abort()
2283 const localAbort = new AbortController()
2284 suggestionState.abortController = localAbort
2285
2286 const cacheSafeParams = getLastCacheSafeParams()
2287 if (!cacheSafeParams) {
2288 logSuggestionSuppressed(
2289 'sdk_no_params',
2290 undefined,
2291 undefined,
2292 'sdk',
2293 )
2294 } else {
2295 // Use a ref object so the IIFE's finally can compare against its own
2296 // promise without a self-reference (which upsets TypeScript's flow analysis).
2297 const ref: { promise: Promise<void> | null } = { promise: null }
2298 ref.promise = (async () => {
2299 try {
2300 const result = await tryGenerateSuggestion(
2301 localAbort,
2302 mutableMessages,
2303 getAppState,
2304 cacheSafeParams,
2305 'sdk',
2306 )
2307 if (!result || localAbort.signal.aborted) return
2308 const suggestionMsg = {
2309 type: 'prompt_suggestion' as const,
2310 suggestion: result.suggestion,
2311 uuid: randomUUID(),
2312 session_id: getSessionId(),
2313 }
2314 const lastEmittedEntry = {
2315 text: result.suggestion,
2316 emittedAt: Date.now(),
2317 promptId: result.promptId,
2318 generationRequestId: result.generationRequestId,
2319 }
2320 // Defer emission if the result is being held for background agents,
2321 // so that prompt_suggestion always arrives after result.
2322 // Only set lastEmitted when the suggestion is actually delivered
2323 // to the consumer; deferred suggestions may be discarded before
2324 // delivery if a new command arrives first.
2325 if (heldBackResult) {
2326 suggestionState.pendingSuggestion = suggestionMsg
2327 suggestionState.pendingLastEmittedEntry = {
2328 text: lastEmittedEntry.text,
2329 promptId: lastEmittedEntry.promptId,
2330 generationRequestId: lastEmittedEntry.generationRequestId,
2331 }
2332 } else {
2333 suggestionState.lastEmitted = lastEmittedEntry
2334 output.enqueue(suggestionMsg)
2335 }
2336 } catch (error) {
2337 if (
2338 error instanceof Error &&
2339 (error.name === 'AbortError' ||
2340 error.name === 'APIUserAbortError')
2341 ) {
2342 logSuggestionSuppressed(
2343 'aborted',
2344 undefined,
2345 undefined,
2346 'sdk',
2347 )
2348 return
2349 }
2350 logError(toError(error))
2351 } finally {
2352 if (suggestionState.inflightPromise === ref.promise) {
2353 suggestionState.inflightPromise = null
2354 }
2355 }
2356 })()
2357 suggestionState.inflightPromise = ref.promise
2358 }
2359 }
2360
2361 // Log headless profiler metrics for this turn and start next turn
2362 logHeadlessProfilerTurn()
2363 logQueryProfileReport()
2364 headlessProfilerStartTurn()
2365 }
2366 }
2367
2368 // Use a do-while loop to drain commands and then wait for any
2369 // background agents that are still running. When agents complete,
2370 // their notifications are enqueued and the loop re-drains.
2371 do {
2372 // Drain SDK events (task_started, task_progress) before command queue
2373 // so progress events precede task_notification on the stream.
2374 for (const event of drainSdkEvents()) {
2375 output.enqueue(event)
2376 }
2377
2378 runPhase = 'draining_commands'
2379 await drainCommandQueue()
2380
2381 // Check for running background tasks before exiting.
2382 // Exclude in_process_teammate — teammates are long-lived by design
2383 // (status: 'running' for their whole lifetime, cleaned up by the
2384 // shutdown protocol, not by transitioning to 'completed'). Waiting
2385 // on them here loops forever (gh-30008). Same exclusion already
2386 // exists at useBackgroundTaskNavigation.ts:55 for the same reason;
2387 // L1839 above is already narrower (type === 'local_agent') so it
2388 // doesn't hit this.
2389 waitingForAgents = false
2390 {
2391 const state = getAppState()
2392 const hasRunningBg = getRunningTasks(state).some(
2393 t => isBackgroundTask(t) && t.type !== 'in_process_teammate',
2394 )
2395 const hasMainThreadQueued = peek(isMainThread) !== undefined
2396 if (hasRunningBg || hasMainThreadQueued) {
2397 waitingForAgents = true
2398 if (!hasMainThreadQueued) {
2399 runPhase = 'waiting_for_agents'
2400 // No commands ready yet, wait for tasks to complete
2401 await sleep(100)
2402 }
2403 // Loop back to drain any newly queued commands
2404 }
2405 }
2406 } while (waitingForAgents)
2407
2408 if (heldBackResult) {
2409 output.enqueue(heldBackResult)
2410 heldBackResult = null
2411 if (suggestionState.pendingSuggestion) {
2412 output.enqueue(suggestionState.pendingSuggestion)
2413 // Now that the suggestion is actually delivered, record it for acceptance tracking
2414 if (suggestionState.pendingLastEmittedEntry) {
2415 suggestionState.lastEmitted = {
2416 ...suggestionState.pendingLastEmittedEntry,
2417 emittedAt: Date.now(),
2418 }
2419 suggestionState.pendingLastEmittedEntry = null
2420 }
2421 suggestionState.pendingSuggestion = null
2422 }
2423 }
2424 } catch (error) {
2425 // Emit error result message before shutting down
2426 // Write directly to structuredIO to ensure immediate delivery
2427 try {
2428 await structuredIO.write({
2429 type: 'result',
2430 subtype: 'error_during_execution',
2431 duration_ms: 0,
2432 duration_api_ms: 0,
2433 is_error: true,
2434 num_turns: 0,
2435 stop_reason: null,
2436 session_id: getSessionId(),
2437 total_cost_usd: 0,
2438 usage: EMPTY_USAGE,
2439 modelUsage: {},
2440 permission_denials: [],
2441 uuid: randomUUID(),
2442 errors: [
2443 errorMessage(error),
2444 ...getInMemoryErrors().map(_ => _.error),
2445 ],
2446 })
2447 } catch {
2448 // If we can't emit the error result, continue with shutdown anyway
2449 }
2450 suggestionState.abortController?.abort()
2451 gracefulShutdownSync(1)
2452 return
2453 } finally {
2454 runPhase = 'finally_flush'
2455 // Flush pending internal events before going idle
2456 await structuredIO.flushInternalEvents()
2457 runPhase = 'finally_post_flush'
2458 if (!isShuttingDown()) {
2459 notifySessionStateChanged('idle')
2460 // Drain so the idle session_state_changed SDK event (plus any
2461 // terminal task_notification bookends emitted during bg-agent
2462 // teardown) reach the output stream before we block on the next
2463 // command. The do-while drain above only runs while
2464 // waitingForAgents; once we're here the next drain would be the
2465 // top of the next run(), which won't come if input is idle.
2466 for (const event of drainSdkEvents()) {
2467 output.enqueue(event)
2468 }
2469 }
2470 running = false
2471 // Start idle timer when we finish processing and are waiting for input
2472 idleTimeout.start()
2473 }
2474
2475 // Proactive tick: if proactive is active and queue is empty, inject a tick
2476 if (
2477 (feature('PROACTIVE') || feature('KAIROS')) &&
2478 proactiveModule?.isProactiveActive() &&
2479 !proactiveModule.isProactivePaused()
2480 ) {
2481 if (peek(isMainThread) === undefined && !inputClosed) {
2482 scheduleProactiveTick!()
2483 return
2484 }
2485 }
2486
2487 // Re-check the queue after releasing the mutex. A message may have
2488 // arrived (and called run()) between the last dequeue() returning
2489 // undefined and `running = false` above. In that case the caller
2490 // saw `running === true` and returned immediately, leaving the
2491 // message stranded in the queue with no one to process it.
2492 if (peek(isMainThread) !== undefined) {
2493 void run()
2494 return
2495 }
2496
2497 // Check for unread teammate messages and process them
2498 // This mirrors what useInboxPoller does in interactive REPL mode
2499 // Poll until no more messages (teammates may still be working)
2500 {
2501 const currentAppState = getAppState()
2502 const teamContext = currentAppState.teamContext
2503
2504 if (teamContext && isTeamLead(teamContext)) {
2505 const agentName = 'team-lead'
2506
2507 // Poll for messages while teammates are active
2508 // This is needed because teammates may send messages while we're waiting
2509 // Keep polling until the team is shut down
2510 const POLL_INTERVAL_MS = 500
2511
2512 while (true) {
2513 // Check if teammates are still active
2514 const refreshedState = getAppState()
2515 const hasActiveTeammates =
2516 hasActiveInProcessTeammates(refreshedState) ||
2517 (refreshedState.teamContext &&
2518 Object.keys(refreshedState.teamContext.teammates).length > 0)
2519
2520 if (!hasActiveTeammates) {
2521 logForDebugging(
2522 '[print.ts] No more active teammates, stopping poll',
2523 )
2524 break
2525 }
2526
2527 const unread = await readUnreadMessages(
2528 agentName,
2529 refreshedState.teamContext?.teamName,
2530 )
2531
2532 if (unread.length > 0) {
2533 logForDebugging(
2534 `[print.ts] Team-lead found ${unread.length} unread messages`,
2535 )
2536
2537 // Mark as read immediately to avoid duplicate processing
2538 await markMessagesAsRead(
2539 agentName,
2540 refreshedState.teamContext?.teamName,
2541 )
2542
2543 // Process shutdown_approved messages - remove teammates from team file
2544 // This mirrors what useInboxPoller does in interactive mode (lines 546-606)
2545 const teamName = refreshedState.teamContext?.teamName
2546 for (const m of unread) {
2547 const shutdownApproval = isShutdownApproved(m.text)
2548 if (shutdownApproval && teamName) {
2549 const teammateToRemove = shutdownApproval.from
2550 logForDebugging(
2551 `[print.ts] Processing shutdown_approved from ${teammateToRemove}`,
2552 )
2553
2554 // Find the teammate ID by name
2555 const teammateId = refreshedState.teamContext?.teammates
2556 ? Object.entries(refreshedState.teamContext.teammates).find(
2557 ([, t]) => t.name === teammateToRemove,
2558 )?.[0]
2559 : undefined
2560
2561 if (teammateId) {
2562 // Remove from team file
2563 removeTeammateFromTeamFile(teamName, {
2564 agentId: teammateId,
2565 name: teammateToRemove,
2566 })
2567 logForDebugging(
2568 `[print.ts] Removed ${teammateToRemove} from team file`,
2569 )
2570
2571 // Unassign tasks owned by this teammate
2572 await unassignTeammateTasks(
2573 teamName,
2574 teammateId,
2575 teammateToRemove,
2576 'shutdown',
2577 )
2578
2579 // Remove from teamContext in AppState
2580 setAppState(prev => {
2581 if (!prev.teamContext?.teammates) return prev
2582 if (!(teammateId in prev.teamContext.teammates)) return prev
2583 const { [teammateId]: _, ...remainingTeammates } =
2584 prev.teamContext.teammates
2585 return {
2586 ...prev,
2587 teamContext: {
2588 ...prev.teamContext,
2589 teammates: remainingTeammates,
2590 },
2591 }
2592 })
2593 }
2594 }
2595 }
2596
2597 // Format messages same as useInboxPoller
2598 const formatted = unread
2599 .map(
2600 (m: { from: string; text: string; color?: string }) =>
2601 `<${TEAMMATE_MESSAGE_TAG} teammate_id="${m.from}"${m.color ? ` color="${m.color}"` : ''}>\n${m.text}\n</${TEAMMATE_MESSAGE_TAG}>`,
2602 )
2603 .join('\n\n')
2604
2605 // Enqueue and process
2606 enqueue({
2607 mode: 'prompt',
2608 value: formatted,
2609 uuid: randomUUID(),
2610 })
2611 void run()
2612 return // run() will come back here after processing
2613 }
2614
2615 // No messages - check if we need to prompt for shutdown
2616 // If input is closed and teammates are active, inject shutdown prompt once
2617 if (inputClosed && !shutdownPromptInjected) {
2618 shutdownPromptInjected = true
2619 logForDebugging(
2620 '[print.ts] Input closed with active teammates, injecting shutdown prompt',
2621 )
2622 enqueue({
2623 mode: 'prompt',
2624 value: SHUTDOWN_TEAM_PROMPT,
2625 uuid: randomUUID(),
2626 })
2627 void run()
2628 return // run() will come back here after processing
2629 }
2630
2631 // Wait and check again
2632 await sleep(POLL_INTERVAL_MS)
2633 }
2634 }
2635 }
2636
2637 if (inputClosed) {
2638 // Check for active swarm that needs shutdown
2639 const hasActiveSwarm = await (async () => {
2640 // Wait for any working in-process team members to finish
2641 const currentAppState = getAppState()
2642 if (hasWorkingInProcessTeammates(currentAppState)) {
2643 await waitForTeammatesToBecomeIdle(setAppState, currentAppState)
2644 }
2645
2646 // Re-fetch state after potential wait
2647 const refreshedAppState = getAppState()
2648 const refreshedTeamContext = refreshedAppState.teamContext
2649 const hasTeamMembersNotCleanedUp =
2650 refreshedTeamContext &&
2651 Object.keys(refreshedTeamContext.teammates).length > 0
2652
2653 return (
2654 hasTeamMembersNotCleanedUp ||
2655 hasActiveInProcessTeammates(refreshedAppState)
2656 )
2657 })()
2658
2659 if (hasActiveSwarm) {
2660 // Team members are idle or pane-based - inject prompt to shut down team
2661 enqueue({
2662 mode: 'prompt',
2663 value: SHUTDOWN_TEAM_PROMPT,
2664 uuid: randomUUID(),
2665 })
2666 void run()
2667 } else {
2668 // Wait for any in-flight push suggestion before closing the output stream.
2669 if (suggestionState.inflightPromise) {
2670 await Promise.race([suggestionState.inflightPromise, sleep(5000)])
2671 }
2672 suggestionState.abortController?.abort()
2673 suggestionState.abortController = null
2674 await finalizePendingAsyncHooks()
2675 unsubscribeSkillChanges()
2676 unsubscribeAuthStatus?.()
2677 statusListeners.delete(rateLimitListener)
2678 output.done()
2679 }
2680 }
2681 }
2682
2683 // Set up UDS inbox callback so the query loop is kicked off
2684 // when a message arrives via the UDS socket in headless mode.
2685 if (feature('UDS_INBOX')) {
2686 /* eslint-disable @typescript-eslint/no-require-imports */
2687 const { setOnEnqueue } = require('../utils/udsMessaging.js')
2688 /* eslint-enable @typescript-eslint/no-require-imports */
2689 setOnEnqueue(() => {
2690 if (!inputClosed) {
2691 void run()
2692 }
2693 })
2694 }
2695
2696 // Cron scheduler: runs scheduled_tasks.json tasks in SDK/-p mode.
2697 // Mirrors REPL's useScheduledTasks hook. Fired prompts enqueue + kick
2698 // off run() directly — unlike REPL, there's no queue subscriber here
2699 // that drains on enqueue while idle. The run() mutex makes this safe
2700 // during an active turn: the call no-ops and the post-run recheck at
2701 // the end of run() picks up the queued command.
2702 let cronScheduler: import('../utils/cronScheduler.js').CronScheduler | null =
2703 null
2704 if (
2705 feature('AGENT_TRIGGERS') &&
2706 cronSchedulerModule &&
2707 cronGate?.isKairosCronEnabled()
2708 ) {
2709 cronScheduler = cronSchedulerModule.createCronScheduler({
2710 onFire: prompt => {
2711 if (inputClosed) return
2712 enqueue({
2713 mode: 'prompt',
2714 value: prompt,
2715 uuid: randomUUID(),
2716 priority: 'later',
2717 // System-generated — matches useScheduledTasks.ts REPL equivalent.
2718 // Without this, messages.ts metaProp eval is {} → prompt leaks
2719 // into visible transcript when cron fires mid-turn in -p mode.
2720 isMeta: true,
2721 // Threaded to cc_workload= in the billing-header attribution block
2722 // so the API can serve cron requests at lower QoS. drainCommandQueue
2723 // reads this per-iteration and hoists it into bootstrap state for
2724 // the ask() call.
2725 workload: WORKLOAD_CRON,
2726 })
2727 void run()
2728 },
2729 isLoading: () => running || inputClosed,
2730 getJitterConfig: cronJitterConfigModule?.getCronJitterConfig,
2731 isKilled: () => !cronGate?.isKairosCronEnabled(),
2732 })
2733 cronScheduler.start()
2734 }
2735
2736 const sendControlResponseSuccess = function (
2737 message: SDKControlRequest,
2738 response?: Record<string, unknown>,
2739 ) {
2740 output.enqueue({
2741 type: 'control_response',
2742 response: {
2743 subtype: 'success',
2744 request_id: message.request_id,
2745 response: response,
2746 },
2747 })
2748 }
2749
2750 const sendControlResponseError = function (
2751 message: SDKControlRequest,
2752 errorMessage: string,
2753 ) {
2754 output.enqueue({
2755 type: 'control_response',
2756 response: {
2757 subtype: 'error',
2758 request_id: message.request_id,
2759 error: errorMessage,
2760 },
2761 })
2762 }
2763
2764 // Handle unexpected permission responses by looking up the unresolved tool
2765 // call in the transcript and executing it
2766 const handledOrphanedToolUseIds = new Set<string>()
2767 structuredIO.setUnexpectedResponseCallback(async message => {
2768 await handleOrphanedPermissionResponse({
2769 message,
2770 setAppState,
2771 handledToolUseIds: handledOrphanedToolUseIds,
2772 onEnqueued: () => {
2773 // The first message of a session might be the orphaned permission
2774 // check rather than a user prompt, so kick off the loop.
2775 void run()
2776 },
2777 })
2778 })
2779
2780 // Track active OAuth flows per server so we can abort a previous flow
2781 // when a new mcp_authenticate request arrives for the same server.
2782 const activeOAuthFlows = new Map<string, AbortController>()
2783 // Track manual callback URL submit functions for active OAuth flows.
2784 // Used when localhost is not reachable (e.g., browser-based IDEs).
2785 const oauthCallbackSubmitters = new Map<
2786 string,
2787 (callbackUrl: string) => void
2788 >()
2789 // Track servers where the manual callback was actually invoked (so the
2790 // automatic reconnect path knows to skip — the extension will reconnect).
2791 const oauthManualCallbackUsed = new Set<string>()
2792 // Track OAuth auth-only promises so mcp_oauth_callback_url can await
2793 // token exchange completion. Reconnect is handled separately by the
2794 // extension via handleAuthDone → mcp_reconnect.
2795 const oauthAuthPromises = new Map<string, Promise<void>>()
2796
2797 // In-flight Anthropic OAuth flow (claude_authenticate). Single-slot: a
2798 // second authenticate request cleans up the first. The service holds the
2799 // PKCE verifier + localhost listener; the promise settles after
2800 // installOAuthTokens — after it resolves, the in-process memoized token
2801 // cache is already cleared and the next API call picks up the new creds.
2802 let claudeOAuth: {
2803 service: OAuthService
2804 flow: Promise<void>
2805 } | null = null
2806
2807 // This is essentially spawning a parallel async task- we have two
2808 // running in parallel- one reading from stdin and adding to the
2809 // queue to be processed and another reading from the queue,
2810 // processing and returning the result of the generation.
2811 // The process is complete when the input stream completes and
2812 // the last generation of the queue has complete.
2813 void (async () => {
2814 let initialized = false
2815 logForDiagnosticsNoPII('info', 'cli_message_loop_started')
2816 for await (const message of structuredIO.structuredInput) {
2817 // Non-user events are handled inline (no queue). started→completed in
2818 // the same tick carries no information, so only fire completed.
2819 // control_response is reported by StructuredIO.processLine (which also
2820 // sees orphans that never yield here).
2821 const eventId = 'uuid' in message ? message.uuid : undefined
2822 if (
2823 eventId &&
2824 message.type !== 'user' &&
2825 message.type !== 'control_response'
2826 ) {
2827 notifyCommandLifecycle(eventId, 'completed')
2828 }
2829
2830 if (message.type === 'control_request') {
2831 if (message.request.subtype === 'interrupt') {
2832 // Track escapes for attribution (ant-only feature)
2833 if (feature('COMMIT_ATTRIBUTION')) {
2834 setAppState(prev => ({
2835 ...prev,
2836 attribution: {
2837 ...prev.attribution,
2838 escapeCount: prev.attribution.escapeCount + 1,
2839 },
2840 }))
2841 }
2842 if (abortController) {
2843 abortController.abort()
2844 }
2845 suggestionState.abortController?.abort()
2846 suggestionState.abortController = null
2847 suggestionState.lastEmitted = null
2848 suggestionState.pendingSuggestion = null
2849 sendControlResponseSuccess(message)
2850 } else if (message.request.subtype === 'end_session') {
2851 logForDebugging(
2852 `[print.ts] end_session received, reason=${message.request.reason ?? 'unspecified'}`,
2853 )
2854 if (abortController) {
2855 abortController.abort()
2856 }
2857 suggestionState.abortController?.abort()
2858 suggestionState.abortController = null
2859 suggestionState.lastEmitted = null
2860 suggestionState.pendingSuggestion = null
2861 sendControlResponseSuccess(message)
2862 break // exits for-await → falls through to inputClosed=true drain below
2863 } else if (message.request.subtype === 'initialize') {
2864 // SDK MCP server names from the initialize message
2865 // Populated by both browser and ProcessTransport sessions
2866 if (
2867 message.request.sdkMcpServers &&
2868 message.request.sdkMcpServers.length > 0
2869 ) {
2870 for (const serverName of message.request.sdkMcpServers) {
2871 // Create placeholder config for SDK MCP servers
2872 // The actual server connection is managed by the SDK Query class
2873 sdkMcpConfigs[serverName] = {
2874 type: 'sdk',
2875 name: serverName,
2876 }
2877 }
2878 }
2879
2880 await handleInitializeRequest(
2881 message.request,
2882 message.request_id,
2883 initialized,
2884 output,
2885 commands,
2886 modelInfos,
2887 structuredIO,
2888 !!options.enableAuthStatus,
2889 options,
2890 agents,
2891 getAppState,
2892 )
2893
2894 // Enable prompt suggestions in AppState when SDK consumer opts in.
2895 // shouldEnablePromptSuggestion() returns false for non-interactive
2896 // sessions, but the SDK consumer explicitly requested suggestions.
2897 if (message.request.promptSuggestions) {
2898 setAppState(prev => {
2899 if (prev.promptSuggestionEnabled) return prev
2900 return { ...prev, promptSuggestionEnabled: true }
2901 })
2902 }
2903
2904 if (
2905 message.request.agentProgressSummaries &&
2906 getFeatureValue_CACHED_MAY_BE_STALE('tengu_slate_prism', true)
2907 ) {
2908 setSdkAgentProgressSummariesEnabled(true)
2909 }
2910
2911 initialized = true
2912
2913 // If the auto-resume logic pre-enqueued a command, drain it now
2914 // that initialize has set up systemPrompt, agents, hooks, etc.
2915 if (hasCommandsInQueue()) {
2916 void run()
2917 }
2918 } else if (message.request.subtype === 'set_permission_mode') {
2919 const m = message.request // for typescript (TODO: use readonly types to avoid this)
2920 setAppState(prev => ({
2921 ...prev,
2922 toolPermissionContext: handleSetPermissionMode(
2923 m,
2924 message.request_id,
2925 prev.toolPermissionContext,
2926 output,
2927 ),
2928 isUltraplanMode: m.ultraplan ?? prev.isUltraplanMode,
2929 }))
2930 // handleSetPermissionMode sends the control_response; the
2931 // notifySessionMetadataChanged that used to follow here is
2932 // now fired by onChangeAppState (with externalized mode name).
2933 } else if (message.request.subtype === 'set_model') {
2934 const requestedModel = message.request.model ?? 'default'
2935 const model =
2936 requestedModel === 'default'
2937 ? getDefaultMainLoopModel()
2938 : requestedModel
2939 activeUserSpecifiedModel = model
2940 setMainLoopModelOverride(model)
2941 notifySessionMetadataChanged({ model })
2942 injectModelSwitchBreadcrumbs(requestedModel, model)
2943
2944 sendControlResponseSuccess(message)
2945 } else if (message.request.subtype === 'set_max_thinking_tokens') {
2946 if (message.request.max_thinking_tokens === null) {
2947 options.thinkingConfig = undefined
2948 } else if (message.request.max_thinking_tokens === 0) {
2949 options.thinkingConfig = { type: 'disabled' }
2950 } else {
2951 options.thinkingConfig = {
2952 type: 'enabled',
2953 budgetTokens: message.request.max_thinking_tokens,
2954 }
2955 }
2956 sendControlResponseSuccess(message)
2957 } else if (message.request.subtype === 'mcp_status') {
2958 sendControlResponseSuccess(message, {
2959 mcpServers: buildMcpServerStatuses(),
2960 })
2961 } else if (message.request.subtype === 'get_context_usage') {
2962 try {
2963 const appState = getAppState()
2964 const data = await collectContextData({
2965 messages: mutableMessages,
2966 getAppState,
2967 options: {
2968 mainLoopModel: getMainLoopModel(),
2969 tools: buildAllTools(appState),
2970 agentDefinitions: appState.agentDefinitions,
2971 customSystemPrompt: options.systemPrompt,
2972 appendSystemPrompt: options.appendSystemPrompt,
2973 },
2974 })
2975 sendControlResponseSuccess(message, { ...data })
2976 } catch (error) {
2977 sendControlResponseError(message, errorMessage(error))
2978 }
2979 } else if (message.request.subtype === 'mcp_message') {
2980 // Handle MCP notifications from SDK servers
2981 const mcpRequest = message.request
2982 const sdkClient = sdkClients.find(
2983 client => client.name === mcpRequest.server_name,
2984 )
2985 // Check client exists - dynamically added SDK servers may have
2986 // placeholder clients with null client until updateSdkMcp() runs
2987 if (
2988 sdkClient &&
2989 sdkClient.type === 'connected' &&
2990 sdkClient.client?.transport?.onmessage
2991 ) {
2992 sdkClient.client.transport.onmessage(mcpRequest.message)
2993 }
2994 sendControlResponseSuccess(message)
2995 } else if (message.request.subtype === 'rewind_files') {
2996 const appState = getAppState()
2997 const result = await handleRewindFiles(
2998 message.request.user_message_id as UUID,
2999 appState,
3000 setAppState,
3001 message.request.dry_run ?? false,
3002 )
3003 if (result.canRewind || message.request.dry_run) {
3004 sendControlResponseSuccess(message, result)
3005 } else {
3006 sendControlResponseError(
3007 message,
3008 result.error ?? 'Unexpected error',
3009 )
3010 }
3011 } else if (message.request.subtype === 'cancel_async_message') {
3012 const targetUuid = message.request.message_uuid
3013 const removed = dequeueAllMatching(cmd => cmd.uuid === targetUuid)
3014 sendControlResponseSuccess(message, {
3015 cancelled: removed.length > 0,
3016 })
3017 } else if (message.request.subtype === 'seed_read_state') {
3018 // Client observed a Read that was later removed from context (e.g.
3019 // by snip), so transcript-based seeding missed it. Queued into
3020 // pendingSeeds; applied at the next clone-replace boundary.
3021 try {
3022 // expandPath: all other readFileState writers normalize (~, relative,
3023 // session cwd vs process cwd). FileEditTool looks up by expandPath'd
3024 // key — a verbatim client path would miss.
3025 const normalizedPath = expandPath(message.request.path)
3026 // Check disk mtime before reading content. If the file changed
3027 // since the client's observation, readFile would return C_current
3028 // but we'd store it with the client's M_observed — getChangedFiles
3029 // then sees disk > cache.timestamp, re-reads, diffs C_current vs
3030 // C_current = empty, emits no attachment, and the model is never
3031 // told about the C_observed → C_current change. Skipping the seed
3032 // makes Edit fail "file not read yet" → forces a fresh Read.
3033 // Math.floor matches FileReadTool and getFileModificationTime.
3034 const diskMtime = Math.floor((await stat(normalizedPath)).mtimeMs)
3035 if (diskMtime <= message.request.mtime) {
3036 const raw = await readFile(normalizedPath, 'utf-8')
3037 // Strip BOM + normalize CRLF→LF to match readFileInRange and
3038 // readFileSyncWithMetadata. FileEditTool's content-compare
3039 // fallback (for Windows mtime bumps without content change)
3040 // compares against LF-normalized disk reads.
3041 const content = (
3042 raw.charCodeAt(0) === 0xfeff ? raw.slice(1) : raw
3043 ).replaceAll('\r\n', '\n')
3044 pendingSeeds.set(normalizedPath, {
3045 content,
3046 timestamp: diskMtime,
3047 offset: undefined,
3048 limit: undefined,
3049 })
3050 }
3051 } catch {
3052 // ENOENT etc — skip seeding but still succeed
3053 }
3054 sendControlResponseSuccess(message)
3055 } else if (message.request.subtype === 'mcp_set_servers') {
3056 const { response, sdkServersChanged } = await applyMcpServerChanges(
3057 message.request.servers,
3058 )
3059 sendControlResponseSuccess(message, response)
3060
3061 // Connect SDK servers AFTER response to avoid deadlock
3062 if (sdkServersChanged) {
3063 void updateSdkMcp()
3064 }
3065 } else if (message.request.subtype === 'reload_plugins') {
3066 try {
3067 if (
3068 feature('DOWNLOAD_USER_SETTINGS') &&
3069 (isEnvTruthy(process.env.CLAUDE_CODE_REMOTE) || getIsRemoteMode())
3070 ) {
3071 // Re-pull user settings so enabledPlugins pushed from the
3072 // user's local CLI take effect before the cache sweep.
3073 const applied = await redownloadUserSettings()
3074 if (applied) {
3075 settingsChangeDetector.notifyChange('userSettings')
3076 }
3077 }
3078
3079 const r = await refreshActivePlugins(setAppState)
3080
3081 const sdkAgents = currentAgents.filter(
3082 a => a.source === 'flagSettings',
3083 )
3084 currentAgents = [...r.agentDefinitions.allAgents, ...sdkAgents]
3085
3086 // Reload succeeded — gather response data best-effort so a
3087 // read failure doesn't mask the successful state change.
3088 // allSettled so one failure doesn't discard the others.
3089 let plugins: SDKControlReloadPluginsResponse['plugins'] = []
3090 const [cmdsR, mcpR, pluginsR] = await Promise.allSettled([
3091 getCommands(cwd()),
3092 applyPluginMcpDiff(),
3093 loadAllPluginsCacheOnly(),
3094 ])
3095 if (cmdsR.status === 'fulfilled') {
3096 currentCommands = cmdsR.value
3097 } else {
3098 logError(cmdsR.reason)
3099 }
3100 if (mcpR.status === 'rejected') {
3101 logError(mcpR.reason)
3102 }
3103 if (pluginsR.status === 'fulfilled') {
3104 plugins = pluginsR.value.enabled.map(p => ({
3105 name: p.name,
3106 path: p.path,
3107 source: p.source,
3108 }))
3109 } else {
3110 logError(pluginsR.reason)
3111 }
3112
3113 sendControlResponseSuccess(message, {
3114 commands: currentCommands
3115 .filter(cmd => cmd.userInvocable !== false)
3116 .map(cmd => ({
3117 name: getCommandName(cmd),
3118 description: formatDescriptionWithSource(cmd),
3119 argumentHint: cmd.argumentHint || '',
3120 })),
3121 agents: currentAgents.map(a => ({
3122 name: a.agentType,
3123 description: a.whenToUse,
3124 model: a.model === 'inherit' ? undefined : a.model,
3125 })),
3126 plugins,
3127 mcpServers: buildMcpServerStatuses(),
3128 error_count: r.error_count,
3129 } satisfies SDKControlReloadPluginsResponse)
3130 } catch (error) {
3131 sendControlResponseError(message, errorMessage(error))
3132 }
3133 } else if (message.request.subtype === 'mcp_reconnect') {
3134 const currentAppState = getAppState()
3135 const { serverName } = message.request
3136 elicitationRegistered.delete(serverName)
3137 // Config-existence gate must cover the SAME sources as the
3138 // operations below. SDK-injected servers (query({mcpServers:{...}}))
3139 // and dynamically-added servers were missing here, so
3140 // toggleMcpServer/reconnect returned "Server not found" even though
3141 // the disconnect/reconnect would have worked (gh-31339 / CC-314).
3142 const config =
3143 getMcpConfigByName(serverName) ??
3144 mcpClients.find(c => c.name === serverName)?.config ??
3145 sdkClients.find(c => c.name === serverName)?.config ??
3146 dynamicMcpState.clients.find(c => c.name === serverName)?.config ??
3147 currentAppState.mcp.clients.find(c => c.name === serverName)
3148 ?.config ??
3149 null
3150 if (!config) {
3151 sendControlResponseError(message, `Server not found: ${serverName}`)
3152 } else {
3153 const result = await reconnectMcpServerImpl(serverName, config)
3154 // Update appState.mcp with the new client, tools, commands, and resources
3155 const prefix = getMcpPrefix(serverName)
3156 setAppState(prev => ({
3157 ...prev,
3158 mcp: {
3159 ...prev.mcp,
3160 clients: prev.mcp.clients.map(c =>
3161 c.name === serverName ? result.client : c,
3162 ),
3163 tools: [
3164 ...reject(prev.mcp.tools, t => t.name?.startsWith(prefix)),
3165 ...result.tools,
3166 ],
3167 commands: [
3168 ...reject(prev.mcp.commands, c =>
3169 commandBelongsToServer(c, serverName),
3170 ),
3171 ...result.commands,
3172 ],
3173 resources:
3174 result.resources && result.resources.length > 0
3175 ? { ...prev.mcp.resources, [serverName]: result.resources }
3176 : omit(prev.mcp.resources, serverName),
3177 },
3178 }))
3179 // Also update dynamicMcpState so run() picks up the new tools
3180 // on the next turn (run() reads dynamicMcpState, not appState)
3181 dynamicMcpState = {
3182 ...dynamicMcpState,
3183 clients: [
3184 ...dynamicMcpState.clients.filter(c => c.name !== serverName),
3185 result.client,
3186 ],
3187 tools: [
3188 ...dynamicMcpState.tools.filter(
3189 t => !t.name?.startsWith(prefix),
3190 ),
3191 ...result.tools,
3192 ],
3193 }
3194 if (result.client.type === 'connected') {
3195 registerElicitationHandlers([result.client])
3196 reregisterChannelHandlerAfterReconnect(result.client)
3197 sendControlResponseSuccess(message)
3198 } else {
3199 const errorMessage =
3200 result.client.type === 'failed'
3201 ? (result.client.error ?? 'Connection failed')
3202 : `Server status: ${result.client.type}`
3203 sendControlResponseError(message, errorMessage)
3204 }
3205 }
3206 } else if (message.request.subtype === 'mcp_toggle') {
3207 const currentAppState = getAppState()
3208 const { serverName, enabled } = message.request
3209 elicitationRegistered.delete(serverName)
3210 // Gate must match the client-lookup spread below (which
3211 // includes sdkClients and dynamicMcpState.clients). Same fix as
3212 // mcp_reconnect above (gh-31339 / CC-314).
3213 const config =
3214 getMcpConfigByName(serverName) ??
3215 mcpClients.find(c => c.name === serverName)?.config ??
3216 sdkClients.find(c => c.name === serverName)?.config ??
3217 dynamicMcpState.clients.find(c => c.name === serverName)?.config ??
3218 currentAppState.mcp.clients.find(c => c.name === serverName)
3219 ?.config ??
3220 null
3221
3222 if (!config) {
3223 sendControlResponseError(message, `Server not found: ${serverName}`)
3224 } else if (!enabled) {
3225 // Disabling: persist + disconnect (matches TUI toggleMcpServer behavior)
3226 setMcpServerEnabled(serverName, false)
3227 const client = [
3228 ...mcpClients,
3229 ...sdkClients,
3230 ...dynamicMcpState.clients,
3231 ...currentAppState.mcp.clients,
3232 ].find(c => c.name === serverName)
3233 if (client && client.type === 'connected') {
3234 await clearServerCache(serverName, config)
3235 }
3236 // Update appState.mcp to reflect disabled status and remove tools/commands/resources
3237 const prefix = getMcpPrefix(serverName)
3238 setAppState(prev => ({
3239 ...prev,
3240 mcp: {
3241 ...prev.mcp,
3242 clients: prev.mcp.clients.map(c =>
3243 c.name === serverName
3244 ? { name: serverName, type: 'disabled' as const, config }
3245 : c,
3246 ),
3247 tools: reject(prev.mcp.tools, t => t.name?.startsWith(prefix)),
3248 commands: reject(prev.mcp.commands, c =>
3249 commandBelongsToServer(c, serverName),
3250 ),
3251 resources: omit(prev.mcp.resources, serverName),
3252 },
3253 }))
3254 sendControlResponseSuccess(message)
3255 } else {
3256 // Enabling: persist + reconnect
3257 setMcpServerEnabled(serverName, true)
3258 const result = await reconnectMcpServerImpl(serverName, config)
3259 // Update appState.mcp with the new client, tools, commands, and resources
3260 // This ensures the LLM sees updated tools after enabling the server
3261 const prefix = getMcpPrefix(serverName)
3262 setAppState(prev => ({
3263 ...prev,
3264 mcp: {
3265 ...prev.mcp,
3266 clients: prev.mcp.clients.map(c =>
3267 c.name === serverName ? result.client : c,
3268 ),
3269 tools: [
3270 ...reject(prev.mcp.tools, t => t.name?.startsWith(prefix)),
3271 ...result.tools,
3272 ],
3273 commands: [
3274 ...reject(prev.mcp.commands, c =>
3275 commandBelongsToServer(c, serverName),
3276 ),
3277 ...result.commands,
3278 ],
3279 resources:
3280 result.resources && result.resources.length > 0
3281 ? { ...prev.mcp.resources, [serverName]: result.resources }
3282 : omit(prev.mcp.resources, serverName),
3283 },
3284 }))
3285 if (result.client.type === 'connected') {
3286 registerElicitationHandlers([result.client])
3287 reregisterChannelHandlerAfterReconnect(result.client)
3288 sendControlResponseSuccess(message)
3289 } else {
3290 const errorMessage =
3291 result.client.type === 'failed'
3292 ? (result.client.error ?? 'Connection failed')
3293 : `Server status: ${result.client.type}`
3294 sendControlResponseError(message, errorMessage)
3295 }
3296 }
3297 } else if (message.request.subtype === 'channel_enable') {
3298 const currentAppState = getAppState()
3299 handleChannelEnable(
3300 message.request_id,
3301 message.request.serverName,
3302 // Pool spread matches mcp_status — all three client sources.
3303 [
3304 ...currentAppState.mcp.clients,
3305 ...sdkClients,
3306 ...dynamicMcpState.clients,
3307 ],
3308 output,
3309 )
3310 } else if (message.request.subtype === 'mcp_authenticate') {
3311 const { serverName } = message.request
3312 const currentAppState = getAppState()
3313 const config =
3314 getMcpConfigByName(serverName) ??
3315 mcpClients.find(c => c.name === serverName)?.config ??
3316 currentAppState.mcp.clients.find(c => c.name === serverName)
3317 ?.config ??
3318 null
3319 if (!config) {
3320 sendControlResponseError(message, `Server not found: ${serverName}`)
3321 } else if (config.type !== 'sse' && config.type !== 'http') {
3322 sendControlResponseError(
3323 message,
3324 `Server type "${config.type}" does not support OAuth authentication`,
3325 )
3326 } else {
3327 try {
3328 // Abort any previous in-flight OAuth flow for this server
3329 activeOAuthFlows.get(serverName)?.abort()
3330 const controller = new AbortController()
3331 activeOAuthFlows.set(serverName, controller)
3332
3333 // Capture the auth URL from the callback
3334 let resolveAuthUrl: (url: string) => void
3335 const authUrlPromise = new Promise<string>(resolve => {
3336 resolveAuthUrl = resolve
3337 })
3338
3339 // Start the OAuth flow in the background
3340 const oauthPromise = performMCPOAuthFlow(
3341 serverName,
3342 config,
3343 url => resolveAuthUrl!(url),
3344 controller.signal,
3345 {
3346 skipBrowserOpen: true,
3347 onWaitingForCallback: submit => {
3348 oauthCallbackSubmitters.set(serverName, submit)
3349 },
3350 },
3351 )
3352
3353 // Wait for the auth URL (or the flow to complete without needing redirect)
3354 const authUrl = await Promise.race([
3355 authUrlPromise,
3356 oauthPromise.then(() => null as string | null),
3357 ])
3358
3359 if (authUrl) {
3360 sendControlResponseSuccess(message, {
3361 authUrl,
3362 requiresUserAction: true,
3363 })
3364 } else {
3365 sendControlResponseSuccess(message, {
3366 requiresUserAction: false,
3367 })
3368 }
3369
3370 // Store auth-only promise for mcp_oauth_callback_url handler.
3371 // Don't swallow errors — the callback handler needs to detect
3372 // auth failures and report them to the caller.
3373 oauthAuthPromises.set(serverName, oauthPromise)
3374
3375 // Handle background completion — reconnect after auth.
3376 // When manual callback is used, skip the reconnect here;
3377 // the extension's handleAuthDone → mcp_reconnect handles it
3378 // (which also updates dynamicMcpState for tool registration).
3379 const fullFlowPromise = oauthPromise
3380 .then(async () => {
3381 // Don't reconnect if the server was disabled during the OAuth flow
3382 if (isMcpServerDisabled(serverName)) {
3383 return
3384 }
3385 // Skip reconnect if the manual callback path was used —
3386 // handleAuthDone will do it via mcp_reconnect (which
3387 // updates dynamicMcpState for tool registration).
3388 if (oauthManualCallbackUsed.has(serverName)) {
3389 return
3390 }
3391 // Reconnect the server after successful auth
3392 const result = await reconnectMcpServerImpl(
3393 serverName,
3394 config,
3395 )
3396 const prefix = getMcpPrefix(serverName)
3397 setAppState(prev => ({
3398 ...prev,
3399 mcp: {
3400 ...prev.mcp,
3401 clients: prev.mcp.clients.map(c =>
3402 c.name === serverName ? result.client : c,
3403 ),
3404 tools: [
3405 ...reject(prev.mcp.tools, t =>
3406 t.name?.startsWith(prefix),
3407 ),
3408 ...result.tools,
3409 ],
3410 commands: [
3411 ...reject(prev.mcp.commands, c =>
3412 commandBelongsToServer(c, serverName),
3413 ),
3414 ...result.commands,
3415 ],
3416 resources:
3417 result.resources && result.resources.length > 0
3418 ? {
3419 ...prev.mcp.resources,
3420 [serverName]: result.resources,
3421 }
3422 : omit(prev.mcp.resources, serverName),
3423 },
3424 }))
3425 // Also update dynamicMcpState so run() picks up the new tools
3426 // on the next turn (run() reads dynamicMcpState, not appState)
3427 dynamicMcpState = {
3428 ...dynamicMcpState,
3429 clients: [
3430 ...dynamicMcpState.clients.filter(
3431 c => c.name !== serverName,
3432 ),
3433 result.client,
3434 ],
3435 tools: [
3436 ...dynamicMcpState.tools.filter(
3437 t => !t.name?.startsWith(prefix),
3438 ),
3439 ...result.tools,
3440 ],
3441 }
3442 })
3443 .catch(error => {
3444 logForDebugging(
3445 `MCP OAuth failed for ${serverName}: ${error}`,
3446 { level: 'error' },
3447 )
3448 })
3449 .finally(() => {
3450 // Clean up only if this is still the active flow
3451 if (activeOAuthFlows.get(serverName) === controller) {
3452 activeOAuthFlows.delete(serverName)
3453 oauthCallbackSubmitters.delete(serverName)
3454 oauthManualCallbackUsed.delete(serverName)
3455 oauthAuthPromises.delete(serverName)
3456 }
3457 })
3458 void fullFlowPromise
3459 } catch (error) {
3460 sendControlResponseError(message, errorMessage(error))
3461 }
3462 }
3463 } else if (message.request.subtype === 'mcp_oauth_callback_url') {
3464 const { serverName, callbackUrl } = message.request
3465 const submit = oauthCallbackSubmitters.get(serverName)
3466 if (submit) {
3467 // Validate the callback URL before submitting. The submit
3468 // callback in auth.ts silently ignores URLs missing a code
3469 // param, which would leave the auth promise unresolved and
3470 // block the control message loop until timeout.
3471 let hasCodeOrError = false
3472 try {
3473 const parsed = new URL(callbackUrl)
3474 hasCodeOrError =
3475 parsed.searchParams.has('code') ||
3476 parsed.searchParams.has('error')
3477 } catch {
3478 // Invalid URL
3479 }
3480 if (!hasCodeOrError) {
3481 sendControlResponseError(
3482 message,
3483 'Invalid callback URL: missing authorization code. Please paste the full redirect URL including the code parameter.',
3484 )
3485 } else {
3486 oauthManualCallbackUsed.add(serverName)
3487 submit(callbackUrl)
3488 // Wait for auth (token exchange) to complete before responding.
3489 // Reconnect is handled by the extension via handleAuthDone →
3490 // mcp_reconnect (which updates dynamicMcpState for tools).
3491 const authPromise = oauthAuthPromises.get(serverName)
3492 if (authPromise) {
3493 try {
3494 await authPromise
3495 sendControlResponseSuccess(message)
3496 } catch (error) {
3497 sendControlResponseError(
3498 message,
3499 error instanceof Error
3500 ? error.message
3501 : 'OAuth authentication failed',
3502 )
3503 }
3504 } else {
3505 sendControlResponseSuccess(message)
3506 }
3507 }
3508 } else {
3509 sendControlResponseError(
3510 message,
3511 `No active OAuth flow for server: ${serverName}`,
3512 )
3513 }
3514 } else if (message.request.subtype === 'claude_authenticate') {
3515 // Anthropic OAuth over the control channel. The SDK client owns
3516 // the user's browser (we're headless in -p mode); we hand back
3517 // both URLs and wait. Automatic URL → localhost listener catches
3518 // the redirect if the browser is on this host; manual URL → the
3519 // success page shows "code#state" for claude_oauth_callback.
3520 const { loginWithClaudeAi } = message.request
3521
3522 // Clean up any prior flow. cleanup() closes the localhost listener
3523 // and nulls the manual resolver. The prior `flow` promise is left
3524 // pending (AuthCodeListener.close() does not reject) but its object
3525 // graph becomes unreachable once the server handle is released and
3526 // is GC'd — no fd or port is held.
3527 claudeOAuth?.service.cleanup()
3528
3529 logEvent('tengu_oauth_flow_start', {
3530 loginWithClaudeAi: loginWithClaudeAi ?? true,
3531 })
3532
3533 const service = new OAuthService()
3534 let urlResolver!: (urls: {
3535 manualUrl: string
3536 automaticUrl: string
3537 }) => void
3538 const urlPromise = new Promise<{
3539 manualUrl: string
3540 automaticUrl: string
3541 }>(resolve => {
3542 urlResolver = resolve
3543 })
3544
3545 const flow = service
3546 .startOAuthFlow(
3547 async (manualUrl, automaticUrl) => {
3548 // automaticUrl is always defined when skipBrowserOpen is set;
3549 // the signature is optional only for the existing single-arg callers.
3550 urlResolver({ manualUrl, automaticUrl: automaticUrl! })
3551 },
3552 {
3553 loginWithClaudeAi: loginWithClaudeAi ?? true,
3554 skipBrowserOpen: true,
3555 },
3556 )
3557 .then(async tokens => {
3558 // installOAuthTokens: performLogout (clear stale state) →
3559 // store profile → saveOAuthTokensIfNeeded → clearOAuthTokenCache
3560 // → clearAuthRelatedCaches. After this resolves, the memoized
3561 // getClaudeAIOAuthTokens in this process is invalidated; the
3562 // next API call re-reads keychain/file and works. No respawn.
3563 await installOAuthTokens(tokens)
3564 logEvent('tengu_oauth_success', {
3565 loginWithClaudeAi: loginWithClaudeAi ?? true,
3566 })
3567 })
3568 .finally(() => {
3569 service.cleanup()
3570 if (claudeOAuth?.service === service) {
3571 claudeOAuth = null
3572 }
3573 })
3574
3575 claudeOAuth = { service, flow }
3576
3577 // Attach the rejection handler before awaiting so a synchronous
3578 // startOAuthFlow failure doesn't surface as an unhandled rejection.
3579 // The claude_oauth_callback handler re-awaits flow for the manual
3580 // path and surfaces the real error to the client.
3581 void flow.catch(err =>
3582 logForDebugging(`claude_authenticate flow ended: ${err}`, {
3583 level: 'info',
3584 }),
3585 )
3586
3587 try {
3588 // Race against flow: if startOAuthFlow rejects before calling
3589 // the authURLHandler (e.g. AuthCodeListener.start() fails with
3590 // EACCES or fd exhaustion), urlPromise would pend forever and
3591 // wedge the stdin loop. flow resolving first is unreachable in
3592 // practice (it's suspended on the same urls we're waiting for).
3593 const { manualUrl, automaticUrl } = await Promise.race([
3594 urlPromise,
3595 flow.then(() => {
3596 throw new Error(
3597 'OAuth flow completed without producing auth URLs',
3598 )
3599 }),
3600 ])
3601 sendControlResponseSuccess(message, {
3602 manualUrl,
3603 automaticUrl,
3604 })
3605 } catch (error) {
3606 sendControlResponseError(message, errorMessage(error))
3607 }
3608 } else if (
3609 message.request.subtype === 'claude_oauth_callback' ||
3610 message.request.subtype === 'claude_oauth_wait_for_completion'
3611 ) {
3612 if (!claudeOAuth) {
3613 sendControlResponseError(
3614 message,
3615 'No active claude_authenticate flow',
3616 )
3617 } else {
3618 // Inject the manual code synchronously — must happen in stdin
3619 // message order so a subsequent claude_authenticate doesn't
3620 // replace the service before this code lands.
3621 if (message.request.subtype === 'claude_oauth_callback') {
3622 claudeOAuth.service.handleManualAuthCodeInput({
3623 authorizationCode: message.request.authorizationCode,
3624 state: message.request.state,
3625 })
3626 }
3627 // Detach the await — the stdin reader is serial and blocking
3628 // here deadlocks claude_oauth_wait_for_completion: flow may
3629 // only resolve via a future claude_oauth_callback on stdin,
3630 // which can't be read while we're parked. Capture the binding;
3631 // claudeOAuth is nulled in flow's own .finally.
3632 const { flow } = claudeOAuth
3633 void flow.then(
3634 () => {
3635 const accountInfo = getAccountInformation()
3636 sendControlResponseSuccess(message, {
3637 account: {
3638 email: accountInfo?.email,
3639 organization: accountInfo?.organization,
3640 subscriptionType: accountInfo?.subscription,
3641 tokenSource: accountInfo?.tokenSource,
3642 apiKeySource: accountInfo?.apiKeySource,
3643 apiProvider: getAPIProvider(),
3644 },
3645 })
3646 },
3647 (error: unknown) =>
3648 sendControlResponseError(message, errorMessage(error)),
3649 )
3650 }
3651 } else if (message.request.subtype === 'mcp_clear_auth') {
3652 const { serverName } = message.request
3653 const currentAppState = getAppState()
3654 const config =
3655 getMcpConfigByName(serverName) ??
3656 mcpClients.find(c => c.name === serverName)?.config ??
3657 currentAppState.mcp.clients.find(c => c.name === serverName)
3658 ?.config ??
3659 null
3660 if (!config) {
3661 sendControlResponseError(message, `Server not found: ${serverName}`)
3662 } else if (config.type !== 'sse' && config.type !== 'http') {
3663 sendControlResponseError(
3664 message,
3665 `Cannot clear auth for server type "${config.type}"`,
3666 )
3667 } else {
3668 await revokeServerTokens(serverName, config)
3669 const result = await reconnectMcpServerImpl(serverName, config)
3670 const prefix = getMcpPrefix(serverName)
3671 setAppState(prev => ({
3672 ...prev,
3673 mcp: {
3674 ...prev.mcp,
3675 clients: prev.mcp.clients.map(c =>
3676 c.name === serverName ? result.client : c,
3677 ),
3678 tools: [
3679 ...reject(prev.mcp.tools, t => t.name?.startsWith(prefix)),
3680 ...result.tools,
3681 ],
3682 commands: [
3683 ...reject(prev.mcp.commands, c =>
3684 commandBelongsToServer(c, serverName),
3685 ),
3686 ...result.commands,
3687 ],
3688 resources:
3689 result.resources && result.resources.length > 0
3690 ? {
3691 ...prev.mcp.resources,
3692 [serverName]: result.resources,
3693 }
3694 : omit(prev.mcp.resources, serverName),
3695 },
3696 }))
3697 sendControlResponseSuccess(message, {})
3698 }
3699 } else if (message.request.subtype === 'apply_flag_settings') {
3700 // Snapshot the current model before applying — we need to detect
3701 // model switches so we can inject breadcrumbs and notify listeners.
3702 const prevModel = getMainLoopModel()
3703
3704 // Merge the provided settings into the in-memory flag settings
3705 const existing = getFlagSettingsInline() ?? {}
3706 const incoming = message.request.settings
3707 // Shallow-merge top-level keys; getSettingsForSource handles
3708 // the deep merge with file-based flag settings via mergeWith.
3709 // JSON serialization drops `undefined`, so callers use `null`
3710 // to signal "clear this key". Convert nulls to deletions so
3711 // SettingsSchema().safeParse() doesn't reject the whole object
3712 // (z.string().optional() accepts string | undefined, not null).
3713 const merged = { ...existing, ...incoming }
3714 for (const key of Object.keys(merged)) {
3715 if (merged[key as keyof typeof merged] === null) {
3716 delete merged[key as keyof typeof merged]
3717 }
3718 }
3719 setFlagSettingsInline(merged)
3720 // Route through notifyChange so fanOut() resets the settings cache
3721 // before listeners run. The subscriber at :392 calls
3722 // applySettingsChange for us. Pre-#20625 this was a direct
3723 // applySettingsChange() call that relied on its own internal reset —
3724 // now that the reset is centralized in fanOut, a direct call here
3725 // would read stale cached settings and silently drop the update.
3726 // Bonus: going through notifyChange also tells the other subscribers
3727 // (loadPluginHooks, sandbox-adapter) about the change, which the
3728 // previous direct call skipped.
3729 settingsChangeDetector.notifyChange('flagSettings')
3730
3731 // If the incoming settings include a model change, update the
3732 // override so getMainLoopModel() reflects it. The override has
3733 // higher priority than the settings cascade in
3734 // getUserSpecifiedModelSetting(), so without this update,
3735 // getMainLoopModel() returns the stale override and the model
3736 // change is silently ignored (matching set_model at :2811).
3737 if ('model' in incoming) {
3738 if (incoming.model != null) {
3739 setMainLoopModelOverride(String(incoming.model))
3740 } else {
3741 setMainLoopModelOverride(undefined)
3742 }
3743 }
3744
3745 // If the model changed, inject breadcrumbs so the model sees the
3746 // mid-conversation switch, and notify metadata listeners (CCR).
3747 const newModel = getMainLoopModel()
3748 if (newModel !== prevModel) {
3749 activeUserSpecifiedModel = newModel
3750 const modelArg = incoming.model ? String(incoming.model) : 'default'
3751 notifySessionMetadataChanged({ model: newModel })
3752 injectModelSwitchBreadcrumbs(modelArg, newModel)
3753 }
3754
3755 sendControlResponseSuccess(message)
3756 } else if (message.request.subtype === 'get_settings') {
3757 const currentAppState = getAppState()
3758 const model = getMainLoopModel()
3759 // modelSupportsEffort gate matches claude.ts — applied.effort must
3760 // mirror what actually goes to the API, not just what's configured.
3761 const effort = modelSupportsEffort(model)
3762 ? resolveAppliedEffort(model, currentAppState.effortValue)
3763 : undefined
3764 sendControlResponseSuccess(message, {
3765 ...getSettingsWithSources(),
3766 applied: {
3767 model,
3768 // Numeric effort (ant-only) → null; SDK schema is string-level only.
3769 effort: typeof effort === 'string' ? effort : null,
3770 },
3771 })
3772 } else if (message.request.subtype === 'stop_task') {
3773 const { task_id: taskId } = message.request
3774 try {
3775 await stopTask(taskId, {
3776 getAppState,
3777 setAppState,
3778 })
3779 sendControlResponseSuccess(message, {})
3780 } catch (error) {
3781 sendControlResponseError(message, errorMessage(error))
3782 }
3783 } else if (message.request.subtype === 'generate_session_title') {
3784 // Fire-and-forget so the Haiku call does not block the stdin loop
3785 // (which would delay processing of subsequent user messages /
3786 // interrupts for the duration of the API roundtrip).
3787 const { description, persist } = message.request
3788 // Reuse the live controller only if it has not already been aborted
3789 // (e.g. by interrupt()); an aborted signal would cause queryHaiku to
3790 // immediately throw APIUserAbortError → {title: null}.
3791 const titleSignal = (
3792 abortController && !abortController.signal.aborted
3793 ? abortController
3794 : createAbortController()
3795 ).signal
3796 void (async () => {
3797 try {
3798 const title = await generateSessionTitle(description, titleSignal)
3799 if (title && persist) {
3800 try {
3801 saveAiGeneratedTitle(getSessionId() as UUID, title)
3802 } catch (e) {
3803 logError(e)
3804 }
3805 }
3806 sendControlResponseSuccess(message, { title })
3807 } catch (e) {
3808 // Unreachable in practice — generateSessionTitle wraps its
3809 // own body and returns null, saveAiGeneratedTitle is wrapped
3810 // above. Propagate (not swallow) so unexpected failures are
3811 // visible to the SDK caller (hostComms.ts catches and logs).
3812 sendControlResponseError(message, errorMessage(e))
3813 }
3814 })()
3815 } else if (message.request.subtype === 'side_question') {
3816 // Same fire-and-forget pattern as generate_session_title above —
3817 // the forked agent's API roundtrip must not block the stdin loop.
3818 //
3819 // The snapshot captured by stopHooks (for querySource === 'sdk')
3820 // holds the exact systemPrompt/userContext/systemContext/messages
3821 // sent on the last main-thread turn. Reusing them gives a byte-
3822 // identical prefix → prompt cache hit.
3823 //
3824 // Fallback (resume before first turn completes — no snapshot yet):
3825 // rebuild from scratch. buildSideQuestionFallbackParams mirrors
3826 // QueryEngine.ts:ask()'s system prompt assembly (including
3827 // --system-prompt / --append-system-prompt) so the rebuilt prefix
3828 // matches in the common case. May still miss the cache for
3829 // coordinator mode or memory-mechanics extras — acceptable, the
3830 // alternative is the side question failing entirely.
3831 const { question } = message.request
3832 void (async () => {
3833 try {
3834 const saved = getLastCacheSafeParams()
3835 const cacheSafeParams = saved
3836 ? {
3837 ...saved,
3838 // If the last turn was interrupted, the snapshot holds an
3839 // already-aborted controller; createChildAbortController in
3840 // createSubagentContext would propagate it and the fork
3841 // would die before sending a request. The controller is
3842 // not part of the cache key — swapping in a fresh one is
3843 // safe. Same guard as generate_session_title above.
3844 toolUseContext: {
3845 ...saved.toolUseContext,
3846 abortController: createAbortController(),
3847 },
3848 }
3849 : await buildSideQuestionFallbackParams({
3850 tools: buildAllTools(getAppState()),
3851 commands: currentCommands,
3852 mcpClients: [
3853 ...getAppState().mcp.clients,
3854 ...sdkClients,
3855 ...dynamicMcpState.clients,
3856 ],
3857 messages: mutableMessages,
3858 readFileState,
3859 getAppState,
3860 setAppState,
3861 customSystemPrompt: options.systemPrompt,
3862 appendSystemPrompt: options.appendSystemPrompt,
3863 thinkingConfig: options.thinkingConfig,
3864 agents: currentAgents,
3865 })
3866 const result = await runSideQuestion({
3867 question,
3868 cacheSafeParams,
3869 })
3870 sendControlResponseSuccess(message, { response: result.response })
3871 } catch (e) {
3872 sendControlResponseError(message, errorMessage(e))
3873 }
3874 })()
3875 } else if (
3876 (feature('PROACTIVE') || feature('KAIROS')) &&
3877 (message.request as { subtype: string }).subtype === 'set_proactive'
3878 ) {
3879 const req = message.request as unknown as {
3880 subtype: string
3881 enabled: boolean
3882 }
3883 if (req.enabled) {
3884 if (!proactiveModule!.isProactiveActive()) {
3885 proactiveModule!.activateProactive('command')
3886 scheduleProactiveTick!()
3887 }
3888 } else {
3889 proactiveModule!.deactivateProactive()
3890 }
3891 sendControlResponseSuccess(message)
3892 } else if (message.request.subtype === 'remote_control') {
3893 if (message.request.enabled) {
3894 if (bridgeHandle) {
3895 // Already connected
3896 sendControlResponseSuccess(message, {
3897 session_url: getRemoteSessionUrl(
3898 bridgeHandle.bridgeSessionId,
3899 bridgeHandle.sessionIngressUrl,
3900 ),
3901 connect_url: buildBridgeConnectUrl(
3902 bridgeHandle.environmentId,
3903 bridgeHandle.sessionIngressUrl,
3904 ),
3905 environment_id: bridgeHandle.environmentId,
3906 })
3907 } else {
3908 // initReplBridge surfaces gate-failure reasons via
3909 // onStateChange('failed', detail) before returning null.
3910 // Capture so the control-response error is actionable
3911 // ("/login", "disabled by your organization's policy", etc.)
3912 // instead of a generic "initialization failed".
3913 let bridgeFailureDetail: string | undefined
3914 try {
3915 const { initReplBridge } = await import(
3916 'src/bridge/initReplBridge.js'
3917 )
3918 const handle = await initReplBridge({
3919 onInboundMessage(msg) {
3920 const fields = extractInboundMessageFields(msg)
3921 if (!fields) return
3922 const { content, uuid } = fields
3923 enqueue({
3924 value: content,
3925 mode: 'prompt' as const,
3926 uuid,
3927 skipSlashCommands: true,
3928 })
3929 void run()
3930 },
3931 onPermissionResponse(response) {
3932 // Forward bridge permission responses into the
3933 // stdin processing loop so they resolve pending
3934 // permission requests from the SDK consumer.
3935 structuredIO.injectControlResponse(response)
3936 },
3937 onInterrupt() {
3938 abortController?.abort()
3939 },
3940 onSetModel(model) {
3941 const resolved =
3942 model === 'default' ? getDefaultMainLoopModel() : model
3943 activeUserSpecifiedModel = resolved
3944 setMainLoopModelOverride(resolved)
3945 },
3946 onSetMaxThinkingTokens(maxTokens) {
3947 if (maxTokens === null) {
3948 options.thinkingConfig = undefined
3949 } else if (maxTokens === 0) {
3950 options.thinkingConfig = { type: 'disabled' }
3951 } else {
3952 options.thinkingConfig = {
3953 type: 'enabled',
3954 budgetTokens: maxTokens,
3955 }
3956 }
3957 },
3958 onStateChange(state, detail) {
3959 if (state === 'failed') {
3960 bridgeFailureDetail = detail
3961 }
3962 logForDebugging(
3963 `[bridge:sdk] State change: ${state}${detail ? ` — ${detail}` : ''}`,
3964 )
3965 output.enqueue({
3966 type: 'system' as StdoutMessage['type'],
3967 subtype: 'bridge_state' as string,
3968 state,
3969 detail,
3970 uuid: randomUUID(),
3971 session_id: getSessionId(),
3972 } as StdoutMessage)
3973 },
3974 initialMessages:
3975 mutableMessages.length > 0 ? mutableMessages : undefined,
3976 })
3977 if (!handle) {
3978 sendControlResponseError(
3979 message,
3980 bridgeFailureDetail ??
3981 'Remote Control initialization failed',
3982 )
3983 } else {
3984 bridgeHandle = handle
3985 bridgeLastForwardedIndex = mutableMessages.length
3986 // Forward permission requests to the bridge
3987 structuredIO.setOnControlRequestSent(request => {
3988 handle.sendControlRequest(request)
3989 })
3990 // Cancel stale bridge permission prompts when the SDK
3991 // consumer resolves a can_use_tool request first.
3992 structuredIO.setOnControlRequestResolved(requestId => {
3993 handle.sendControlCancelRequest(requestId)
3994 })
3995 sendControlResponseSuccess(message, {
3996 session_url: getRemoteSessionUrl(
3997 handle.bridgeSessionId,
3998 handle.sessionIngressUrl,
3999 ),
4000 connect_url: buildBridgeConnectUrl(
4001 handle.environmentId,
4002 handle.sessionIngressUrl,
4003 ),
4004 environment_id: handle.environmentId,
4005 })
4006 }
4007 } catch (err) {
4008 sendControlResponseError(message, errorMessage(err))
4009 }
4010 }
4011 } else {
4012 // Disable
4013 if (bridgeHandle) {
4014 structuredIO.setOnControlRequestSent(undefined)
4015 structuredIO.setOnControlRequestResolved(undefined)
4016 await bridgeHandle.teardown()
4017 bridgeHandle = null
4018 }
4019 sendControlResponseSuccess(message)
4020 }
4021 } else {
4022 // Unknown control request subtype — send an error response so
4023 // the caller doesn't hang waiting for a reply that never comes.
4024 sendControlResponseError(
4025 message,
4026 `Unsupported control request subtype: ${(message.request as { subtype: string }).subtype}`,
4027 )
4028 }
4029 continue
4030 } else if (message.type === 'control_response') {
4031 // Replay control_response messages when replay mode is enabled
4032 if (options.replayUserMessages) {
4033 output.enqueue(message)
4034 }
4035 continue
4036 } else if (message.type === 'keep_alive') {
4037 // Silently ignore keep-alive messages
4038 continue
4039 } else if (message.type === 'update_environment_variables') {
4040 // Handled in structuredIO.ts, but TypeScript needs the type guard
4041 continue
4042 } else if (message.type === 'assistant' || message.type === 'system') {
4043 // History replay from bridge: inject into mutableMessages as
4044 // conversation context so the model sees prior turns.
4045 const internalMsgs = toInternalMessages([message])
4046 mutableMessages.push(...internalMsgs)
4047 // Echo assistant messages back so CCR displays them
4048 if (message.type === 'assistant' && options.replayUserMessages) {
4049 output.enqueue(message)
4050 }
4051 continue
4052 }
4053 // After handling control, keep-alive, env-var, assistant, and system
4054 // messages above, only user messages should remain.
4055 if (message.type !== 'user') {
4056 continue
4057 }
4058
4059 // First prompt message implicitly initializes if not already done.
4060 initialized = true
4061
4062 // Check for duplicate user message - skip if already processed
4063 if (message.uuid) {
4064 const sessionId = getSessionId() as UUID
4065 const existsInSession = await doesMessageExistInSession(
4066 sessionId,
4067 message.uuid,
4068 )
4069
4070 // Check both historical duplicates (from file) and runtime duplicates (this session)
4071 if (existsInSession || receivedMessageUuids.has(message.uuid)) {
4072 logForDebugging(`Skipping duplicate user message: ${message.uuid}`)
4073 // Send acknowledgment for duplicate message if replay mode is enabled
4074 if (options.replayUserMessages) {
4075 logForDebugging(
4076 `Sending acknowledgment for duplicate user message: ${message.uuid}`,
4077 )
4078 output.enqueue({
4079 type: 'user',
4080 message: message.message,
4081 session_id: sessionId,
4082 parent_tool_use_id: null,
4083 uuid: message.uuid,
4084 timestamp: message.timestamp,
4085 isReplay: true,
4086 } as SDKUserMessageReplay)
4087 }
4088 // Historical dup = transcript already has this turn's output, so it
4089 // ran but its lifecycle was never closed (interrupted before ack).
4090 // Runtime dups don't need this — the original enqueue path closes them.
4091 if (existsInSession) {
4092 notifyCommandLifecycle(message.uuid, 'completed')
4093 }
4094 // Don't enqueue duplicate messages for execution
4095 continue
4096 }
4097
4098 // Track this UUID to prevent runtime duplicates
4099 trackReceivedMessageUuid(message.uuid)
4100 }
4101
4102 enqueue({
4103 mode: 'prompt' as const,
4104 // file_attachments rides the protobuf catchall from the web composer.
4105 // Same-ref no-op when absent (no 'file_attachments' key).
4106 value: await resolveAndPrepend(message, message.message.content),
4107 uuid: message.uuid,
4108 priority: message.priority,
4109 })
4110 // Increment prompt count for attribution tracking and save snapshot
4111 // The snapshot persists promptCount so it survives compaction
4112 if (feature('COMMIT_ATTRIBUTION')) {
4113 setAppState(prev => ({
4114 ...prev,
4115 attribution: incrementPromptCount(prev.attribution, snapshot => {
4116 void recordAttributionSnapshot(snapshot).catch(error => {
4117 logForDebugging(`Attribution: Failed to save snapshot: ${error}`)
4118 })
4119 }),
4120 }))
4121 }
4122 void run()
4123 }
4124 inputClosed = true
4125 cronScheduler?.stop()
4126 if (!running) {
4127 // If a push-suggestion is in-flight, wait for it to emit before closing
4128 // the output stream (5 s safety timeout to prevent hanging).
4129 if (suggestionState.inflightPromise) {
4130 await Promise.race([suggestionState.inflightPromise, sleep(5000)])
4131 }
4132 suggestionState.abortController?.abort()
4133 suggestionState.abortController = null
4134 await finalizePendingAsyncHooks()
4135 unsubscribeSkillChanges()
4136 unsubscribeAuthStatus?.()
4137 statusListeners.delete(rateLimitListener)
4138 output.done()
4139 }
4140 })()
4141
4142 return output
4143}
4144
4145/**
4146 * Creates a CanUseToolFn that incorporates a custom permission prompt tool.
4147 * This function converts the permissionPromptTool into a CanUseToolFn that can be used in ask.tsx
4148 */
4149export function createCanUseToolWithPermissionPrompt(
4150 permissionPromptTool: PermissionPromptTool,
4151): CanUseToolFn {
4152 const canUseTool: CanUseToolFn = async (
4153 tool,
4154 input,
4155 toolUseContext,
4156 assistantMessage,
4157 toolUseId,
4158 forceDecision,
4159 ) => {
4160 const mainPermissionResult =
4161 forceDecision ??
4162 (await hasPermissionsToUseTool(
4163 tool,
4164 input,
4165 toolUseContext,
4166 assistantMessage,
4167 toolUseId,
4168 ))
4169
4170 // If the tool is allowed or denied, return the result
4171 if (
4172 mainPermissionResult.behavior === 'allow' ||
4173 mainPermissionResult.behavior === 'deny'
4174 ) {
4175 return mainPermissionResult
4176 }
4177
4178 // Race the permission prompt tool against the abort signal.
4179 //
4180 // Why we need this: The permission prompt tool may block indefinitely waiting
4181 // for user input (e.g., via stdin or a UI dialog). If the user triggers an
4182 // interrupt (Ctrl+C), we need to detect it even while the tool is blocked.
4183 // Without this race, the abort check would only run AFTER the tool completes,
4184 // which may never happen if the tool is waiting for input that will never come.
4185 //
4186 // The second check (combinedSignal.aborted) handles a race condition where
4187 // abort fires after Promise.race resolves but before we reach this check.
4188 const { signal: combinedSignal, cleanup: cleanupAbortListener } =
4189 createCombinedAbortSignal(toolUseContext.abortController.signal)
4190
4191 // Check if already aborted before starting the race
4192 if (combinedSignal.aborted) {
4193 cleanupAbortListener()
4194 return {
4195 behavior: 'deny',
4196 message: 'Permission prompt was aborted.',
4197 decisionReason: {
4198 type: 'permissionPromptTool' as const,
4199 permissionPromptToolName: tool.name,
4200 toolResult: undefined,
4201 },
4202 }
4203 }
4204
4205 const abortPromise = new Promise<'aborted'>(resolve => {
4206 combinedSignal.addEventListener('abort', () => resolve('aborted'), {
4207 once: true,
4208 })
4209 })
4210
4211 const toolCallPromise = permissionPromptTool.call(
4212 {
4213 tool_name: tool.name,
4214 input,
4215 tool_use_id: toolUseId,
4216 },
4217 toolUseContext,
4218 canUseTool,
4219 assistantMessage,
4220 )
4221
4222 const raceResult = await Promise.race([toolCallPromise, abortPromise])
4223 cleanupAbortListener()
4224
4225 if (raceResult === 'aborted' || combinedSignal.aborted) {
4226 return {
4227 behavior: 'deny',
4228 message: 'Permission prompt was aborted.',
4229 decisionReason: {
4230 type: 'permissionPromptTool' as const,
4231 permissionPromptToolName: tool.name,
4232 toolResult: undefined,
4233 },
4234 }
4235 }
4236
4237 // TypeScript narrowing: after the abort check, raceResult must be ToolResult
4238 const result = raceResult as Awaited<typeof toolCallPromise>
4239
4240 const permissionToolResultBlockParam =
4241 permissionPromptTool.mapToolResultToToolResultBlockParam(result.data, '1')
4242 if (
4243 !permissionToolResultBlockParam.content ||
4244 !Array.isArray(permissionToolResultBlockParam.content) ||
4245 !permissionToolResultBlockParam.content[0] ||
4246 permissionToolResultBlockParam.content[0].type !== 'text' ||
4247 typeof permissionToolResultBlockParam.content[0].text !== 'string'
4248 ) {
4249 throw new Error(
4250 'Permission prompt tool returned an invalid result. Expected a single text block param with type="text" and a string text value.',
4251 )
4252 }
4253 return permissionPromptToolResultToPermissionDecision(
4254 permissionToolOutputSchema().parse(
4255 safeParseJSON(permissionToolResultBlockParam.content[0].text),
4256 ),
4257 permissionPromptTool,
4258 input,
4259 toolUseContext,
4260 )
4261 }
4262 return canUseTool
4263}
4264
4265// Exported for testing — regression: this used to crash at construction when
4266// getMcpTools() was empty (before per-server connects populated appState).
4267export function getCanUseToolFn(
4268 permissionPromptToolName: string | undefined,
4269 structuredIO: StructuredIO,
4270 getMcpTools: () => Tool[],
4271 onPermissionPrompt?: (details: RequiresActionDetails) => void,
4272): CanUseToolFn {
4273 if (permissionPromptToolName === 'stdio') {
4274 return structuredIO.createCanUseTool(onPermissionPrompt)
4275 }
4276 if (!permissionPromptToolName) {
4277 return async (
4278 tool,
4279 input,
4280 toolUseContext,
4281 assistantMessage,
4282 toolUseId,
4283 forceDecision,
4284 ) =>
4285 forceDecision ??
4286 (await hasPermissionsToUseTool(
4287 tool,
4288 input,
4289 toolUseContext,
4290 assistantMessage,
4291 toolUseId,
4292 ))
4293 }
4294 // Lazy lookup: MCP connects are per-server incremental in print mode, so
4295 // the tool may not be in appState yet at init time. Resolve on first call
4296 // (first permission prompt), by which point connects have had time to finish.
4297 let resolved: CanUseToolFn | null = null
4298 return async (
4299 tool,
4300 input,
4301 toolUseContext,
4302 assistantMessage,
4303 toolUseId,
4304 forceDecision,
4305 ) => {
4306 if (!resolved) {
4307 const mcpTools = getMcpTools()
4308 const permissionPromptTool = mcpTools.find(t =>
4309 toolMatchesName(t, permissionPromptToolName),
4310 ) as PermissionPromptTool | undefined
4311 if (!permissionPromptTool) {
4312 const error = `Error: MCP tool ${permissionPromptToolName} (passed via --permission-prompt-tool) not found. Available MCP tools: ${mcpTools.map(t => t.name).join(', ') || 'none'}`
4313 process.stderr.write(`${error}\n`)
4314 gracefulShutdownSync(1)
4315 throw new Error(error)
4316 }
4317 if (!permissionPromptTool.inputJSONSchema) {
4318 const error = `Error: tool ${permissionPromptToolName} (passed via --permission-prompt-tool) must be an MCP tool`
4319 process.stderr.write(`${error}\n`)
4320 gracefulShutdownSync(1)
4321 throw new Error(error)
4322 }
4323 resolved = createCanUseToolWithPermissionPrompt(permissionPromptTool)
4324 }
4325 return resolved(
4326 tool,
4327 input,
4328 toolUseContext,
4329 assistantMessage,
4330 toolUseId,
4331 forceDecision,
4332 )
4333 }
4334}
4335
4336async function handleInitializeRequest(
4337 request: SDKControlInitializeRequest,
4338 requestId: string,
4339 initialized: boolean,
4340 output: Stream<StdoutMessage>,
4341 commands: Command[],
4342 modelInfos: ModelInfo[],
4343 structuredIO: StructuredIO,
4344 enableAuthStatus: boolean,
4345 options: {
4346 systemPrompt: string | undefined
4347 appendSystemPrompt: string | undefined
4348 agent?: string | undefined
4349 userSpecifiedModel?: string | undefined
4350 [key: string]: unknown
4351 },
4352 agents: AgentDefinition[],
4353 getAppState: () => AppState,
4354): Promise<void> {
4355 if (initialized) {
4356 output.enqueue({
4357 type: 'control_response',
4358 response: {
4359 subtype: 'error',
4360 error: 'Already initialized',
4361 request_id: requestId,
4362 pending_permission_requests:
4363 structuredIO.getPendingPermissionRequests(),
4364 },
4365 })
4366 return
4367 }
4368
4369 // Apply systemPrompt/appendSystemPrompt from stdin to avoid ARG_MAX limits
4370 if (request.systemPrompt !== undefined) {
4371 options.systemPrompt = request.systemPrompt
4372 }
4373 if (request.appendSystemPrompt !== undefined) {
4374 options.appendSystemPrompt = request.appendSystemPrompt
4375 }
4376 if (request.promptSuggestions !== undefined) {
4377 options.promptSuggestions = request.promptSuggestions
4378 }
4379
4380 // Merge agents from stdin to avoid ARG_MAX limits
4381 if (request.agents) {
4382 const stdinAgents = parseAgentsFromJson(request.agents, 'flagSettings')
4383 agents.push(...stdinAgents)
4384 }
4385
4386 // Re-evaluate main thread agent after SDK agents are merged
4387 // This allows --agent to reference agents defined via SDK
4388 if (options.agent) {
4389 // If main.tsx already found this agent (filesystem-defined), it already
4390 // applied systemPrompt/model/initialPrompt. Skip to avoid double-apply.
4391 const alreadyResolved = getMainThreadAgentType() === options.agent
4392 const mainThreadAgent = agents.find(a => a.agentType === options.agent)
4393 if (mainThreadAgent && !alreadyResolved) {
4394 // Update the main thread agent type in bootstrap state
4395 setMainThreadAgentType(mainThreadAgent.agentType)
4396
4397 // Apply the agent's system prompt if user hasn't specified a custom one
4398 // SDK agents are always custom agents (not built-in), so getSystemPrompt() takes no args
4399 if (!options.systemPrompt && !isBuiltInAgent(mainThreadAgent)) {
4400 const agentSystemPrompt = mainThreadAgent.getSystemPrompt()
4401 if (agentSystemPrompt) {
4402 options.systemPrompt = agentSystemPrompt
4403 }
4404 }
4405
4406 // Apply the agent's model if user didn't specify one and agent has a model
4407 if (
4408 !options.userSpecifiedModel &&
4409 mainThreadAgent.model &&
4410 mainThreadAgent.model !== 'inherit'
4411 ) {
4412 const agentModel = parseUserSpecifiedModel(mainThreadAgent.model)
4413 setMainLoopModelOverride(agentModel)
4414 }
4415
4416 // SDK-defined agents arrive via init, so main.tsx's lookup missed them.
4417 if (mainThreadAgent.initialPrompt) {
4418 structuredIO.prependUserMessage(mainThreadAgent.initialPrompt)
4419 }
4420 } else if (mainThreadAgent?.initialPrompt) {
4421 // Filesystem-defined agent (alreadyResolved by main.tsx). main.tsx
4422 // handles initialPrompt for the string inputPrompt case, but when
4423 // inputPrompt is an AsyncIterable (SDK stream-json), it can't
4424 // concatenate — fall back to prependUserMessage here.
4425 structuredIO.prependUserMessage(mainThreadAgent.initialPrompt)
4426 }
4427 }
4428
4429 const settings = getSettings_DEPRECATED()
4430 const outputStyle = settings?.outputStyle || DEFAULT_OUTPUT_STYLE_NAME
4431 const availableOutputStyles = await getAllOutputStyles(getCwd())
4432
4433 // Get account information
4434 const accountInfo = getAccountInformation()
4435 if (request.hooks) {
4436 const hooks: Partial<Record<HookEvent, HookCallbackMatcher[]>> = {}
4437 for (const [event, matchers] of Object.entries(request.hooks)) {
4438 hooks[event as HookEvent] = matchers.map(matcher => {
4439 const callbacks = matcher.hookCallbackIds.map(callbackId => {
4440 return structuredIO.createHookCallback(callbackId, matcher.timeout)
4441 })
4442 return {
4443 matcher: matcher.matcher,
4444 hooks: callbacks,
4445 }
4446 })
4447 }
4448 registerHookCallbacks(hooks)
4449 }
4450 if (request.jsonSchema) {
4451 setInitJsonSchema(request.jsonSchema)
4452 }
4453 const initResponse: SDKControlInitializeResponse = {
4454 commands: commands
4455 .filter(cmd => cmd.userInvocable !== false)
4456 .map(cmd => ({
4457 name: getCommandName(cmd),
4458 description: formatDescriptionWithSource(cmd),
4459 argumentHint: cmd.argumentHint || '',
4460 })),
4461 agents: agents.map(agent => ({
4462 name: agent.agentType,
4463 description: agent.whenToUse,
4464 // 'inherit' is an internal sentinel; normalize to undefined for the public API
4465 model: agent.model === 'inherit' ? undefined : agent.model,
4466 })),
4467 output_style: outputStyle,
4468 available_output_styles: Object.keys(availableOutputStyles),
4469 models: modelInfos,
4470 account: {
4471 email: accountInfo?.email,
4472 organization: accountInfo?.organization,
4473 subscriptionType: accountInfo?.subscription,
4474 tokenSource: accountInfo?.tokenSource,
4475 apiKeySource: accountInfo?.apiKeySource,
4476 // getAccountInformation() returns undefined under 3P providers, so the
4477 // other fields are all absent. apiProvider disambiguates "not logged
4478 // in" (firstParty + tokenSource:none) from "3P, login not applicable".
4479 apiProvider: getAPIProvider(),
4480 },
4481 pid: process.pid,
4482 }
4483
4484 if (isFastModeEnabled() && isFastModeAvailable()) {
4485 const appState = getAppState()
4486 initResponse.fast_mode_state = getFastModeState(
4487 options.userSpecifiedModel ?? null,
4488 appState.fastMode,
4489 )
4490 }
4491
4492 output.enqueue({
4493 type: 'control_response',
4494 response: {
4495 subtype: 'success',
4496 request_id: requestId,
4497 response: initResponse,
4498 },
4499 })
4500
4501 // After the initialize message, check the auth status-
4502 // This will get notified of changes, but we also want to send the
4503 // initial state.
4504 if (enableAuthStatus) {
4505 const authStatusManager = AwsAuthStatusManager.getInstance()
4506 const status = authStatusManager.getStatus()
4507 if (status) {
4508 output.enqueue({
4509 type: 'auth_status',
4510 isAuthenticating: status.isAuthenticating,
4511 output: status.output,
4512 error: status.error,
4513 uuid: randomUUID(),
4514 session_id: getSessionId(),
4515 })
4516 }
4517 }
4518}
4519
4520async function handleRewindFiles(
4521 userMessageId: UUID,
4522 appState: AppState,
4523 setAppState: (updater: (prev: AppState) => AppState) => void,
4524 dryRun: boolean,
4525): Promise<RewindFilesResult> {
4526 if (!fileHistoryEnabled()) {
4527 return { canRewind: false, error: 'File rewinding is not enabled.' }
4528 }
4529 if (!fileHistoryCanRestore(appState.fileHistory, userMessageId)) {
4530 return {
4531 canRewind: false,
4532 error: 'No file checkpoint found for this message.',
4533 }
4534 }
4535
4536 if (dryRun) {
4537 const diffStats = await fileHistoryGetDiffStats(
4538 appState.fileHistory,
4539 userMessageId,
4540 )
4541 return {
4542 canRewind: true,
4543 filesChanged: diffStats?.filesChanged,
4544 insertions: diffStats?.insertions,
4545 deletions: diffStats?.deletions,
4546 }
4547 }
4548
4549 try {
4550 await fileHistoryRewind(
4551 updater =>
4552 setAppState(prev => ({
4553 ...prev,
4554 fileHistory: updater(prev.fileHistory),
4555 })),
4556 userMessageId,
4557 )
4558 } catch (error) {
4559 return {
4560 canRewind: false,
4561 error: `Failed to rewind: ${errorMessage(error)}`,
4562 }
4563 }
4564
4565 return { canRewind: true }
4566}
4567
4568function handleSetPermissionMode(
4569 request: { mode: InternalPermissionMode },
4570 requestId: string,
4571 toolPermissionContext: ToolPermissionContext,
4572 output: Stream<StdoutMessage>,
4573): ToolPermissionContext {
4574 // Check if trying to switch to bypassPermissions mode
4575 if (request.mode === 'bypassPermissions') {
4576 if (isBypassPermissionsModeDisabled()) {
4577 output.enqueue({
4578 type: 'control_response',
4579 response: {
4580 subtype: 'error',
4581 request_id: requestId,
4582 error:
4583 'Cannot set permission mode to bypassPermissions because it is disabled by settings or configuration',
4584 },
4585 })
4586 return toolPermissionContext
4587 }
4588 if (!toolPermissionContext.isBypassPermissionsModeAvailable) {
4589 output.enqueue({
4590 type: 'control_response',
4591 response: {
4592 subtype: 'error',
4593 request_id: requestId,
4594 error:
4595 'Cannot set permission mode to bypassPermissions because the session was not launched with --dangerously-skip-permissions',
4596 },
4597 })
4598 return toolPermissionContext
4599 }
4600 }
4601
4602 // Check if trying to switch to auto mode without the classifier gate
4603 if (
4604 feature('TRANSCRIPT_CLASSIFIER') &&
4605 request.mode === 'auto' &&
4606 !isAutoModeGateEnabled()
4607 ) {
4608 const reason = getAutoModeUnavailableReason()
4609 output.enqueue({
4610 type: 'control_response',
4611 response: {
4612 subtype: 'error',
4613 request_id: requestId,
4614 error: reason
4615 ? `Cannot set permission mode to auto: ${getAutoModeUnavailableNotification(reason)}`
4616 : 'Cannot set permission mode to auto',
4617 },
4618 })
4619 return toolPermissionContext
4620 }
4621
4622 // Allow the mode switch
4623 output.enqueue({
4624 type: 'control_response',
4625 response: {
4626 subtype: 'success',
4627 request_id: requestId,
4628 response: {
4629 mode: request.mode,
4630 },
4631 },
4632 })
4633
4634 return {
4635 ...transitionPermissionMode(
4636 toolPermissionContext.mode,
4637 request.mode,
4638 toolPermissionContext,
4639 ),
4640 mode: request.mode,
4641 }
4642}
4643
4644/**
4645 * IDE-triggered channel enable. Derives the ChannelEntry from the connection's
4646 * pluginSource (IDE can't spoof kind/marketplace — we only take the server
4647 * name), appends it to session allowedChannels, and runs the full gate. On
4648 * gate failure, rolls back the append. On success, registers a notification
4649 * handler that enqueues channel messages at priority:'next' — drainCommandQueue
4650 * picks them up between turns.
4651 *
4652 * Intentionally does NOT register the claude/channel/permission handler that
4653 * useManageMCPConnections sets up for interactive mode. That handler resolves
4654 * a pending dialog inside handleInteractivePermission — but print.ts never
4655 * calls handleInteractivePermission. When SDK permission lands on 'ask', it
4656 * goes to the consumer's canUseTool callback over stdio; there is no CLI-side
4657 * dialog for a remote "yes tbxkq" to resolve. If an IDE wants channel-relayed
4658 * tool approval, that's IDE-side plumbing against its own pending-map. (Also
4659 * gated separately by tengu_harbor_permissions — not yet shipping on
4660 * interactive either.)
4661 */
4662function handleChannelEnable(
4663 requestId: string,
4664 serverName: string,
4665 connectionPool: readonly MCPServerConnection[],
4666 output: Stream<StdoutMessage>,
4667): void {
4668 const respondError = (error: string) =>
4669 output.enqueue({
4670 type: 'control_response',
4671 response: { subtype: 'error', request_id: requestId, error },
4672 })
4673
4674 if (!(feature('KAIROS') || feature('KAIROS_CHANNELS'))) {
4675 return respondError('channels feature not available in this build')
4676 }
4677
4678 // Only a 'connected' client has .capabilities and .client to register the
4679 // handler on. The pool spread at the call site matches mcp_status.
4680 const connection = connectionPool.find(
4681 c => c.name === serverName && c.type === 'connected',
4682 )
4683 if (!connection || connection.type !== 'connected') {
4684 return respondError(`server ${serverName} is not connected`)
4685 }
4686
4687 const pluginSource = connection.config.pluginSource
4688 const parsed = pluginSource ? parsePluginIdentifier(pluginSource) : undefined
4689 if (!parsed?.marketplace) {
4690 // No pluginSource or @-less source — can never pass the {plugin,
4691 // marketplace}-keyed allowlist. Short-circuit with the same reason the
4692 // gate would produce.
4693 return respondError(
4694 `server ${serverName} is not plugin-sourced; channel_enable requires a marketplace plugin`,
4695 )
4696 }
4697
4698 const entry: ChannelEntry = {
4699 kind: 'plugin',
4700 name: parsed.name,
4701 marketplace: parsed.marketplace,
4702 }
4703 // Idempotency: don't double-append on repeat enable.
4704 const prior = getAllowedChannels()
4705 const already = prior.some(
4706 e =>
4707 e.kind === 'plugin' &&
4708 e.name === entry.name &&
4709 e.marketplace === entry.marketplace,
4710 )
4711 if (!already) setAllowedChannels([...prior, entry])
4712
4713 const gate = gateChannelServer(
4714 serverName,
4715 connection.capabilities,
4716 pluginSource,
4717 )
4718 if (gate.action === 'skip') {
4719 // Rollback — only remove the entry we appended.
4720 if (!already) setAllowedChannels(prior)
4721 return respondError(gate.reason)
4722 }
4723
4724 const pluginId =
4725 `${entry.name}@${entry.marketplace}` as AnalyticsMetadata_I_VERIFIED_THIS_IS_NOT_CODE_OR_FILEPATHS
4726 logMCPDebug(serverName, 'Channel notifications registered')
4727 logEvent('tengu_mcp_channel_enable', { plugin: pluginId })
4728
4729 // Identical enqueue shape to the interactive register block in
4730 // useManageMCPConnections. drainCommandQueue processes it between turns —
4731 // channel messages queue at priority 'next' and are seen by the model on
4732 // the turn after they arrive.
4733 connection.client.setNotificationHandler(
4734 ChannelMessageNotificationSchema(),
4735 async notification => {
4736 const { content, meta } = notification.params
4737 logMCPDebug(
4738 serverName,
4739 `notifications/claude/channel: ${content.slice(0, 80)}`,
4740 )
4741 logEvent('tengu_mcp_channel_message', {
4742 content_length: content.length,
4743 meta_key_count: Object.keys(meta ?? {}).length,
4744 entry_kind:
4745 'plugin' as AnalyticsMetadata_I_VERIFIED_THIS_IS_NOT_CODE_OR_FILEPATHS,
4746 is_dev: false,
4747 plugin: pluginId,
4748 })
4749 enqueue({
4750 mode: 'prompt',
4751 value: wrapChannelMessage(serverName, content, meta),
4752 priority: 'next',
4753 isMeta: true,
4754 origin: { kind: 'channel', server: serverName },
4755 skipSlashCommands: true,
4756 })
4757 },
4758 )
4759
4760 output.enqueue({
4761 type: 'control_response',
4762 response: {
4763 subtype: 'success',
4764 request_id: requestId,
4765 response: undefined,
4766 },
4767 })
4768}
4769
4770/**
4771 * Re-register the channel notification handler after mcp_reconnect /
4772 * mcp_toggle creates a new client. handleChannelEnable bound the handler to
4773 * the OLD client object; allowedChannels survives the reconnect but the
4774 * handler binding does not. Without this, channel messages silently drop
4775 * after a reconnect while the IDE still believes the channel is live.
4776 *
4777 * Mirrors the interactive CLI's onConnectionAttempt in
4778 * useManageMCPConnections, which re-gates on every new connection. Paired
4779 * with registerElicitationHandlers at the same call sites.
4780 *
4781 * No-op if the server was never channel-enabled: gateChannelServer calls
4782 * findChannelEntry internally and returns skip/session for an unlisted
4783 * server, so reconnecting a non-channel MCP server costs one feature-flag
4784 * check.
4785 */
4786function reregisterChannelHandlerAfterReconnect(
4787 connection: MCPServerConnection,
4788): void {
4789 if (!(feature('KAIROS') || feature('KAIROS_CHANNELS'))) return
4790 if (connection.type !== 'connected') return
4791
4792 const gate = gateChannelServer(
4793 connection.name,
4794 connection.capabilities,
4795 connection.config.pluginSource,
4796 )
4797 if (gate.action !== 'register') return
4798
4799 const entry = findChannelEntry(connection.name, getAllowedChannels())
4800 const pluginId =
4801 entry?.kind === 'plugin'
4802 ? (`${entry.name}@${entry.marketplace}` as AnalyticsMetadata_I_VERIFIED_THIS_IS_NOT_CODE_OR_FILEPATHS)
4803 : undefined
4804
4805 logMCPDebug(
4806 connection.name,
4807 'Channel notifications re-registered after reconnect',
4808 )
4809 connection.client.setNotificationHandler(
4810 ChannelMessageNotificationSchema(),
4811 async notification => {
4812 const { content, meta } = notification.params
4813 logMCPDebug(
4814 connection.name,
4815 `notifications/claude/channel: ${content.slice(0, 80)}`,
4816 )
4817 logEvent('tengu_mcp_channel_message', {
4818 content_length: content.length,
4819 meta_key_count: Object.keys(meta ?? {}).length,
4820 entry_kind:
4821 entry?.kind as AnalyticsMetadata_I_VERIFIED_THIS_IS_NOT_CODE_OR_FILEPATHS,
4822 is_dev: entry?.dev ?? false,
4823 plugin: pluginId,
4824 })
4825 enqueue({
4826 mode: 'prompt',
4827 value: wrapChannelMessage(connection.name, content, meta),
4828 priority: 'next',
4829 isMeta: true,
4830 origin: { kind: 'channel', server: connection.name },
4831 skipSlashCommands: true,
4832 })
4833 },
4834 )
4835}
4836
4837/**
4838 * Emits an error message in the correct format based on outputFormat.
4839 * When using stream-json, writes JSON to stdout; otherwise writes plain text to stderr.
4840 */
4841function emitLoadError(
4842 message: string,
4843 outputFormat: string | undefined,
4844): void {
4845 if (outputFormat === 'stream-json') {
4846 const errorResult = {
4847 type: 'result',
4848 subtype: 'error_during_execution',
4849 duration_ms: 0,
4850 duration_api_ms: 0,
4851 is_error: true,
4852 num_turns: 0,
4853 stop_reason: null,
4854 session_id: getSessionId(),
4855 total_cost_usd: 0,
4856 usage: EMPTY_USAGE,
4857 modelUsage: {},
4858 permission_denials: [],
4859 uuid: randomUUID(),
4860 errors: [message],
4861 }
4862 process.stdout.write(jsonStringify(errorResult) + '\n')
4863 } else {
4864 process.stderr.write(message + '\n')
4865 }
4866}
4867
4868/**
4869 * Removes an interrupted user message and its synthetic assistant sentinel
4870 * from the message array. Used during gateway-triggered restarts to clean up
4871 * the message history before re-enqueuing the interrupted prompt.
4872 *
4873 * @internal Exported for testing
4874 */
4875export function removeInterruptedMessage(
4876 messages: Message[],
4877 interruptedUserMessage: NormalizedUserMessage,
4878): void {
4879 const idx = messages.findIndex(m => m.uuid === interruptedUserMessage.uuid)
4880 if (idx !== -1) {
4881 // Remove the user message and the sentinel that immediately follows it.
4882 // splice safely handles the case where idx is the last element.
4883 messages.splice(idx, 2)
4884 }
4885}
4886
4887type LoadInitialMessagesResult = {
4888 messages: Message[]
4889 turnInterruptionState?: TurnInterruptionState
4890 agentSetting?: string
4891}
4892
4893async function loadInitialMessages(
4894 setAppState: (f: (prev: AppState) => AppState) => void,
4895 options: {
4896 continue: boolean | undefined
4897 teleport: string | true | null | undefined
4898 resume: string | boolean | undefined
4899 resumeSessionAt: string | undefined
4900 forkSession: boolean | undefined
4901 outputFormat: string | undefined
4902 sessionStartHooksPromise?: ReturnType<typeof processSessionStartHooks>
4903 restoredWorkerState: Promise<SessionExternalMetadata | null>
4904 },
4905): Promise<LoadInitialMessagesResult> {
4906 const persistSession = !isSessionPersistenceDisabled()
4907 // Handle continue in print mode
4908 if (options.continue) {
4909 try {
4910 logEvent('tengu_continue_print', {})
4911
4912 const result = await loadConversationForResume(
4913 undefined /* sessionId */,
4914 undefined /* file path */,
4915 )
4916 if (result) {
4917 // Match coordinator mode to the resumed session's mode
4918 if (feature('COORDINATOR_MODE') && coordinatorModeModule) {
4919 const warning = coordinatorModeModule.matchSessionMode(result.mode)
4920 if (warning) {
4921 process.stderr.write(warning + '\n')
4922 // Refresh agent definitions to reflect the mode switch
4923 const {
4924 getAgentDefinitionsWithOverrides,
4925 getActiveAgentsFromList,
4926 } =
4927 // eslint-disable-next-line @typescript-eslint/no-require-imports
4928 require('../tools/AgentTool/loadAgentsDir.js') as typeof import('../tools/AgentTool/loadAgentsDir.js')
4929 getAgentDefinitionsWithOverrides.cache.clear?.()
4930 const freshAgentDefs = await getAgentDefinitionsWithOverrides(
4931 getCwd(),
4932 )
4933
4934 setAppState(prev => ({
4935 ...prev,
4936 agentDefinitions: {
4937 ...freshAgentDefs,
4938 allAgents: freshAgentDefs.allAgents,
4939 activeAgents: getActiveAgentsFromList(freshAgentDefs.allAgents),
4940 },
4941 }))
4942 }
4943 }
4944
4945 // Reuse the resumed session's ID
4946 if (!options.forkSession) {
4947 if (result.sessionId) {
4948 switchSession(
4949 asSessionId(result.sessionId),
4950 result.fullPath ? dirname(result.fullPath) : null,
4951 )
4952 if (persistSession) {
4953 await resetSessionFilePointer()
4954 }
4955 }
4956 }
4957 restoreSessionStateFromLog(result, setAppState)
4958
4959 // Restore session metadata so it's re-appended on exit via reAppendSessionMetadata
4960 restoreSessionMetadata(
4961 options.forkSession
4962 ? { ...result, worktreeSession: undefined }
4963 : result,
4964 )
4965
4966 // Write mode entry for the resumed session
4967 if (feature('COORDINATOR_MODE') && coordinatorModeModule) {
4968 saveMode(
4969 coordinatorModeModule.isCoordinatorMode()
4970 ? 'coordinator'
4971 : 'normal',
4972 )
4973 }
4974
4975 return {
4976 messages: result.messages,
4977 turnInterruptionState: result.turnInterruptionState,
4978 agentSetting: result.agentSetting,
4979 }
4980 }
4981 } catch (error) {
4982 logError(error)
4983 gracefulShutdownSync(1)
4984 return { messages: [] }
4985 }
4986 }
4987
4988 // Handle teleport in print mode
4989 if (options.teleport) {
4990 try {
4991 if (!isPolicyAllowed('allow_remote_sessions')) {
4992 throw new Error(
4993 "Remote sessions are disabled by your organization's policy.",
4994 )
4995 }
4996
4997 logEvent('tengu_teleport_print', {})
4998
4999 if (typeof options.teleport !== 'string') {
5000 throw new Error('No session ID provided for teleport')
5001 }
5002
5003 const {
5004 checkOutTeleportedSessionBranch,
5005 processMessagesForTeleportResume,
5006 teleportResumeCodeSession,
5007 validateGitState,
5008 } = await import('src/utils/teleport.js')
5009 await validateGitState()
5010 const teleportResult = await teleportResumeCodeSession(options.teleport)
5011 const { branchError } = await checkOutTeleportedSessionBranch(
5012 teleportResult.branch,
5013 )
5014 return {
5015 messages: processMessagesForTeleportResume(
5016 teleportResult.log,
5017 branchError,
5018 ),
5019 }
5020 } catch (error) {
5021 logError(error)
5022 gracefulShutdownSync(1)
5023 return { messages: [] }
5024 }
5025 }
5026
5027 // Handle resume in print mode (accepts session ID or URL)
5028 // URLs are [ANT-ONLY]
5029 if (options.resume) {
5030 try {
5031 logEvent('tengu_resume_print', {})
5032
5033 // In print mode - we require a valid session ID, JSONL file or URL
5034 const parsedSessionId = parseSessionIdentifier(
5035 typeof options.resume === 'string' ? options.resume : '',
5036 )
5037 if (!parsedSessionId) {
5038 let errorMessage =
5039 'Error: --resume requires a valid session ID when used with --print. Usage: claude -p --resume <session-id>'
5040 if (typeof options.resume === 'string') {
5041 errorMessage += `. Session IDs must be in UUID format (e.g., 550e8400-e29b-41d4-a716-446655440000). Provided value "${options.resume}" is not a valid UUID`
5042 }
5043 emitLoadError(errorMessage, options.outputFormat)
5044 gracefulShutdownSync(1)
5045 return { messages: [] }
5046 }
5047
5048 // Hydrate local transcript from remote before loading
5049 if (isEnvTruthy(process.env.CLAUDE_CODE_USE_CCR_V2)) {
5050 // Await restore alongside hydration so SSE catchup lands on
5051 // restored state, not a fresh default.
5052 const [, metadata] = await Promise.all([
5053 hydrateFromCCRv2InternalEvents(parsedSessionId.sessionId),
5054 options.restoredWorkerState,
5055 ])
5056 if (metadata) {
5057 setAppState(externalMetadataToAppState(metadata))
5058 if (typeof metadata.model === 'string') {
5059 setMainLoopModelOverride(metadata.model)
5060 }
5061 }
5062 } else if (
5063 parsedSessionId.isUrl &&
5064 parsedSessionId.ingressUrl &&
5065 isEnvTruthy(process.env.ENABLE_SESSION_PERSISTENCE)
5066 ) {
5067 // v1: fetch session logs from Session Ingress
5068 await hydrateRemoteSession(
5069 parsedSessionId.sessionId,
5070 parsedSessionId.ingressUrl,
5071 )
5072 }
5073
5074 // Load the conversation with the specified session ID
5075 const result = await loadConversationForResume(
5076 parsedSessionId.sessionId,
5077 parsedSessionId.jsonlFile || undefined,
5078 )
5079
5080 // hydrateFromCCRv2InternalEvents writes an empty transcript file for
5081 // fresh sessions (writeFile(sessionFile, '') with zero events), so
5082 // loadConversationForResume returns {messages: []} not null. Treat
5083 // empty the same as null so SessionStart still fires.
5084 if (!result || result.messages.length === 0) {
5085 // For URL-based or CCR v2 resume, start with empty session (it was hydrated but empty)
5086 if (
5087 parsedSessionId.isUrl ||
5088 isEnvTruthy(process.env.CLAUDE_CODE_USE_CCR_V2)
5089 ) {
5090 // Execute SessionStart hooks for startup since we're starting a new session
5091 return {
5092 messages: await (options.sessionStartHooksPromise ??
5093 processSessionStartHooks('startup')),
5094 }
5095 } else {
5096 emitLoadError(
5097 `No conversation found with session ID: ${parsedSessionId.sessionId}`,
5098 options.outputFormat,
5099 )
5100 gracefulShutdownSync(1)
5101 return { messages: [] }
5102 }
5103 }
5104
5105 // Handle resumeSessionAt feature
5106 if (options.resumeSessionAt) {
5107 const index = result.messages.findIndex(
5108 m => m.uuid === options.resumeSessionAt,
5109 )
5110 if (index < 0) {
5111 emitLoadError(
5112 `No message found with message.uuid of: ${options.resumeSessionAt}`,
5113 options.outputFormat,
5114 )
5115 gracefulShutdownSync(1)
5116 return { messages: [] }
5117 }
5118
5119 result.messages = index >= 0 ? result.messages.slice(0, index + 1) : []
5120 }
5121
5122 // Match coordinator mode to the resumed session's mode
5123 if (feature('COORDINATOR_MODE') && coordinatorModeModule) {
5124 const warning = coordinatorModeModule.matchSessionMode(result.mode)
5125 if (warning) {
5126 process.stderr.write(warning + '\n')
5127 // Refresh agent definitions to reflect the mode switch
5128 const { getAgentDefinitionsWithOverrides, getActiveAgentsFromList } =
5129 // eslint-disable-next-line @typescript-eslint/no-require-imports
5130 require('../tools/AgentTool/loadAgentsDir.js') as typeof import('../tools/AgentTool/loadAgentsDir.js')
5131 getAgentDefinitionsWithOverrides.cache.clear?.()
5132 const freshAgentDefs = await getAgentDefinitionsWithOverrides(
5133 getCwd(),
5134 )
5135
5136 setAppState(prev => ({
5137 ...prev,
5138 agentDefinitions: {
5139 ...freshAgentDefs,
5140 allAgents: freshAgentDefs.allAgents,
5141 activeAgents: getActiveAgentsFromList(freshAgentDefs.allAgents),
5142 },
5143 }))
5144 }
5145 }
5146
5147 // Reuse the resumed session's ID
5148 if (!options.forkSession && result.sessionId) {
5149 switchSession(
5150 asSessionId(result.sessionId),
5151 result.fullPath ? dirname(result.fullPath) : null,
5152 )
5153 if (persistSession) {
5154 await resetSessionFilePointer()
5155 }
5156 }
5157 restoreSessionStateFromLog(result, setAppState)
5158
5159 // Restore session metadata so it's re-appended on exit via reAppendSessionMetadata
5160 restoreSessionMetadata(
5161 options.forkSession
5162 ? { ...result, worktreeSession: undefined }
5163 : result,
5164 )
5165
5166 // Write mode entry for the resumed session
5167 if (feature('COORDINATOR_MODE') && coordinatorModeModule) {
5168 saveMode(
5169 coordinatorModeModule.isCoordinatorMode() ? 'coordinator' : 'normal',
5170 )
5171 }
5172
5173 return {
5174 messages: result.messages,
5175 turnInterruptionState: result.turnInterruptionState,
5176 agentSetting: result.agentSetting,
5177 }
5178 } catch (error) {
5179 logError(error)
5180 const errorMessage =
5181 error instanceof Error
5182 ? `Failed to resume session: ${error.message}`
5183 : 'Failed to resume session with --print mode'
5184 emitLoadError(errorMessage, options.outputFormat)
5185 gracefulShutdownSync(1)
5186 return { messages: [] }
5187 }
5188 }
5189
5190 // Join the SessionStart hooks promise kicked in main.tsx (or run fresh if
5191 // it wasn't kicked — e.g. --continue with no prior session falls through
5192 // here with sessionStartHooksPromise undefined because main.tsx guards on continue)
5193 return {
5194 messages: await (options.sessionStartHooksPromise ??
5195 processSessionStartHooks('startup')),
5196 }
5197}
5198
5199function getStructuredIO(
5200 inputPrompt: string | AsyncIterable<string>,
5201 options: {
5202 sdkUrl: string | undefined
5203 replayUserMessages?: boolean
5204 },
5205): StructuredIO {
5206 let inputStream: AsyncIterable<string>
5207 if (typeof inputPrompt === 'string') {
5208 if (inputPrompt.trim() !== '') {
5209 // Normalize to a streaming input.
5210 inputStream = fromArray([
5211 jsonStringify({
5212 type: 'user',
5213 session_id: '',
5214 message: {
5215 role: 'user',
5216 content: inputPrompt,
5217 },
5218 parent_tool_use_id: null,
5219 } satisfies SDKUserMessage),
5220 ])
5221 } else {
5222 // Empty string - create empty stream
5223 inputStream = fromArray([])
5224 }
5225 } else {
5226 inputStream = inputPrompt
5227 }
5228
5229 // Use RemoteIO if sdkUrl is provided, otherwise use regular StructuredIO
5230 return options.sdkUrl
5231 ? new RemoteIO(options.sdkUrl, inputStream, options.replayUserMessages)
5232 : new StructuredIO(inputStream, options.replayUserMessages)
5233}
5234
5235/**
5236 * Handles unexpected permission responses by looking up the unresolved tool
5237 * call in the transcript and enqueuing it for execution.
5238 *
5239 * Returns true if a permission was enqueued, false otherwise.
5240 */
5241export async function handleOrphanedPermissionResponse({
5242 message,
5243 setAppState,
5244 onEnqueued,
5245 handledToolUseIds,
5246}: {
5247 message: SDKControlResponse
5248 setAppState: (f: (prev: AppState) => AppState) => void
5249 onEnqueued?: () => void
5250 handledToolUseIds: Set<string>
5251}): Promise<boolean> {
5252 if (
5253 message.response.subtype === 'success' &&
5254 message.response.response?.toolUseID &&
5255 typeof message.response.response.toolUseID === 'string'
5256 ) {
5257 const permissionResult = message.response.response as PermissionResult
5258 const { toolUseID } = permissionResult
5259 if (!toolUseID) {
5260 return false
5261 }
5262
5263 logForDebugging(
5264 `handleOrphanedPermissionResponse: received orphaned control_response for toolUseID=${toolUseID} request_id=${message.response.request_id}`,
5265 )
5266
5267 // Prevent re-processing the same orphaned tool_use. Without this guard,
5268 // duplicate control_response deliveries (e.g. from WebSocket reconnect)
5269 // cause the same tool to be executed multiple times, producing duplicate
5270 // tool_use IDs in the messages array and a 400 error from the API.
5271 // Once corrupted, every retry accumulates more duplicates.
5272 if (handledToolUseIds.has(toolUseID)) {
5273 logForDebugging(
5274 `handleOrphanedPermissionResponse: skipping duplicate orphaned permission for toolUseID=${toolUseID} (already handled)`,
5275 )
5276 return false
5277 }
5278
5279 const assistantMessage = await findUnresolvedToolUse(toolUseID)
5280 if (!assistantMessage) {
5281 logForDebugging(
5282 `handleOrphanedPermissionResponse: no unresolved tool_use found for toolUseID=${toolUseID} (already resolved in transcript)`,
5283 )
5284 return false
5285 }
5286
5287 handledToolUseIds.add(toolUseID)
5288 logForDebugging(
5289 `handleOrphanedPermissionResponse: enqueuing orphaned permission for toolUseID=${toolUseID} messageID=${assistantMessage.message.id}`,
5290 )
5291 enqueue({
5292 mode: 'orphaned-permission' as const,
5293 value: [],
5294 orphanedPermission: {
5295 permissionResult,
5296 assistantMessage,
5297 },
5298 })
5299
5300 onEnqueued?.()
5301 return true
5302 }
5303 return false
5304}
5305
5306export type DynamicMcpState = {
5307 clients: MCPServerConnection[]
5308 tools: Tools
5309 configs: Record<string, ScopedMcpServerConfig>
5310}
5311
5312/**
5313 * Converts a process transport config to a scoped config.
5314 * The types are structurally compatible, so we just add the scope.
5315 */
5316function toScopedConfig(
5317 config: McpServerConfigForProcessTransport,
5318): ScopedMcpServerConfig {
5319 // McpServerConfigForProcessTransport is a subset of McpServerConfig
5320 // (it excludes IDE-specific types like sse-ide and ws-ide)
5321 // Adding scope makes it a valid ScopedMcpServerConfig
5322 return { ...config, scope: 'dynamic' } as ScopedMcpServerConfig
5323}
5324
5325/**
5326 * State for SDK MCP servers that run in the SDK process.
5327 */
5328export type SdkMcpState = {
5329 configs: Record<string, McpSdkServerConfig>
5330 clients: MCPServerConnection[]
5331 tools: Tools
5332}
5333
5334/**
5335 * Result of handleMcpSetServers - contains new state and response data.
5336 */
5337export type McpSetServersResult = {
5338 response: SDKControlMcpSetServersResponse
5339 newSdkState: SdkMcpState
5340 newDynamicState: DynamicMcpState
5341 sdkServersChanged: boolean
5342}
5343
5344/**
5345 * Handles mcp_set_servers requests by processing both SDK and process-based servers.
5346 * SDK servers run in the SDK process; process-based servers are spawned by the CLI.
5347 *
5348 * Applies enterprise allowedMcpServers/deniedMcpServers policy — same filter as
5349 * --mcp-config (see filterMcpServersByPolicy call in main.tsx). Without this,
5350 * SDK V2 Query.setMcpServers() was a second policy bypass vector. Blocked servers
5351 * are reported in response.errors so the SDK consumer knows why they weren't added.
5352 */
5353export async function handleMcpSetServers(
5354 servers: Record<string, McpServerConfigForProcessTransport>,
5355 sdkState: SdkMcpState,
5356 dynamicState: DynamicMcpState,
5357 setAppState: (f: (prev: AppState) => AppState) => void,
5358): Promise<McpSetServersResult> {
5359 // Enforce enterprise MCP policy on process-based servers (stdio/http/sse).
5360 // Mirrors the --mcp-config filter in main.tsx — both user-controlled injection
5361 // paths must have the same gate. type:'sdk' servers are exempt (SDK-managed,
5362 // CLI never spawns/connects for them — see filterMcpServersByPolicy jsdoc).
5363 // Blocked servers go into response.errors so the SDK caller sees why.
5364 const { allowed: allowedServers, blocked } = filterMcpServersByPolicy(servers)
5365 const policyErrors: Record<string, string> = {}
5366 for (const name of blocked) {
5367 policyErrors[name] =
5368 'Blocked by enterprise policy (allowedMcpServers/deniedMcpServers)'
5369 }
5370
5371 // Separate SDK servers from process-based servers
5372 const sdkServers: Record<string, McpSdkServerConfig> = {}
5373 const processServers: Record<string, McpServerConfigForProcessTransport> = {}
5374
5375 for (const [name, config] of Object.entries(allowedServers)) {
5376 if (config.type === 'sdk') {
5377 sdkServers[name] = config
5378 } else {
5379 processServers[name] = config
5380 }
5381 }
5382
5383 // Handle SDK servers
5384 const currentSdkNames = new Set(Object.keys(sdkState.configs))
5385 const newSdkNames = new Set(Object.keys(sdkServers))
5386 const sdkAdded: string[] = []
5387 const sdkRemoved: string[] = []
5388
5389 const newSdkConfigs = { ...sdkState.configs }
5390 let newSdkClients = [...sdkState.clients]
5391 let newSdkTools = [...sdkState.tools]
5392
5393 // Remove SDK servers no longer in desired state
5394 for (const name of currentSdkNames) {
5395 if (!newSdkNames.has(name)) {
5396 const client = newSdkClients.find(c => c.name === name)
5397 if (client && client.type === 'connected') {
5398 await client.cleanup()
5399 }
5400 newSdkClients = newSdkClients.filter(c => c.name !== name)
5401 const prefix = `mcp__${name}__`
5402 newSdkTools = newSdkTools.filter(t => !t.name.startsWith(prefix))
5403 delete newSdkConfigs[name]
5404 sdkRemoved.push(name)
5405 }
5406 }
5407
5408 // Add new SDK servers as pending - they'll be upgraded to connected
5409 // when updateSdkMcp() runs on the next query
5410 for (const [name, config] of Object.entries(sdkServers)) {
5411 if (!currentSdkNames.has(name)) {
5412 newSdkConfigs[name] = config
5413 const pendingClient: MCPServerConnection = {
5414 type: 'pending',
5415 name,
5416 config: { ...config, scope: 'dynamic' as const },
5417 }
5418 newSdkClients = [...newSdkClients, pendingClient]
5419 sdkAdded.push(name)
5420 }
5421 }
5422
5423 // Handle process-based servers
5424 const processResult = await reconcileMcpServers(
5425 processServers,
5426 dynamicState,
5427 setAppState,
5428 )
5429
5430 return {
5431 response: {
5432 added: [...sdkAdded, ...processResult.response.added],
5433 removed: [...sdkRemoved, ...processResult.response.removed],
5434 errors: { ...policyErrors, ...processResult.response.errors },
5435 },
5436 newSdkState: {
5437 configs: newSdkConfigs,
5438 clients: newSdkClients,
5439 tools: newSdkTools,
5440 },
5441 newDynamicState: processResult.newState,
5442 sdkServersChanged: sdkAdded.length > 0 || sdkRemoved.length > 0,
5443 }
5444}
5445
5446/**
5447 * Reconciles the current set of dynamic MCP servers with a new desired state.
5448 * Handles additions, removals, and config changes.
5449 */
5450export async function reconcileMcpServers(
5451 desiredConfigs: Record<string, McpServerConfigForProcessTransport>,
5452 currentState: DynamicMcpState,
5453 setAppState: (f: (prev: AppState) => AppState) => void,
5454): Promise<{
5455 response: SDKControlMcpSetServersResponse
5456 newState: DynamicMcpState
5457}> {
5458 const currentNames = new Set(Object.keys(currentState.configs))
5459 const desiredNames = new Set(Object.keys(desiredConfigs))
5460
5461 const toRemove = [...currentNames].filter(n => !desiredNames.has(n))
5462 const toAdd = [...desiredNames].filter(n => !currentNames.has(n))
5463
5464 // Check for config changes (same name, different config)
5465 const toCheck = [...currentNames].filter(n => desiredNames.has(n))
5466 const toReplace = toCheck.filter(name => {
5467 const currentConfig = currentState.configs[name]
5468 const desiredConfigRaw = desiredConfigs[name]
5469 if (!currentConfig || !desiredConfigRaw) return true
5470 const desiredConfig = toScopedConfig(desiredConfigRaw)
5471 return !areMcpConfigsEqual(currentConfig, desiredConfig)
5472 })
5473
5474 const removed: string[] = []
5475 const added: string[] = []
5476 const errors: Record<string, string> = {}
5477
5478 let newClients = [...currentState.clients]
5479 let newTools = [...currentState.tools]
5480
5481 // Remove old servers (including ones being replaced)
5482 for (const name of [...toRemove, ...toReplace]) {
5483 const client = newClients.find(c => c.name === name)
5484 const config = currentState.configs[name]
5485 if (client && config) {
5486 if (client.type === 'connected') {
5487 try {
5488 await client.cleanup()
5489 } catch (e) {
5490 logError(e)
5491 }
5492 }
5493 // Clear the memoization cache
5494 await clearServerCache(name, config)
5495 }
5496
5497 // Remove tools from this server
5498 const prefix = `mcp__${name}__`
5499 newTools = newTools.filter(t => !t.name.startsWith(prefix))
5500
5501 // Remove from clients list
5502 newClients = newClients.filter(c => c.name !== name)
5503
5504 // Track removal (only for actually removed, not replaced)
5505 if (toRemove.includes(name)) {
5506 removed.push(name)
5507 }
5508 }
5509
5510 // Add new servers (including replacements)
5511 for (const name of [...toAdd, ...toReplace]) {
5512 const config = desiredConfigs[name]
5513 if (!config) continue
5514 const scopedConfig = toScopedConfig(config)
5515
5516 // SDK servers are managed by the SDK process, not the CLI.
5517 // Just track them without trying to connect.
5518 if (config.type === 'sdk') {
5519 added.push(name)
5520 continue
5521 }
5522
5523 try {
5524 const client = await connectToServer(name, scopedConfig)
5525 newClients.push(client)
5526
5527 if (client.type === 'connected') {
5528 const serverTools = await fetchToolsForClient(client)
5529 newTools.push(...serverTools)
5530 } else if (client.type === 'failed') {
5531 errors[name] = client.error || 'Connection failed'
5532 }
5533
5534 added.push(name)
5535 } catch (e) {
5536 const err = toError(e)
5537 errors[name] = err.message
5538 logError(err)
5539 }
5540 }
5541
5542 // Build new configs
5543 const newConfigs: Record<string, ScopedMcpServerConfig> = {}
5544 for (const name of desiredNames) {
5545 const config = desiredConfigs[name]
5546 if (config) {
5547 newConfigs[name] = toScopedConfig(config)
5548 }
5549 }
5550
5551 const newState: DynamicMcpState = {
5552 clients: newClients,
5553 tools: newTools,
5554 configs: newConfigs,
5555 }
5556
5557 // Update AppState with the new tools
5558 setAppState(prev => {
5559 // Get all dynamic server names (current + new)
5560 const allDynamicServerNames = new Set([
5561 ...Object.keys(currentState.configs),
5562 ...Object.keys(newConfigs),
5563 ])
5564
5565 // Remove old dynamic tools
5566 const nonDynamicTools = prev.mcp.tools.filter(t => {
5567 for (const serverName of allDynamicServerNames) {
5568 if (t.name.startsWith(`mcp__${serverName}__`)) {
5569 return false
5570 }
5571 }
5572 return true
5573 })
5574
5575 // Remove old dynamic clients
5576 const nonDynamicClients = prev.mcp.clients.filter(c => {
5577 return !allDynamicServerNames.has(c.name)
5578 })
5579
5580 return {
5581 ...prev,
5582 mcp: {
5583 ...prev.mcp,
5584 tools: [...nonDynamicTools, ...newTools],
5585 clients: [...nonDynamicClients, ...newClients],
5586 },
5587 }
5588 })
5589
5590 return {
5591 response: { added, removed, errors },
5592 newState,
5593 }
5594}