basant307/AI_Governance_Project
045
1/**2 * @license3 * Copyright 2025 Google LLC4 * SPDX-License-Identifier: Apache-2.05 */6import { isSlashCommand } from './ui/utils/commandUtils.js';7import { isInlineModelOverrideAllowed } from './utils/acpModelUtils.js';8import { executeToolCall, shutdownTelemetry, isTelemetrySdkInitialized, GeminiEventType, FatalInputError, promptIdContext, OutputFormat, InputFormat, LoopType, ToolNames, uiTelemetryService, parseAndFormatApiError, createDebugLogger, detectAutonomousSentinel, detectLoopSentinel, SendMessageType, buildSyntheticToolResponseParts, detectTurnInterruption, ORPHAN_TOOL_USE_REPAIR_REASON, restoreWorktreeContext, TeamEventType, ApprovalMode, ToolConfirmationOutcome, createDuplicateProviderToolCallResponse, isSystemReminderContent, markDuplicateProviderToolCallResponseSent, findRepeatedDuplicateProviderToolCall, } from '@qwen-code/qwen-code-core';9import { JsonOutputAdapter } from './nonInteractive/io/JsonOutputAdapter.js';10import { StreamJsonOutputAdapter } from './nonInteractive/io/StreamJsonOutputAdapter.js';11import { handleSlashCommand } from './nonInteractiveCliCommands.js';12import { handleAtCommand } from './ui/hooks/atCommandProcessor.js';13import { AlreadyReportedError, handleError, handleToolError, handleCancellationError, handleMaxTurnsExceededError, handleBudgetExceededError, } from './utils/errors.js';14import { RunBudgetEnforcer } from './utils/runBudget.js';15const debugLogger = createDebugLogger('NON_INTERACTIVE_CLI');16/**17 * Maximum wait, in milliseconds, for in-flight background tasks to emit18 * their terminal `task_notification` after `abortAll()` on the19 * structured-output success path. Tasks are marked cancelled20 * synchronously by `abortAll`, but the natural task handler emits the21 * notification on a later microtask — without a brief holdback the22 * structured-output run would silently drop those events. Capped so a23 * slow agent can't block exit indefinitely.24 */25const STRUCTURED_SHUTDOWN_HOLDBACK_MS = 500;26function isHeadlessLoopSentinel(prompt) {27 return (detectLoopSentinel(prompt) !== null ||28 detectAutonomousSentinel(prompt) !== null);29}30/**31 * Body of the synthesised `tool_result` for a `tool_use` block that was32 * suppressed because a sibling `structured_output` call took precedence33 * as the terminal output for the same turn.34 *35 * Two variants — the success-path body drops the trailing "Re-issue this36 * call in a separate turn if needed." sentence because the session37 * terminates immediately after synthesis (no model or SDK consumer can38 * act on the advice). The retry-path body keeps it: when the structured39 * call failed validation, the model is about to receive these parts in40 * the next turn and may legitimately re-issue the suppressed call.41 *42 * Shared between the main-turn and drain-turn synthesis sites so a43 * future wording change can't desync them.44 */45const SUPPRESSED_OUTPUT_SUCCESS = "Skipped: this turn's structured_output contract took precedence as the terminal output.";46const SUPPRESSED_OUTPUT_RETRY = `${SUPPRESSED_OUTPUT_SUCCESS} Re-issue this call in a separate turn if needed.`;47function suppressedOutputBody(structuredCaptured) {48 return structuredCaptured49 ? SUPPRESSED_OUTPUT_SUCCESS50 : SUPPRESSED_OUTPUT_RETRY;51}52import { normalizePartList, extractPartsFromUserMessage, buildSystemMessage, createToolProgressHandler, createAgentToolProgressHandler, computeUsageFromMetrics, buildInitialSystemReminders, insertAfterFunctionResponses, } from './utils/nonInteractiveHelpers.js';53// Human-readable labels for the detectors that can fire mid-stream.54// Surfaced to stderr in TEXT mode so a headless run that halts on a loop55// doesn't exit with empty stdout and no explanation — see PR #3236 review.56const LOOP_TYPE_LABELS = {57 [LoopType.CONSECUTIVE_IDENTICAL_TOOL_CALLS]: 'the model repeated the same tool call with identical arguments',58 [LoopType.CHANTING_IDENTICAL_SENTENCES]: 'the model repeated the same sentence in its output',59 [LoopType.REPETITIVE_THOUGHTS]: 'the model repeated the same reasoning thought',60 [LoopType.READ_FILE_LOOP]: 'the model spent too many consecutive calls reading files without making progress',61 [LoopType.ACTION_STAGNATION]: 'the model kept calling the same tool without making progress',62 [LoopType.SHELL_COMMAND_STAGNATION]: 'the model repeated similar shell inspection commands without making progress',63 [LoopType.GLOBAL_TOOL_CALL_DUPLICATE]: 'the model repeated the same tool call across the turn, even when not back-to-back',64 [LoopType.ALTERNATING_TOOL_CALL_PATTERN]: 'the model alternated between the same two tool calls in a repeating pattern',65 [LoopType.TURN_TOOL_CALL_CAP]: 'the model exceeded the maximum number of tool calls allowed in a single turn',66 [LoopType.INVALID_TOOL_PARAMS_STAGNATION]: 'the model repeatedly sent invalid tool parameters without correcting them',67};68function formatLoopDetectedMessage(loopType) {69 const reason = loopType ? LOOP_TYPE_LABELS[loopType] : undefined;70 const detail = reason ? ` (${loopType}: ${reason})` : '';71 // The always-on guards run before the skipLoopDetection gate, so that72 // setting can't disable them — don't suggest it for those loop types. The73 // per-turn cap is also always-on but has its own knob, so it gets a74 // dedicated hint instead of membership in this list.75 const isAlwaysOn = loopType === LoopType.CONSECUTIVE_IDENTICAL_TOOL_CALLS ||76 loopType === LoopType.SHELL_COMMAND_STAGNATION ||77 loopType === LoopType.GLOBAL_TOOL_CALL_DUPLICATE ||78 loopType === LoopType.INVALID_TOOL_PARAMS_STAGNATION;79 const hint = loopType === LoopType.TURN_TOOL_CALL_CAP80 ? ' Raise the `model.maxToolCallsPerTurn` setting to allow longer turns, or set it to 0 to disable the cap.'81 : isAlwaysOn82 ? ' This is an always-on guard and cannot be disabled via `model.skipLoopDetection`.'83 : ' Set the `model.skipLoopDetection` setting to true to disable.';84 return `Loop detection halted the run${detail}.${hint}`;85}86/**87 * Headless handling for fired loop sentinels. loop.md and autonomous sentinel88 * expansion is interactive-only for now, so a bare sentinel can't be turned into89 * a real prompt here — the tick is skipped (no-op) rather than sent to the model90 * as empty content. Returns true when `job` was a sentinel so the caller skips91 * enqueuing it.92 *93 * A recurring SESSION (non-durable) loop.md job would otherwise stay in94 * `scheduler.sessionSize` and re-fire every interval, pinning the headless run95 * open forever (the hold-open resolves only when sessionSize hits zero); delete96 * it so the run can terminate. Durable jobs are left untouched here — they97 * persist for a future owning session and never count toward sessionSize — and98 * a one-shot job is already removed before it fires.99 *100 * Note: a DURABLE loop.md sentinel never even reaches this callback in headless,101 * because `setSkipDurableFire` filters it at the scheduler before any fire or102 * lastFiredAt persist (otherwise the tick would be marked fired while the work103 * is skipped — silent loss). This guard's durable branch is kept defensive.104 */105export function skipHeadlessLoopSentinel(scheduler, job) {106 if (!isHeadlessLoopSentinel(job.prompt)) {107 return false;108 }109 if (job.recurring && !job.durable) {110 // A user created this recurring loop.md cron via /loop in interactive mode;111 // deleting it here is otherwise silent, so leave a trace of why it vanished112 // from `cron list` when the same workspace is later run headless.113 debugLogger.debug('skipHeadlessLoopSentinel: cleaning up recurring session loop.md cron in headless mode', { jobId: job.id });114 // delete() removes the in-memory job synchronously before any await, so the115 // sessionSize check that follows this call sees it gone; the returned promise116 // has no on-disk work for a session job. Fire-and-forget, but swallow a117 // rejection so a future async delete() can't surface as an unhandled118 // rejection (fatal under Node's --unhandled-rejections=throw).119 void scheduler.delete(job.id).catch(() => {120 /* session job: nothing to clean up on a delete failure */121 });122 }123 return true;124}125function emitLoopDetectedMessage(config, loopType) {126 const message = formatLoopDetectedMessage(loopType);127 // In TEXT mode the adapter swallows LoopDetected, so we print here. In128 // JSON modes the adapter emits a structured result, which is enough.129 if (config.getOutputFormat() !== OutputFormat.TEXT) {130 return message;131 }132 process.stderr.write(`${message}\n`);133 return message;134}135/**136 * Emits a final message for slash command results.137 * Note: systemMessage should already be emitted before calling this function.138 */139async function emitNonInteractiveFinalMessage(params) {140 const { message, isError, adapter, config } = params;141 // JSON output mode: emit assistant message and result142 // (systemMessage should already be emitted by caller)143 adapter.startAssistantMessage();144 adapter.processEvent({145 type: GeminiEventType.Content,146 value: message,147 });148 adapter.finalizeAssistantMessage();149 const metrics = uiTelemetryService.getMetrics();150 const usage = computeUsageFromMetrics(metrics);151 const outputFormat = config.getOutputFormat();152 const stats = outputFormat === OutputFormat.JSON153 ? uiTelemetryService.getMetrics()154 : undefined;155 adapter.emitResult({156 isError,157 durationMs: Date.now() - params.startTimeMs,158 apiDurationMs: 0,159 numTurns: 0,160 errorMessage: isError ? message : undefined,161 usage,162 stats,163 summary: message,164 });165}166/**167 * Executes the non-interactive CLI flow for a single request.168 */169export async function runNonInteractive(config, settings, input, prompt_id, options = {}) {170 return promptIdContext.run(prompt_id, async () => {171 // Create output adapter based on format172 let adapter;173 const outputFormat = config.getOutputFormat();174 if (options.adapter) {175 adapter = options.adapter;176 }177 else if (outputFormat === OutputFormat.STREAM_JSON) {178 adapter = new StreamJsonOutputAdapter(config, config.getIncludePartialMessages());179 }180 else {181 adapter = new JsonOutputAdapter(config);182 }183 const emitResult = (result) => {184 // Fire the callback only after a successful emit. The continue caller185 // (session.ts) uses it to mark the result as delivered and swallow any186 // later error; if emitResult itself throws, the flag must stay unset so187 // the error still surfaces instead of losing both result and error.188 adapter.emitResult(result);189 options.onResultEmitted?.();190 };191 // Get readonly values once at the start192 const sessionId = config.getSessionId();193 const permissionMode = config.getApprovalMode();194 let turnCount = 0;195 let totalApiDurationMs = 0;196 const startTime = Date.now();197 const geminiClient = config.getGeminiClient();198 const abortController = options.abortController ?? new AbortController();199 // Run-level budget enforcement for headless / unattended runs200 // (issue #4103). Tied to the same abortController as user-initiated201 // SIGINT so the existing cancellation plumbing carries the abort;202 // `routeAbort` below interprets the reason so the user sees203 // "budget exceeded" instead of a generic "cancelled" envelope.204 const budgetEnforcer = new RunBudgetEnforcer({205 maxWallTimeSeconds: config.getMaxWallTimeSeconds(),206 maxToolCalls: config.getMaxToolCalls(),207 }, abortController);208 budgetEnforcer.start();209 /**210 * Called at every abort-detection site in place of211 * `handleCancellationError` directly. If a budget tripped, surface the212 * structured budget error (exit 55); otherwise fall through to the213 * SIGINT / user-cancel path (exit 130) so existing behavior is214 * preserved. Both branches call into `process.exit(...)` so the215 * `unreachable` throw is only present to keep the type-checker honest.216 */217 const routeAbort = async () => {218 const exceeded = budgetEnforcer.getExceeded();219 if (exceeded) {220 await handleBudgetExceededError(config, exceeded);221 // Explicit unreachable — `handleBudgetExceededError` is `never`222 // in production (it calls `process.exit`). If a test stubs223 // `process.exit` or a future refactor makes the handler224 // resumable, this throw carries the original budget message225 // so the outer catch's `errorMessage` field stays actionable226 // (vs. a useless literal "unreachable").227 throw new Error(exceeded.message);228 }229 await handleCancellationError(config);230 throw new Error('Operation cancelled.');231 };232 const localQueue = [];233 const sdkOnlyMonitorQueue = [];234 const emitNotificationToSdk = (item) => {235 if (item.sendMessageType !== SendMessageType.Notification)236 return;237 adapter.emitUserMessage([{ text: item.displayText }]);238 if (item.sdkNotification) {239 adapter.emitSystemMessage('task_notification', item.sdkNotification);240 }241 };242 const flushQueuedNotificationsToSdk = (queue) => {243 while (queue.length > 0) {244 emitNotificationToSdk(queue.shift());245 }246 };247 let captureMonitorTurnsInLocalQueue = true;248 let oneShotMonitorsFinalized = false;249 const finalizeOneShotMonitors = () => {250 if (options.captureMonitorNotifications === false ||251 oneShotMonitorsFinalized)252 return;253 oneShotMonitorsFinalized = true;254 captureMonitorTurnsInLocalQueue = false;255 config.getMonitorRegistry().abortAll();256 flushQueuedNotificationsToSdk(sdkOnlyMonitorQueue);257 };258 // EPIPE: don't process.exit here — that bypasses the caller's259 // runExitCleanup → flush() and drops queued JSONL writes. Destroy260 // stdout instead and let the natural return drive cleanup. (Aborting261 // is also wrong: the abort path runs handleCancellationError → exit262 // 130 and re-introduces the same bypass.)263 let pipeBroken = false;264 const stdoutErrorHandler = (err) => {265 if (err.code === 'EPIPE' && !pipeBroken) {266 pipeBroken = true;267 process.stdout.destroy();268 }269 };270 // Setup signal handlers for graceful shutdown271 const shutdownHandler = () => {272 debugLogger.debug('[runNonInteractive] Shutdown signal received');273 abortController.abort();274 };275 // ─── Teammate message queue ─────────────────────────276 // When teammates send messages to the leader, they277 // accumulate here and are drained into the LLM278 // conversation between turns.279 const pendingTeammateMessages = [];280 // Track the manager we're currently bound to so we can281 // detach the leader callback and approval listener before282 // a new manager is installed (or in `finally`). Without283 // this, a reused stream-json session could leave callbacks284 // attached to a stale TeamManager.285 let boundManager = null;286 let approvalListener = null;287 const detachFromManager = (m) => {288 m.setLeaderMessageCallback(null);289 if (approvalListener) {290 m.getEventEmitter().off(TeamEventType.TEAMMATE_APPROVAL_REQUEST, approvalListener);291 approvalListener = null;292 }293 };294 const onTeamManagerChangeHandler = (manager) => {295 // Detach from the previous manager before rebinding.296 if (boundManager && boundManager !== manager) {297 detachFromManager(boundManager);298 }299 boundManager = manager;300 if (manager) {301 manager.setLeaderMessageCallback((formatted) => {302 pendingTeammateMessages.push(formatted);303 });304 // Route teammate tool approvals through the session's305 // permission channel.306 if (options.controlService) {307 // Stream-json mode: SDK handles approvals. Catch instead of308 // void: the handler's own error path re-issues a respond()309 // that can reject (teammate terminated mid-request), and a310 // voided rejection here is an unhandledRejection in an SDK311 // session — mirror the headless listeners below.312 approvalListener = (event) => {313 options314 .controlService.permission.handleTeammateApproval(event)315 .catch((err) => {316 debugLogger.warn('Teammate approval handling failed:', err);317 });318 };319 }320 else {321 // Headless / non-stream-json mode: there is no UI to322 // surface a prompt, so the only safe options are323 // YOLO (auto-approve) or Cancel. Without this fallback324 // listener, the event has no subscriber and the teammate325 // hangs until its 600s stall timeout fires.326 approvalListener = (event) => {327 const mode = config.getApprovalMode();328 if (mode === ApprovalMode.YOLO) {329 // `respond` may reject if the teammate terminates between the330 // approval request and our response — catch it so it doesn't331 // become an unhandledRejection that can crash the process.332 event333 .respond(ToolConfirmationOutcome.ProceedOnce)334 .catch((err) => {335 debugLogger.warn('Teammate approval ProceedOnce failed:', err);336 });337 return;338 }339 // Surface a clear reason on stderr — otherwise the340 // failure looks like the teammate gave up for no reason.341 const reason = `Auto-cancelling tool ${event.toolName} requested by ` +342 `teammate "${event.teammateName}": current approval mode ` +343 `(${mode}) cannot prompt in non-stream-json mode. ` +344 `Use --yolo or stream-json to allow teammate tool calls.`;345 process.stderr.write(`[team] ${reason}\n`);346 // Also surface to the leader's LLM, otherwise it just347 // sees the teammate fail without any signal that an348 // approval was needed and the host couldn't prompt.349 pendingTeammateMessages.push(`<team_notice>\n${reason}\n</team_notice>`);350 event.respond(ToolConfirmationOutcome.Cancel).catch((err) => {351 debugLogger.warn('Teammate approval Cancel failed:', err);352 });353 };354 }355 manager356 .getEventEmitter()357 .on(TeamEventType.TEAMMATE_APPROVAL_REQUEST, approvalListener);358 }359 };360 // First-turn SendMessageType override for continuation turns; null means361 // the regular options.sendMessageType / UserQuery selection applies.362 let continueSendType = null;363 try {364 process.stdout.on('error', stdoutErrorHandler);365 process.on('SIGINT', shutdownHandler);366 process.on('SIGTERM', shutdownHandler);367 config.onTeamManagerChange(onTeamManagerChangeHandler);368 // Handle the case where a manager already exists (e.g.,369 // a follow-up turn in a stream-json session that created370 // a team on a previous turn).371 const existingManager = config.getTeamManager();372 if (existingManager) {373 onTeamManagerChangeHandler(existingManager);374 }375 // Emit systemMessage first (always the first message in JSON mode)376 const systemMessage = await buildSystemMessage(config, sessionId, permissionMode);377 adapter.emitMessage(systemMessage);378 let initialPartList = extractPartsFromUserMessage(options.userMessage);379 // Per-turn model override captured from an inline `/model <id> <prompt>`380 // slash command; seeds the loop-scoped `modelOverride` below so the381 // submitted prompt runs on the chosen model without a session switch.382 let inlineModelOverride;383 if (options.continueInterrupted) {384 // Read the full history, not a bounded tail: the Retry send path in385 // client.ts strips the ENTIRE trailing user run, so detection must386 // re-submit exactly that run or the oldest orphans get dropped. This387 // runs once per (rare) continue request, so the full clone is fine.388 const detection = detectTurnInterruption(geminiClient.getChat().getHistory());389 debugLogger.info('[runNonInteractive] continueInterrupted detection', {390 kind: detection.kind,391 partsCount: detection.kind === 'interrupted_prompt'392 ? detection.parts.length393 : 0,394 danglingCallCount: detection.kind === 'interrupted_turn'395 ? detection.danglingCalls.length396 : 0,397 });398 if (detection.kind === 'none') {399 await emitNonInteractiveFinalMessage({400 message: 'No interrupted turn to continue.',401 isError: false,402 adapter,403 config,404 startTimeMs: startTime,405 });406 return 0;407 }408 if (detection.kind === 'interrupted_prompt') {409 // Re-submit the orphaned user content under Retry semantics. The410 // Retry send path strips the orphaned original from history and411 // restores it if the send never starts, so history is neither412 // duplicated nor lost — no separate strip/restore needed here.413 initialPartList = detection.parts;414 continueSendType = SendMessageType.Retry;415 }416 else {417 initialPartList = buildSyntheticToolResponseParts(detection.danglingCalls, ORPHAN_TOOL_USE_REPAIR_REASON);418 continueSendType = SendMessageType.ToolResult;419 }420 const reminderParts = buildInitialSystemReminders(config);421 if (reminderParts.length > 0 && initialPartList) {422 const continuationParts = normalizePartList(initialPartList);423 const hasSystemReminderPart = continuationParts.some((part) => isSystemReminderContent({ role: 'user', parts: [part] }));424 if (!hasSystemReminderPart) {425 initialPartList = insertAfterFunctionResponses(continuationParts, reminderParts);426 }427 }428 }429 if (!initialPartList) {430 let slashHandled = false;431 if (isSlashCommand(input)) {432 const slashCommandResult = await handleSlashCommand(input, abortController, config, settings);433 switch (slashCommandResult.type) {434 case 'submit_prompt':435 // A slash command can replace the prompt entirely; fall back to @-command processing otherwise.436 initialPartList = slashCommandResult.content;437 // Re-validate provider identity rather than trust the producer:438 // any slash command can set `modelOverride`, so the consumer439 // enforces that it names a model on the active provider before440 // redirecting API calls to it.441 if (slashCommandResult.modelOverride !== undefined &&442 isInlineModelOverrideAllowed(config, slashCommandResult.modelOverride)) {443 inlineModelOverride = slashCommandResult.modelOverride;444 debugLogger.debug(`[runNonInteractive] inline model override captured: ${inlineModelOverride}`);445 }446 else if (slashCommandResult.modelOverride !== undefined) {447 debugLogger.warn(`[runNonInteractive] ignoring model override '${slashCommandResult.modelOverride}': not a model on the active provider`);448 }449 slashHandled = true;450 break;451 case 'message': {452 // systemMessage already emitted above453 await emitNonInteractiveFinalMessage({454 message: slashCommandResult.content,455 isError: slashCommandResult.messageType === 'error',456 adapter,457 config,458 startTimeMs: startTime,459 });460 return slashCommandResult.messageType === 'error' ? 1 : 0;461 }462 case 'stream_messages':463 throw new FatalInputError('Stream messages mode is not supported in non-interactive CLI');464 case 'unsupported': {465 await emitNonInteractiveFinalMessage({466 message: slashCommandResult.reason,467 isError: true,468 adapter,469 config,470 startTimeMs: startTime,471 });472 return 1;473 }474 case 'no_command':475 break;476 default: {477 const _exhaustive = slashCommandResult;478 throw new FatalInputError(`Unhandled slash command result type: ${_exhaustive.type}`);479 }480 }481 }482 if (!slashHandled) {483 const { processedQuery, shouldProceed } = await handleAtCommand({484 query: input,485 config,486 onDebugMessage: () => { },487 messageId: Date.now(),488 signal: abortController.signal,489 });490 if (!shouldProceed || !processedQuery) {491 // An error occurred during @include processing (e.g., file not found).492 // The error message is already logged by handleAtCommand.493 throw new FatalInputError('Exiting due to an error processing the @ command.');494 }495 initialPartList = processedQuery;496 }497 }498 if (!initialPartList) {499 initialPartList = [{ text: input }];500 }501 // Inject a worktree context notice into the model's first prompt.502 // Two sources: the `--worktree` startup flag (set by gemini.tsx503 // before loadCliConfig) takes precedence over the Phase C resume504 // restore. TUI does this via historyManager.addItem(INFO); here in505 // headless we prepend a `<system-reminder>` block since there is506 // no UI history to write into.507 const withReminder = (existing, text) => {508 const reminderPart = {509 text: `<system-reminder>\n${text}\n</system-reminder>\n\n`,510 };511 return Array.isArray(existing)512 ? [reminderPart, ...existing]513 : [reminderPart, existing];514 };515 // Continuation turns must not prepend reminder text: a ToolResult516 // payload's functionResponse parts have to stay at the HEAD of the517 // user message or Anthropic-compatible backends reject the pairing.518 const startupNotice = options.continueInterrupted519 ? null520 : config.consumePendingStartupWorktreeNotice();521 if (startupNotice) {522 initialPartList = withReminder(initialPartList, startupNotice);523 adapter.emitSystemMessage('worktree_started', {524 notice: startupNotice,525 });526 }527 else if (!options.continueInterrupted &&528 config.getResumedSessionData()) {529 try {530 const sessionPath = config531 .getSessionService()532 .getWorktreeSessionPath(sessionId);533 const restored = await restoreWorktreeContext(sessionPath);534 if (restored.contextMessage) {535 initialPartList = withReminder(initialPartList, restored.contextMessage);536 // Surface the notice in the JSON stream so SDK consumers537 // can react to it (logging, UI hints, etc.).538 adapter.emitSystemMessage('worktree_restored', {539 slug: restored.session?.slug,540 path: restored.session?.worktreePath,541 branch: restored.session?.worktreeBranch,542 });543 }544 }545 catch (error) {546 debugLogger.warn(`worktree restore failed (non-fatal):`, error);547 }548 }549 const initialParts = normalizePartList(initialPartList);550 let currentMessages = [{ role: 'user', parts: initialParts }];551 // Register the callback early so background agents launched during the main552 // tool-call chain can push completions onto the queue.553 const registry = config.getBackgroundTaskRegistry();554 registry.setNotificationCallback((displayText, modelText, meta) => {555 localQueue.push({556 displayText,557 modelText,558 sendMessageType: SendMessageType.Notification,559 sdkNotification: {560 task_id: meta.agentId,561 tool_use_id: meta.toolUseId,562 status: meta.status,563 usage: meta.stats564 ? {565 total_tokens: meta.stats.totalTokens,566 tool_uses: meta.stats.toolUses,567 duration_ms: meta.stats.durationMs,568 }569 : undefined,570 },571 });572 });573 registry.setRegisterCallback((entry) => {574 adapter.emitSystemMessage('task_started', {575 task_id: entry.agentId,576 tool_use_id: entry.toolUseId,577 description: entry.description,578 subagent_type: entry.subagentType,579 });580 });581 const monitorRegistry = config.getMonitorRegistry();582 if (options.captureMonitorNotifications !== false) {583 // One-shot headless runs capture monitor notifications locally so any584 // events already emitted before exit can be surfaced to the SDK/model.585 // Persistent stream-json sessions own this callback at the Session586 // layer instead, so future monitor events can continue after the587 // originating turn has already completed.588 monitorRegistry.setNotificationCallback((displayText, modelText, meta) => {589 if (meta.status === 'running' &&590 typeof monitorRegistry.get === 'function') {591 const entry = monitorRegistry.get(meta.monitorId);592 if (!entry || entry.status !== 'running')593 return;594 }595 const queueItem = {596 displayText,597 modelText,598 sendMessageType: SendMessageType.Notification,599 sdkNotification: {600 task_id: meta.monitorId,601 tool_use_id: meta.toolUseId,602 status: meta.status,603 },604 };605 if (captureMonitorTurnsInLocalQueue) {606 localQueue.push(queueItem);607 }608 else {609 sdkOnlyMonitorQueue.push(queueItem);610 flushQueuedNotificationsToSdk(sdkOnlyMonitorQueue);611 }612 });613 }614 if (options.captureMonitorRegistrations !== false) {615 monitorRegistry.setRegisterCallback((entry) => {616 adapter.emitSystemMessage('task_started', {617 task_id: entry.monitorId,618 tool_use_id: entry.toolUseId,619 description: entry.description,620 });621 });622 }623 let isFirstTurn = true;624 let hasUnsentToolResponse = false;625 let modelOverride = inlineModelOverride;626 // An explicit inline `/model <id> <prompt>` override wins for the whole627 // turn: while active, skill-tool `modelOverride` writes (including the628 // undefined-clears case) are skipped so they cannot silently revert the629 // submitted prompt to the session model mid-turn. Unlike useGeminiStream's630 // ref-based `applyModelOverride`/`clearModelOverride` helpers, this is a631 // run-scoped const — non-interactive mode is single-turn, so there is no632 // retry-clearing or skill-tool takeover to guard against, just the633 // within-turn precedence above.634 const inlineModelOverrideActive = inlineModelOverride !== undefined;635 if (inlineModelOverrideActive) {636 debugLogger.debug(`[runNonInteractive] inline model override active for turn: ${inlineModelOverride}`);637 }638 // Session-scoped because the synthetic `structured_output` tool can639 // be invoked from EITHER the main assistant-turn loop or from a640 // drain-turn (queued notification / cron prompt); whichever fires641 // first wins, and both paths need to surface the same structured642 // result envelope.643 let structuredSubmission = undefined;644 // Captures the first ~200 chars of model-emitted plain text across645 // turns. Used only to enrich the --json-schema "produced plain646 // text" error: the user/operator gets a hint of what the model647 // actually said instead of a static, context-free message.648 let plainTextPreview = '';649 const PLAIN_TEXT_PREVIEW_LIMIT = 200;650 let loopDetected = false;651 let loopDetectedMessage = formatLoopDetectedMessage(undefined);652 // Shared terminal block for the structured-output success653 // contract. Both the main-turn loop and the drain-turn post-loop654 // previously reproduced this block verbatim655 // (`registry.abortAll()` → bounded holdback for in-flight656 // background-task `task_notification` events → flush localQueue →657 // finalize one-shot monitors → `adapter.emitResult` → return 0).658 // `finalizeOneShotMonitors` is idempotent (the659 // `oneShotMonitorsFinalized` guard makes the second call a660 // no-op), so unconditional invocation is safe even when the drain661 // path already finalized monitors before reaching here.662 const emitStructuredSuccess = async () => {663 registry.abortAll();664 // `abortAll()` marks each task `cancelled` synchronously, but665 // the matching `task_notification` is emitted later by the666 // task's natural handler. Hold back briefly (capped at667 // STRUCTURED_SHUTDOWN_HOLDBACK_MS) so consumers see every668 // `task_started` paired with its terminal notification, without669 // blocking exit on a slow agent that the user has already670 // declared done.671 const holdbackDeadline = Date.now() + STRUCTURED_SHUTDOWN_HOLDBACK_MS;672 while (Date.now() < holdbackDeadline &&673 registry.hasUnfinalizedTasks()) {674 await new Promise((r) => setTimeout(r, 50));675 }676 flushQueuedNotificationsToSdk(localQueue);677 finalizeOneShotMonitors();678 const metrics = uiTelemetryService.getMetrics();679 const usage = computeUsageFromMetrics(metrics);680 const stats = outputFormat === OutputFormat.JSON681 ? uiTelemetryService.getMetrics()682 : undefined;683 emitResult({684 isError: false,685 durationMs: Date.now() - startTime,686 apiDurationMs: totalApiDurationMs,687 numTurns: turnCount,688 usage,689 stats,690 structuredResult: structuredSubmission,691 });692 return 0;693 };694 const emitLoopDetectedResult = () => {695 registry.abortAll();696 flushQueuedNotificationsToSdk(localQueue);697 finalizeOneShotMonitors();698 if (outputFormat === OutputFormat.TEXT) {699 return 1;700 }701 const metrics = uiTelemetryService.getMetrics();702 const usage = computeUsageFromMetrics(metrics);703 const stats = outputFormat === OutputFormat.JSON704 ? uiTelemetryService.getMetrics()705 : undefined;706 adapter.emitResult({707 isError: true,708 durationMs: Date.now() - startTime,709 apiDurationMs: totalApiDurationMs,710 numTurns: turnCount,711 errorMessage: loopDetectedMessage,712 usage,713 stats,714 });715 return 1;716 };717 /**718 * Shared per-turn tool-call dispatch for the main-turn loop and719 * `drainBatch`. Both call sites used to reproduce ~120 lines of720 * near-identical logic that filtered `structured_output` to its721 * own pre-scan when `--json-schema` is active, executed each722 * request through `executeToolCall`, captured the `structured_output`723 * args into the session-scoped `structuredSubmission`, and724 * synthesised `tool_result` events for every suppressed sibling725 * `tool_use`. The two blocks differed only by variable name726 * prefixes (`requestsToExecute` vs `itemRequestsToExecute`, etc.)727 * and which scope's `modelOverride` to update — passed in as728 * `setModelOverride` so the caller controls binding.729 *730 * The helper mutates the closure-captured `structuredSubmission`731 * directly (it's session-scoped on purpose: whichever turn732 * captures it terminates the run). The caller is responsible for733 * acting on a non-undefined `structuredSubmission` after the734 * helper returns (main-turn → emitStructuredSuccess(); drain-turn735 * → return so the post-drain code emits success).736 */737 const handledProviderToolCallIds = geminiClient.getHistoryFunctionResponseIds();738 // Tracks duplicate-error responses emitted during this headless run.739 // Once a provider id reaches this set, seeing it again is terminal for740 // the current tool batch so we do not send partial tool responses.741 const duplicateProviderToolCallResponseIds = new Set();742 const processToolCallBatch = async (batchRequests, setModelOverride) => {743 const toolResponseParts = [];744 const structuredOutputActive = config.getJsonSchema() &&745 batchRequests.some((r) => r.name === ToolNames.STRUCTURED_OUTPUT);746 const getProviderResponseId = (request) => request.providerCallId ??747 (structuredOutputActive ? undefined : request.callId || undefined);748 const seenBatchCallIds = new Set();749 const duplicateBatchRequests = [];750 const uniqueBatchRequests = batchRequests.filter((request) => {751 if (request.callId) {752 if (seenBatchCallIds.has(request.callId)) {753 if (structuredOutputActive &&754 request.name === ToolNames.STRUCTURED_OUTPUT) {755 return true;756 }757 debugLogger.debug(`Dropping duplicate non-interactive tool callId=${request.callId} name=${request.name}`);758 duplicateBatchRequests.push(request);759 return false;760 }761 seenBatchCallIds.add(request.callId);762 }763 return true;764 });765 const repeatedDuplicateRequest = findRepeatedDuplicateProviderToolCall([...uniqueBatchRequests, ...duplicateBatchRequests], getProviderResponseId, handledProviderToolCallIds, duplicateProviderToolCallResponseIds);766 if (repeatedDuplicateRequest) {767 const providerCallId = repeatedDuplicateRequest.providerCallId ??768 repeatedDuplicateRequest.callId;769 debugLogger.debug(`[runNonInteractive] Dropping batch after repeated duplicate provider tool-call id: ${providerCallId} (tool: ${repeatedDuplicateRequest.name})`);770 return {771 responseParts: [],772 repeatedDuplicateProviderToolCall: true,773 };774 }775 const respondedRequests = new Set();776 const executableBatchRequests = [];777 const duplicatePendingResponses = [];778 for (const requestInfo of uniqueBatchRequests) {779 const providerCallId = getProviderResponseId(requestInfo);780 if (!providerCallId) {781 executableBatchRequests.push(requestInfo);782 continue;783 }784 if (!handledProviderToolCallIds.has(providerCallId)) {785 handledProviderToolCallIds.add(providerCallId);786 executableBatchRequests.push(requestInfo);787 continue;788 }789 markDuplicateProviderToolCallResponseSent(providerCallId, duplicateProviderToolCallResponseIds);790 const toolResponse = createDuplicateProviderToolCallResponse(requestInfo);791 debugLogger.debug(`[runNonInteractive] Suppressing duplicate provider tool-call id: ${providerCallId} (tool: ${requestInfo.name})`);792 respondedRequests.add(requestInfo);793 adapter.emitToolResult(requestInfo, toolResponse);794 duplicatePendingResponses.push(...toolResponse.responseParts);795 }796 // Duplicate responses must always reach the model. They pair with a797 // tool call the provider already emitted, even when structured_output798 // is the only executable sibling in this batch.799 toolResponseParts.push(...duplicatePendingResponses);800 // Pre-scan: when --json-schema is active and the model emitted801 // a `structured_output` call alongside other tools in the same802 // turn, the structured call is the terminal contract. Execute803 // every structured_output in original order until one succeeds,804 // suppress every non-structured sibling. See the multi-shape805 // examples in the main loop's prior comment for the806 // [bad/good/side-effect] permutations.807 let requestsToExecute = executableBatchRequests;808 if (structuredOutputActive) {809 requestsToExecute = executableBatchRequests.filter((r) => r.name === ToolNames.STRUCTURED_OUTPUT);810 }811 const executedRequests = new Set(respondedRequests);812 for (const requestInfo of requestsToExecute) {813 executedRequests.add(requestInfo);814 const inputFormat = typeof config.getInputFormat === 'function'815 ? config.getInputFormat()816 : InputFormat.TEXT;817 const toolCallUpdateCallback = inputFormat === InputFormat.STREAM_JSON && options.controlService818 ? options.controlService.permission.getToolCallUpdateCallback()819 : undefined;820 // Build outputUpdateHandler for this tool call. Agent tool821 // has its own complex handler (subagent messages). All other822 // tools with canUpdateOutput=true (e.g., MCP tools) get a823 // generic handler that emits progress via the adapter.824 const isAgentTool = requestInfo.name === 'agent';825 const { handler: outputUpdateHandler } = isAgentTool826 ? createAgentToolProgressHandler(config, requestInfo.callId, adapter)827 : createToolProgressHandler(requestInfo, adapter);828 // Tick BEFORE the call so that --max-tool-calls=N caps the run829 // at exactly N executions: the (N+1)th tick aborts before the830 // tool runs. Ticking after would let the (N+1)th tool execute831 // and only then abort. See issue #4103.832 //833 // Exempt `structured_output` ONLY when `--json-schema` is834 // active: under --json-schema this is the terminal "I'm done"835 // contract tool, not real work, and counting it would abort836 // an otherwise-valid completion at the budget edge (budget=3,837 // model used 3 tools then emits structured_output as call #4838 // → exit 55 instead of success). Guarding on839 // `getJsonSchema()` keeps the exemption tied to the feature840 // that owns the tool name — an MCP server that registers an841 // unrelated tool literally named `structured_output` would842 // otherwise inherit a free pass.843 //844 // Caveat: failed structured_output calls (Ajv validation845 // failure) also skip the tick, so a model stuck in a846 // validation-retry loop is not bounded by --max-tool-calls.847 // Documented in docs/users/features/headless.md → "Scope".848 // Combine with --max-session-turns or --max-wall-time.849 const isStructuredOutputExempt = requestInfo.name === ToolNames.STRUCTURED_OUTPUT &&850 config.getJsonSchema?.() !== undefined;851 if (!isStructuredOutputExempt) {852 budgetEnforcer.tickToolCall();853 }854 if (abortController.signal.aborted)855 await routeAbort();856 const toolResponse = await executeToolCall(config, requestInfo, abortController.signal, {857 outputUpdateHandler,858 ...(toolCallUpdateCallback && {859 onToolCallsUpdate: toolCallUpdateCallback,860 }),861 });862 if (toolResponse.error) {863 // In JSON/STREAM_JSON mode, tool errors are tolerated and864 // formatted as tool_result blocks. handleToolError detects865 // mode from config and allows the session to continue so866 // the LLM can decide what to do next. In text mode, we867 // still log the error.868 handleToolError(requestInfo.name, toolResponse.error, config, toolResponse.errorType || 'TOOL_EXECUTION_ERROR', typeof toolResponse.resultDisplay === 'string'869 ? toolResponse.resultDisplay870 : undefined);871 }872 adapter.emitToolResult(requestInfo, toolResponse);873 config874 .getGeminiClient()875 .recordCompletedToolCall(requestInfo.name, requestInfo.args);876 if (toolResponse.responseParts) {877 toolResponseParts.push(...toolResponse.responseParts);878 }879 // Capture model override from skill tool results.880 // Use `in` so that undefined (from inherit/no-model skills)881 // clears a prior override, while non-skill tools (field882 // absent) leave the current override intact.883 if ('modelOverride' in toolResponse) {884 setModelOverride(toolResponse.modelOverride);885 }886 if (requestInfo.name === ToolNames.STRUCTURED_OUTPUT &&887 !toolResponse.error) {888 // Honour the "first valid call ends the session" contract.889 // The break is after the responseParts/modelOverride capture890 // above so future changes to SyntheticOutputTool can't891 // silently drop those signals. structuredSubmission is the892 // session-scoped binding from the enclosing scope.893 structuredSubmission = requestInfo.args;894 break;895 }896 }897 // Synthesise tool_result events + retry parts for every898 // tool_use block from the prior assistant message that we did899 // NOT actually execute — non-structured siblings that were900 // suppressed up front, plus any structured_output calls left901 // unexecuted after an earlier one in the batch already902 // succeeded. Runs for both the success and retry paths so the903 // emitted event log pairs every tool_use with a tool_result904 // AND the retry-turn payload (when reached) doesn't leave905 // Anthropic / OpenAI staring at unpaired tool_use blocks.906 const unexecutedCalls = executableBatchRequests.filter((r) => !executedRequests.has(r));907 if (unexecutedCalls.length > 0) {908 const skippedOutput = suppressedOutputBody(structuredSubmission !== undefined);909 for (const call of unexecutedCalls) {910 const responseParts = [911 {912 functionResponse: {913 id: call.callId,914 name: call.name,915 response: { output: skippedOutput },916 },917 },918 ];919 adapter.emitToolResult(call, {920 callId: call.callId,921 responseParts,922 resultDisplay: skippedOutput,923 error: undefined,924 errorType: undefined,925 });926 toolResponseParts.push(...responseParts);927 }928 }929 for (const requestInfo of duplicateBatchRequests) {930 const providerCallId = getProviderResponseId(requestInfo);931 if (!providerCallId)932 continue;933 markDuplicateProviderToolCallResponseSent(providerCallId, duplicateProviderToolCallResponseIds);934 const toolResponse = createDuplicateProviderToolCallResponse(requestInfo);935 adapter.emitToolResult(requestInfo, toolResponse);936 toolResponseParts.push(...toolResponse.responseParts);937 }938 return {939 responseParts: toolResponseParts,940 repeatedDuplicateProviderToolCall: false,941 };942 };943 while (true) {944 // Drain pending teammate messages into the conversation.945 // sendMessageStream only reads currentMessages[0].parts,946 // so teammate text must be merged into that same parts947 // array to avoid being silently dropped.948 // Skip on the first turn to avoid replacing the user's949 // initial query — early teammate messages will be picked950 // up on the next iteration.951 let isTeammateTurn = false;952 if (!isFirstTurn && pendingTeammateMessages.length > 0) {953 const batch = pendingTeammateMessages.splice(0);954 const teammatePart = { text: batch.join('\n\n') };955 if (hasUnsentToolResponse && currentMessages[0]) {956 currentMessages[0].parts = [957 ...(currentMessages[0].parts || []),958 teammatePart,959 ];960 }961 else {962 currentMessages = [{ role: 'user', parts: [teammatePart] }];963 }964 // Treat BOTH the standalone and the merged-into-tool-response965 // cases as a teammate turn. Teammate text is fresh external966 // input, so the loop detector must reset — otherwise a leader967 // that polls task_list while teammate messages keep merging968 // into its tool-response turns climbs the identical-tool-call969 // counter and trips a false LoopDetected. The Teammate send970 // path prepends nothing to the request, so a merged turn's971 // leading functionResponse parts stay paired with their972 // functionCall.973 isTeammateTurn = true;974 }975 hasUnsentToolResponse = false;976 turnCount++;977 if (config.getMaxSessionTurns() >= 0 &&978 turnCount > config.getMaxSessionTurns()) {979 await handleMaxTurnsExceededError(config);980 }981 let sendType;982 if (isFirstTurn) {983 sendType =984 continueSendType ??985 options.sendMessageType ??986 SendMessageType.UserQuery;987 }988 else if (isTeammateTurn) {989 sendType = SendMessageType.Teammate;990 }991 else {992 sendType = SendMessageType.ToolResult;993 }994 const toolCallRequests = [];995 const apiStartTime = Date.now();996 const responseStream = geminiClient.sendMessageStream(currentMessages[0]?.parts || [], abortController.signal, prompt_id, {997 type: sendType,998 modelOverride,999 ...(isFirstTurn &&1000 options.notificationDisplayText && {1001 notificationDisplayText: options.notificationDisplayText,1002 }),1003 });1004 isFirstTurn = false;1005 // Start assistant message for this turn1006 adapter.startAssistantMessage();1007 for await (const event of responseStream) {1008 if (abortController.signal.aborted) {1009 // Pair the startAssistantMessage() above so stream-json mode1010 // doesn't leave an unterminated message_start when a budget /1011 // SIGINT abort lands mid-stream. Symmetric with the drain-item1012 // loop fix below.1013 adapter.finalizeAssistantMessage();1014 await routeAbort();1015 }1016 // Use adapter for all event processing1017 adapter.processEvent(event);1018 if (event.type === GeminiEventType.ToolCallRequest) {1019 toolCallRequests.push(event.value);1020 }1021 if (event.type === GeminiEventType.ModelFallback) {1022 toolCallRequests.length = 0;1023 }1024 if (event.type === GeminiEventType.Content &&1025 plainTextPreview.length < PLAIN_TEXT_PREVIEW_LIMIT) {1026 const remaining = PLAIN_TEXT_PREVIEW_LIMIT - plainTextPreview.length;1027 plainTextPreview += String(event.value).slice(0, remaining);1028 }1029 if (event.type === GeminiEventType.LoopDetected) {1030 if (!loopDetected) {1031 loopDetectedMessage = emitLoopDetectedMessage(config, event.value?.loopType);1032 }1033 loopDetected = true;1034 }1035 if (outputFormat === OutputFormat.TEXT &&1036 event.type === GeminiEventType.Error) {1037 const errorText = parseAndFormatApiError(event.value.error, config.getContentGeneratorConfig()?.authType);1038 process.stderr.write(`${errorText}\n`);1039 // We have already formatted and written the message; mark the1040 // throw so the top-level handleError doesn't reformat (which1041 // would yield "[API Error: [API Error: ...]]") or print it a1042 // second time. Exit code stays 1 — same as before.1043 throw new AlreadyReportedError(errorText);1044 }1045 }1046 // Finalize assistant message1047 adapter.finalizeAssistantMessage();1048 totalApiDurationMs += Date.now() - apiStartTime;1049 if (loopDetected) {1050 return emitLoopDetectedResult();1051 }1052 if (toolCallRequests.length > 0) {1053 // Dispatch the per-turn tool-call batch through the shared1054 // helper (see processToolCallBatch above). The helper handles1055 // the `--json-schema` pre-scan, executes each request, writes1056 // the first valid `structured_output` call's args into the1057 // session-scoped `structuredSubmission`, and synthesises1058 // tool_result events for every suppressed sibling. The1059 // `modelOverride` setter is the only call-site-specific1060 // binding — the main turn updates the session-scoped1061 // `modelOverride` so the next turn's sendMessageStream sees1062 // it; the drain turn updates a per-item `itemModelOverride`1063 // scoped to that drain item.1064 const { responseParts: toolResponseParts, repeatedDuplicateProviderToolCall, } = await processToolCallBatch(toolCallRequests, (override) => {1065 if (!inlineModelOverrideActive) {1066 modelOverride = override;1067 }1068 });1069 if (structuredSubmission !== undefined) {1070 // Single-shot terminal contract; aborts in-flight background1071 // agents, holds back briefly for their terminal1072 // task_notification events to land, then emits the1073 // structured success envelope. Same helper as the drain-turn1074 // post-loop branch — see emitStructuredSuccess above.1075 return emitStructuredSuccess();1076 }1077 if (repeatedDuplicateProviderToolCall &&1078 toolResponseParts.length === 0) {1079 loopDetectedMessage = emitLoopDetectedMessage(config, LoopType.GLOBAL_TOOL_CALL_DUPLICATE);1080 return emitLoopDetectedResult();1081 }1082 currentMessages = [{ role: 'user', parts: toolResponseParts }];1083 hasUnsentToolResponse = true;1084 }1085 else {1086 // No more tool calls — check if teammates are active.1087 const teamManager = config.getTeamManager();1088 if (teamManager?.hasActiveTeammates()) {1089 // If all remaining teammates are stalled, abort them,1090 // inject a final status, and let the leader wrap up.1091 if (teamManager.allRemainingStalled()) {1092 teamManager.abortStalledTeammates();1093 const status = teamManager.buildTeamStatusSummary();1094 pendingTeammateMessages.push(status);1095 continue;1096 }1097 // Wait for messages or termination. On timeout,1098 // wait again — don't inject status summaries that1099 // cause the leader to poll task_list in a loop.1100 // Only break out when a real message arrives or1101 // all teammates finish.1102 while (teamManager.hasActiveTeammates() &&1103 !abortController.signal.aborted) {1104 if (pendingTeammateMessages.length > 0) {1105 break;1106 }1107 if (teamManager.allRemainingStalled()) {1108 teamManager.abortStalledTeammates();1109 const status = teamManager.buildTeamStatusSummary();1110 pendingTeammateMessages.push(status);1111 break;1112 }1113 const waitResult = await teamManager.waitForTeammateActivity(undefined, abortController.signal);1114 // Without this log a per-call 120s timeout silently1115 // retries until the 600s stall threshold trips —1116 // making "teammate stuck" debugging painful in1117 // production. `terminated`/`aborted` exit on their1118 // own through the loop conditions, so logging1119 // `timeout` is enough.1120 if (waitResult === 'timeout') {1121 debugLogger.warn('[runNonInteractive] waitForTeammateActivity timed ' +1122 'out (120s); will continue waiting until stall ' +1123 'threshold or messages arrive.');1124 }1125 }1126 // Drain messages and loop back.1127 if (pendingTeammateMessages.length > 0) {1128 continue;1129 }1130 // All terminated with no messages — fall through.1131 }1132 // If the session was aborted (e.g. Ctrl+C), stop1133 // immediately instead of falling through to the1134 // success path.1135 if (abortController.signal.aborted) {1136 await handleCancellationError(config);1137 }1138 // Force one final inbox drain before deciding to exit.1139 // A teammate may have written its final send_message1140 // and gone IDLE between the last 500ms poll and now —1141 // without this, that message is lost.1142 if (teamManager) {1143 await teamManager.drainLeaderInbox();1144 }1145 // Also drain any final teammate messages.1146 if (pendingTeammateMessages.length > 0) {1147 continue;1148 }1149 // Drain-turns count toward getMaxSessionTurns() for symmetry with the main1150 // loop — otherwise a looping cron or a model that keeps replying to1151 // notifications could exceed the cap silently in headless runs.1152 const drainBatch = async () => {1153 if (localQueue.length === 0)1154 return;1155 // Batch-drain: take contiguous same-type items from the front1156 // of the queue. Cron prompts run individually — each needs its1157 // own slash/shell/@ preprocessing and approval cycle.1158 const targetType = localQueue[0].sendMessageType;1159 let splitIdx = targetType === SendMessageType.Cron ? 1 : 0;1160 if (splitIdx === 0) {1161 while (splitIdx < localQueue.length &&1162 localQueue[splitIdx].sendMessageType === targetType) {1163 splitIdx++;1164 }1165 }1166 const batch = localQueue.splice(0, splitIdx);1167 if (batch.length === 0)1168 return;1169 for (const queueItem of batch) {1170 emitNotificationToSdk(queueItem);1171 }1172 const item = {1173 displayText: batch.map((i) => i.displayText).join('; '),1174 modelText: batch.map((i) => i.modelText).join('\n\n'),1175 sendMessageType: targetType,1176 };1177 turnCount++;1178 if (config.getMaxSessionTurns() >= 0 &&1179 turnCount > config.getMaxSessionTurns()) {1180 await handleMaxTurnsExceededError(config);1181 }1182 let itemMessages = [1183 { role: 'user', parts: [{ text: item.modelText }] },1184 ];1185 let itemIsFirstTurn = true;1186 let itemModelOverride;1187 while (true) {1188 const itemToolCallRequests = [];1189 const itemApiStartTime = Date.now();1190 const itemStream = geminiClient.sendMessageStream(itemMessages[0]?.parts || [], abortController.signal, prompt_id, {1191 type: itemIsFirstTurn1192 ? item.sendMessageType1193 : SendMessageType.ToolResult,1194 modelOverride: itemModelOverride,1195 ...(itemIsFirstTurn && {1196 notificationDisplayText: item.displayText,1197 }),1198 });1199 itemIsFirstTurn = false;1200 adapter.startAssistantMessage();