MediaMetz/gemini-cli
1
0
Fork 0
mirror of https://github.com/google-gemini/gemini-cli.git synced 2026-04-20 10:10:56 -07:00
Code Issues Packages Projects Releases Wiki Activity

feat(cli): migrate nonInteractiveCli to LegacyAgentSession (#22987)

Browse Source
This commit is contained in:
Adam Weidman
2026-04-02 16:21:40 -04:00
committed by GitHub
parent 7c3469713d
commit 6fb58bd31f
15 changed files with 3241 additions and 106 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/cli/src/__snapshots__/nonInteractiveCliAgentSession.test.ts.snap
+35
View File
@@ -0,0 +1,35 @@
// Vitest Snapshot v1, https://vitest.dev/guide/snapshot.html
exports[`runNonInteractive > should emit appropriate error event in streaming JSON mode: 'loop detected' 1`] = `
"{"type":"init","timestamp":"<TIMESTAMP>","session_id":"test-session-id","model":"test-model"}
{"type":"message","timestamp":"<TIMESTAMP>","role":"user","content":"Loop test"}
{"type":"error","timestamp":"<TIMESTAMP>","severity":"warning","message":"Loop detected, stopping execution"}
{"type":"result","timestamp":"<TIMESTAMP>","status":"success","stats":{"total_tokens":0,"input_tokens":0,"output_tokens":0,"cached":0,"input":0,"duration_ms":<DURATION>,"tool_calls":0,"models":{}}}
"
`;
exports[`runNonInteractive > should emit appropriate error event in streaming JSON mode: 'max session turns' 1`] = `
"{"type":"init","timestamp":"<TIMESTAMP>","session_id":"test-session-id","model":"test-model"}
{"type":"message","timestamp":"<TIMESTAMP>","role":"user","content":"Max turns test"}
{"type":"error","timestamp":"<TIMESTAMP>","severity":"error","message":"Maximum session turns exceeded"}
{"type":"result","timestamp":"<TIMESTAMP>","status":"success","stats":{"total_tokens":0,"input_tokens":0,"output_tokens":0,"cached":0,"input":0,"duration_ms":<DURATION>,"tool_calls":0,"models":{}}}
"
`;
exports[`runNonInteractive > should emit appropriate events for streaming JSON output 1`] = `
"{"type":"init","timestamp":"<TIMESTAMP>","session_id":"test-session-id","model":"test-model"}
{"type":"message","timestamp":"<TIMESTAMP>","role":"user","content":"Stream test"}
{"type":"message","timestamp":"<TIMESTAMP>","role":"assistant","content":"Thinking...","delta":true}
{"type":"tool_use","timestamp":"<TIMESTAMP>","tool_name":"testTool","tool_id":"tool-1","parameters":{"arg1":"value1"}}
{"type":"tool_result","timestamp":"<TIMESTAMP>","tool_id":"tool-1","status":"success","output":"Tool executed successfully"}
{"type":"message","timestamp":"<TIMESTAMP>","role":"assistant","content":"Final answer","delta":true}
{"type":"result","timestamp":"<TIMESTAMP>","status":"success","stats":{"total_tokens":0,"input_tokens":0,"output_tokens":0,"cached":0,"input":0,"duration_ms":<DURATION>,"tool_calls":0,"models":{}}}
"
`;
exports[`runNonInteractive > should write a single newline between sequential text outputs from the model 1`] = `
"Use mock tool
Use mock tool again
Finished.
"
`;
packages/cli/src/nonInteractiveCli.test.ts
+2 -1
View File
@@ -166,7 +166,7 @@ describe('runNonInteractive', () => {
};
mockConfig = {
initialize: vi.fn().mockResolvedValue(undefined),
initialize: vi.fn().mockReturnValue(Promise.resolve(undefined)),
getMessageBus: vi.fn().mockReturnValue({
subscribe: vi.fn(),
unsubscribe: vi.fn(),
@@ -190,6 +190,7 @@ describe('runNonInteractive', () => {
isTrustedFolder: vi.fn().mockReturnValue(false),
getRawOutput: vi.fn().mockReturnValue(false),
getAcceptRawOutputRisk: vi.fn().mockReturnValue(false),
getAgentSessionNoninteractiveEnabled: vi.fn().mockReturnValue(false),
} as unknown as Config;
mockSettings = {
packages/cli/src/nonInteractiveCli.ts
+11 -7
View File
@@ -46,6 +46,7 @@ import {
handleMaxTurnsExceededError,
} from './utils/errors.js';
import { TextOutput } from './ui/utils/textOutput.js';
import { runNonInteractive as runNonInteractiveAgentSession } from './nonInteractiveCliAgentSession.js';
interface RunNonInteractiveParams {
config: Config;
@@ -55,13 +56,16 @@ interface RunNonInteractiveParams {
resumedSessionData?: ResumedSessionData;
}
export async function runNonInteractive({
config,
settings,
input,
prompt_id,
resumedSessionData,
}: RunNonInteractiveParams): Promise<void> {
export async function runNonInteractive(
params: RunNonInteractiveParams,
): Promise<void> {
const useAgentSession = params.config.getAgentSessionNoninteractiveEnabled();
if (useAgentSession) {
return runNonInteractiveAgentSession(params);
}
const { config, settings, input, prompt_id, resumedSessionData } = params;
return promptIdContext.run(prompt_id, async () => {
const consolePatcher = new ConsolePatcher({
stderr: true,
packages/cli/src/nonInteractiveCliAgentSession.test.ts
+2436
View File
File diff suppressed because it is too large Load Diff
packages/cli/src/nonInteractiveCliAgentSession.ts
+621
View File
@@ -0,0 +1,621 @@
/**
* @license
* Copyright 2025 Google LLC
* SPDX-License-Identifier: Apache-2.0
*/
import type {
Config,
ResumedSessionData,
UserFeedbackPayload,
AgentEvent,
ContentPart,
} from '@google/gemini-cli-core';
import { isSlashCommand } from './ui/utils/commandUtils.js';
import type { LoadedSettings } from './config/settings.js';
import {
convertSessionToClientHistory,
FatalError,
FatalAuthenticationError,
FatalInputError,
FatalSandboxError,
FatalConfigError,
FatalTurnLimitedError,
FatalToolExecutionError,
FatalCancellationError,
promptIdContext,
OutputFormat,
JsonFormatter,
StreamJsonFormatter,
JsonStreamEventType,
uiTelemetryService,
coreEvents,
CoreEvent,
createWorkingStdio,
Scheduler,
ROOT_SCHEDULER_ID,
LegacyAgentSession,
ToolErrorType,
geminiPartsToContentParts,
} from '@google/gemini-cli-core';
import type { Part } from '@google/genai';
import readline from 'node:readline';
import stripAnsi from 'strip-ansi';
import { handleSlashCommand } from './nonInteractiveCliCommands.js';
import { ConsolePatcher } from './ui/utils/ConsolePatcher.js';
import { handleAtCommand } from './ui/hooks/atCommandProcessor.js';
import { handleError, handleToolError } from './utils/errors.js';
import { TextOutput } from './ui/utils/textOutput.js';
interface RunNonInteractiveParams {
config: Config;
settings: LoadedSettings;
input: string;
prompt_id: string;
resumedSessionData?: ResumedSessionData;
}
export async function runNonInteractive({
config,
settings,
input,
prompt_id,
resumedSessionData,
}: RunNonInteractiveParams): Promise<void> {
return promptIdContext.run(prompt_id, async () => {
const consolePatcher = new ConsolePatcher({
stderr: true,
interactive: false,
debugMode: config.getDebugMode(),
onNewMessage: (msg) => {
coreEvents.emitConsoleLog(msg.type, msg.content);
},
});
if (process.env['GEMINI_CLI_ACTIVITY_LOG_TARGET']) {
const { setupInitialActivityLogger } = await import(
'./utils/devtoolsService.js'
);
await setupInitialActivityLogger(config);
}
const { stdout: workingStdout } = createWorkingStdio();
const textOutput = new TextOutput(workingStdout);
const handleUserFeedback = (payload: UserFeedbackPayload) => {
const prefix = payload.severity.toUpperCase();
process.stderr.write(`[${prefix}] ${payload.message}\n`);
if (payload.error && config.getDebugMode()) {
const errorToLog =
payload.error instanceof Error
? payload.error.stack || payload.error.message
: String(payload.error);
process.stderr.write(`${errorToLog}\n`);
}
};
const startTime = Date.now();
const streamFormatter =
config.getOutputFormat() === OutputFormat.STREAM_JSON
? new StreamJsonFormatter()
: null;
const abortController = new AbortController();
// Track cancellation state
let isAborting = false;
let cancelMessageTimer: NodeJS.Timeout | null = null;
// Setup stdin listener for Ctrl+C detection
let stdinWasRaw = false;
let rl: readline.Interface | null = null;
const setupStdinCancellation = () => {
// Only setup if stdin is a TTY (user can interact)
if (!process.stdin.isTTY) {
return;
}
// Save original raw mode state
stdinWasRaw = process.stdin.isRaw || false;
// Enable raw mode to capture individual keypresses
process.stdin.setRawMode(true);
process.stdin.resume();
// Setup readline to emit keypress events
rl = readline.createInterface({
input: process.stdin,
escapeCodeTimeout: 0,
});
readline.emitKeypressEvents(process.stdin, rl);
// Listen for Ctrl+C
const keypressHandler = (
str: string,
key: { name?: string; ctrl?: boolean },
) => {
// Detect Ctrl+C: either ctrl+c key combo or raw character code 3
if ((key && key.ctrl && key.name === 'c') || str === '\u0003') {
// Only handle once
if (isAborting) {
return;
}
isAborting = true;
// Only show message if cancellation takes longer than 200ms
// This reduces verbosity for fast cancellations
cancelMessageTimer = setTimeout(() => {
process.stderr.write('\nCancelling...\n');
}, 200);
abortController.abort();
}
};
process.stdin.on('keypress', keypressHandler);
};
const cleanupStdinCancellation = () => {
// Clear any pending cancel message timer
if (cancelMessageTimer) {
clearTimeout(cancelMessageTimer);
cancelMessageTimer = null;
}
// Cleanup readline and stdin listeners
if (rl) {
rl.close();
rl = null;
}
// Remove keypress listener
process.stdin.removeAllListeners('keypress');
// Restore stdin to original state
if (process.stdin.isTTY) {
process.stdin.setRawMode(stdinWasRaw);
process.stdin.pause();
}
};
let errorToHandle: unknown | undefined;
let abortSession = () => {};
try {
consolePatcher.patch();
if (
config.getRawOutput() &&
!config.getAcceptRawOutputRisk() &&
config.getOutputFormat() === OutputFormat.TEXT
) {
process.stderr.write(
'[WARNING] --raw-output is enabled. Model output is not sanitized and may contain harmful ANSI sequences (e.g. for phishing or command injection). Use --accept-raw-output-risk to suppress this warning.\n',
);
}
// Setup stdin cancellation listener
setupStdinCancellation();
coreEvents.on(CoreEvent.UserFeedback, handleUserFeedback);
coreEvents.drainBacklogs();
// Handle EPIPE errors when the output is piped to a command that closes early.
process.stdout.on('error', (err: NodeJS.ErrnoException) => {
if (err.code === 'EPIPE') {
// Exit gracefully if the pipe is closed.
cleanupStdinCancellation();
consolePatcher.cleanup();
process.exit(0);
}
});
const geminiClient = config.getGeminiClient();
const scheduler = new Scheduler({
context: config,
messageBus: config.getMessageBus(),
getPreferredEditor: () => undefined,
schedulerId: ROOT_SCHEDULER_ID,
});
// Initialize chat. Resume if resume data is passed.
if (resumedSessionData) {
await geminiClient.resumeChat(
convertSessionToClientHistory(
resumedSessionData.conversation.messages,
),
resumedSessionData,
);
}
// Emit init event for streaming JSON
if (streamFormatter) {
streamFormatter.emitEvent({
type: JsonStreamEventType.INIT,
timestamp: new Date().toISOString(),
session_id: config.getSessionId(),
model: config.getModel(),
});
}
let query: Part[] | undefined;
if (isSlashCommand(input)) {
const slashCommandResult = await handleSlashCommand(
input,
abortController,
config,
settings,
);
if (slashCommandResult) {
// eslint-disable-next-line @typescript-eslint/no-unsafe-type-assertion
query = slashCommandResult as Part[];
}
}
if (!query) {
const { processedQuery, error } = await handleAtCommand({
query: input,
config,
addItem: (_item, _timestamp) => 0,
onDebugMessage: () => {},
messageId: Date.now(),
signal: abortController.signal,
escapePastedAtSymbols: false,
});
if (error || !processedQuery) {
throw new FatalInputError(
error || 'Exiting due to an error processing the @ command.',
);
}
// eslint-disable-next-line @typescript-eslint/no-unsafe-type-assertion
query = processedQuery as Part[];
}
// Emit user message event for streaming JSON
if (streamFormatter) {
streamFormatter.emitEvent({
type: JsonStreamEventType.MESSAGE,
timestamp: new Date().toISOString(),
role: 'user',
content: input,
});
}
// Create LegacyAgentSession — owns the agentic loop
const session = new LegacyAgentSession({
client: geminiClient,
scheduler,
config,
promptId: prompt_id,
});
// Wire Ctrl+C to session abort
abortSession = () => {
void session.abort();
};
abortController.signal.addEventListener('abort', abortSession);
if (abortController.signal.aborted) {
throw new FatalCancellationError('Operation cancelled.');
}
// Start the agentic loop (runs in background)
const { streamId } = await session.send({
message: {
content: geminiPartsToContentParts(query),
displayContent: input,
},
});
if (streamId === null) {
throw new Error(
'LegacyAgentSession.send() unexpectedly returned no stream for a message send.',
);
}
const getTextContent = (parts?: ContentPart[]): string | undefined => {
const text = parts
?.map((part) => (part.type === 'text' ? part.text : ''))
.join('');
return text ? text : undefined;
};
const emitFinalSuccessResult = (): void => {
if (streamFormatter) {
const metrics = uiTelemetryService.getMetrics();
const durationMs = Date.now() - startTime;
streamFormatter.emitEvent({
type: JsonStreamEventType.RESULT,
timestamp: new Date().toISOString(),
status: 'success',
stats: streamFormatter.convertToStreamStats(metrics, durationMs),
});
} else if (config.getOutputFormat() === OutputFormat.JSON) {
const formatter = new JsonFormatter();
const stats = uiTelemetryService.getMetrics();
textOutput.write(
formatter.format(config.getSessionId(), responseText, stats),
);
} else {
textOutput.ensureTrailingNewline();
}
};
const reconstructFatalError = (event: AgentEvent<'error'>): Error => {
const errorMeta = event._meta;
const name =
typeof errorMeta?.['errorName'] === 'string'
? errorMeta['errorName']
: undefined;
let errToThrow: Error;
switch (name) {
case 'FatalAuthenticationError':
errToThrow = new FatalAuthenticationError(event.message);
break;
case 'FatalInputError':
errToThrow = new FatalInputError(event.message);
break;
case 'FatalSandboxError':
errToThrow = new FatalSandboxError(event.message);
break;
case 'FatalConfigError':
errToThrow = new FatalConfigError(event.message);
break;
case 'FatalTurnLimitedError':
errToThrow = new FatalTurnLimitedError(event.message);
break;
case 'FatalToolExecutionError':
errToThrow = new FatalToolExecutionError(event.message);
break;
case 'FatalCancellationError':
errToThrow = new FatalCancellationError(event.message);
break;
case 'FatalError':
errToThrow = new FatalError(
event.message,
typeof errorMeta?.['exitCode'] === 'number'
? errorMeta['exitCode']
: 1,
);
break;
default:
errToThrow = new Error(event.message);
if (name) {
Object.defineProperty(errToThrow, 'name', {
value: name,
enumerable: true,
});
}
break;
}
if (errorMeta?.['exitCode'] !== undefined) {
Object.defineProperty(errToThrow, 'exitCode', {
value: errorMeta['exitCode'],
enumerable: true,
});
}
if (errorMeta?.['code'] !== undefined) {
Object.defineProperty(errToThrow, 'code', {
value: errorMeta['code'],
enumerable: true,
});
}
if (errorMeta?.['status'] !== undefined) {
Object.defineProperty(errToThrow, 'status', {
value: errorMeta['status'],
enumerable: true,
});
}
return errToThrow;
};
// Consume AgentEvents for output formatting
let responseText = '';
let preToolResponseText: string | undefined;
let streamEnded = false;
for await (const event of session.stream({ streamId })) {
if (streamEnded) break;
switch (event.type) {
case 'message': {
if (event.role === 'agent') {
for (const part of event.content) {
if (part.type === 'text') {
const isRaw =
config.getRawOutput() || config.getAcceptRawOutputRisk();
const output = isRaw ? part.text : stripAnsi(part.text);
if (streamFormatter) {
streamFormatter.emitEvent({
type: JsonStreamEventType.MESSAGE,
timestamp: new Date().toISOString(),
role: 'assistant',
content: output,
delta: true,
});
} else if (config.getOutputFormat() === OutputFormat.JSON) {
responseText += output;
} else {
if (part.text) {
textOutput.write(output);
}
}
}
}
}
break;
}
case 'tool_request': {
if (config.getOutputFormat() === OutputFormat.JSON) {
// Final JSON output should reflect the last assistant answer after
// any tool orchestration, not intermediate pre-tool text.
preToolResponseText = responseText || preToolResponseText;
responseText = '';
}
if (streamFormatter) {
streamFormatter.emitEvent({
type: JsonStreamEventType.TOOL_USE,
timestamp: new Date().toISOString(),
tool_name: event.name,
tool_id: event.requestId,
parameters: event.args,
});
}
break;
}
case 'tool_response': {
textOutput.ensureTrailingNewline();
if (streamFormatter) {
const displayText = getTextContent(event.displayContent);
const errorMsg = getTextContent(event.content) ?? 'Tool error';
streamFormatter.emitEvent({
type: JsonStreamEventType.TOOL_RESULT,
timestamp: new Date().toISOString(),
tool_id: event.requestId,
status: event.isError ? 'error' : 'success',
output: displayText,
error: event.isError
? {
type:
typeof event.data?.['errorType'] === 'string'
? event.data['errorType']
: 'TOOL_EXECUTION_ERROR',
message: errorMsg,
}
: undefined,
});
}
if (event.isError) {
const displayText = getTextContent(event.displayContent);
const errorMsg = getTextContent(event.content) ?? 'Tool error';
if (event.data?.['errorType'] === ToolErrorType.STOP_EXECUTION) {
if (
config.getOutputFormat() === OutputFormat.JSON &&
!responseText &&
preToolResponseText
) {
responseText = preToolResponseText;
}
const stopMessage = `Agent execution stopped: ${errorMsg}`;
if (config.getOutputFormat() === OutputFormat.TEXT) {
process.stderr.write(`${stopMessage}\n`);
}
}
if (event.data?.['errorType'] === ToolErrorType.NO_SPACE_LEFT) {
throw new FatalToolExecutionError(
'Error executing tool ' +
event.name +
': ' +
(displayText || errorMsg),
);
}
handleToolError(
event.name,
new Error(errorMsg),
config,
typeof event.data?.['errorType'] === 'string'
? event.data['errorType']
: undefined,
displayText,
);
}
break;
}
case 'error': {
if (event.fatal) {
throw reconstructFatalError(event);
}
const errorCode = event._meta?.['code'];
if (errorCode === 'AGENT_EXECUTION_BLOCKED') {
if (config.getOutputFormat() === OutputFormat.TEXT) {
process.stderr.write(`[WARNING] ${event.message}\n`);
}
break;
}
const severity =
event.status === 'RESOURCE_EXHAUSTED' ? 'error' : 'warning';
if (config.getOutputFormat() === OutputFormat.TEXT) {
process.stderr.write(`[WARNING] ${event.message}\n`);
}
if (streamFormatter) {
streamFormatter.emitEvent({
type: JsonStreamEventType.ERROR,
timestamp: new Date().toISOString(),
severity,
message: event.message,
});
}
break;
}
case 'agent_end': {
if (event.reason === 'aborted') {
throw new FatalCancellationError('Operation cancelled.');
} else if (event.reason === 'max_turns') {
const isConfiguredTurnLimit =
typeof event.data?.['maxTurns'] === 'number' ||
typeof event.data?.['turnCount'] === 'number';
if (isConfiguredTurnLimit) {
throw new FatalTurnLimitedError(
'Reached max session turns for this session. Increase the number of turns by specifying maxSessionTurns in settings.json.',
);
} else if (streamFormatter) {
streamFormatter.emitEvent({
type: JsonStreamEventType.ERROR,
timestamp: new Date().toISOString(),
severity: 'error',
message: 'Maximum session turns exceeded',
});
}
}
const stopMessage =
typeof event.data?.['message'] === 'string'
? event.data['message']
: '';
if (stopMessage && config.getOutputFormat() === OutputFormat.TEXT) {
process.stderr.write(`Agent execution stopped: ${stopMessage}\n`);
}
emitFinalSuccessResult();
streamEnded = true;
break;
}
case 'initialize':
case 'session_update':
case 'agent_start':
case 'tool_update':
case 'elicitation_request':
case 'elicitation_response':
case 'usage':
case 'custom':
// Explicitly ignore these non-interactive events
break;
default:
event satisfies never;
break;
}
}
} catch (error) {
errorToHandle = error;
} finally {
// Cleanup stdin cancellation before other cleanup
cleanupStdinCancellation();
abortController.signal.removeEventListener('abort', abortSession);
consolePatcher.cleanup();
coreEvents.off(CoreEvent.UserFeedback, handleUserFeedback);
}
if (errorToHandle) {
handleError(errorToHandle, config);
}
});
}
packages/cli/src/utils/errors.ts
+1 -1
View File
@@ -18,8 +18,8 @@ import {
isFatalToolError,
debugLogger,
coreEvents,
getErrorMessage,
getErrorType,
getErrorMessage,
} from '@google/gemini-cli-core';
import { runSyncCleanup } from './cleanup.js';
packages/core/src/agent/agent-session.test.ts
+19 -7
View File
@@ -7,7 +7,19 @@
import { describe, expect, it } from 'vitest';
import { AgentSession } from './agent-session.js';
import { MockAgentProtocol } from './mock.js';
import type { AgentEvent } from './types.js';
import type { AgentEvent, AgentSend } from './types.js';
function makeMessageSend(
text: string,
displayContent?: string,
): Extract<AgentSend, { message: unknown }> {
return {
message: {
content: [{ type: 'text', text }],
...(displayContent ? { displayContent } : {}),
},
};
}
describe('AgentSession', () => {
it('should passthrough simple methods', async () => {
@@ -51,7 +63,7 @@ describe('AgentSession', () => {
const events: AgentEvent[] = [];
for await (const event of session.sendStream({
message: [{ type: 'text', text: 'hi' }],
...makeMessageSend('hi'),
})) {
events.push(event);
}
@@ -139,7 +151,7 @@ describe('AgentSession', () => {
const events: AgentEvent[] = [];
for await (const event of session.sendStream({
message: [{ type: 'text', text: 'hi' }],
...makeMessageSend('hi'),
})) {
events.push(event);
}
@@ -178,7 +190,7 @@ describe('AgentSession', () => {
protocol.pushResponse([{ type: 'message' }]);
const { streamId } = await session.send({
message: [{ type: 'text', text: 'request' }],
...makeMessageSend('request'),
});
await new Promise((resolve) => setTimeout(resolve, 10));
@@ -242,7 +254,7 @@ describe('AgentSession', () => {
},
]);
await session.send({
message: [{ type: 'text', text: 'request' }],
...makeMessageSend('request'),
});
await new Promise((resolve) => setTimeout(resolve, 10));
@@ -303,7 +315,7 @@ describe('AgentSession', () => {
},
]);
const { streamId: streamId1 } = await session.send({
message: [{ type: 'text', text: 'first request' }],
...makeMessageSend('first request'),
});
await new Promise((resolve) => setTimeout(resolve, 10));
@@ -315,7 +327,7 @@ describe('AgentSession', () => {
},
]);
await session.send({
message: [{ type: 'text', text: 'second request' }],
...makeMessageSend('second request'),
});
await new Promise((resolve) => setTimeout(resolve, 10));
packages/core/src/agent/event-translator.test.ts
+1
View File
@@ -679,6 +679,7 @@ describe('mapError', () => {
expect(result.status).toBe('RESOURCE_EXHAUSTED');
expect(result.message).toBe('Rate limit');
expect(result.fatal).toBe(true);
expect(result._meta?.['status']).toBe(429);
expect(result._meta?.['rawError']).toEqual({
message: 'Rate limit',
status: 429,
packages/core/src/agent/event-translator.ts
+1 -1
View File
@@ -403,7 +403,7 @@ export function mapError(
}
if (isStructuredError(error)) {
const structuredMeta = { ...meta, rawError: error };
const structuredMeta = { ...meta, rawError: error, status: error.status };
return {
status: mapHttpToGrpcStatus(error.status),
message: error.message,
packages/core/src/agent/legacy-agent-session.test.ts
+68 -76
View File
@@ -10,7 +10,7 @@ import { LegacyAgentSession } from './legacy-agent-session.js';
import type { LegacyAgentSessionDeps } from './legacy-agent-session.js';
import { GeminiEventType } from '../core/turn.js';
import type { ServerGeminiStreamEvent } from '../core/turn.js';
import type { AgentEvent } from './types.js';
import type { AgentEvent, AgentSend } from './types.js';
import { ToolErrorType } from '../tools/tool-error.js';
import type {
CompletedToolCall,
@@ -72,6 +72,18 @@ function makeToolRequest(callId: string, name: string): ToolCallRequestInfo {
};
}
function makeMessageSend(
text: string,
displayContent?: string,
): Extract<AgentSend, { message: unknown }> {
return {
message: {
content: [{ type: 'text', text }],
...(displayContent ? { displayContent } : {}),
},
};
}
function makeCompletedToolCall(
callId: string,
name: string,
@@ -140,9 +152,7 @@ describe('LegacyAgentSession', () => {
);
const session = new LegacyAgentSession(deps);
const result = await session.send({
message: [{ type: 'text', text: 'hi' }],
});
const result = await session.send(makeMessageSend('hi'));
expect(result.streamId).toBe('test-stream');
});
@@ -162,7 +172,10 @@ describe('LegacyAgentSession', () => {
const session = new LegacyAgentSession(deps);
const { streamId } = await session.send({
message: [{ type: 'text', text: 'hi' }],
message: {
content: [{ type: 'text', text: 'hi' }],
displayContent: 'raw input',
},
_meta: { source: 'user-test' },
});
@@ -170,8 +183,19 @@ describe('LegacyAgentSession', () => {
(e): e is AgentEvent<'message'> =>
e.type === 'message' && e.role === 'user' && e.streamId === streamId,
);
expect(userMessage?.content).toEqual([{ type: 'text', text: 'hi' }]);
expect(userMessage?.content).toEqual([
{ type: 'text', text: 'raw input' },
]);
expect(userMessage?._meta).toEqual({ source: 'user-test' });
await vi.advanceTimersByTimeAsync(0);
expect(sendMock).toHaveBeenCalledWith(
[{ text: 'hi' }],
expect.any(AbortSignal),
'test-prompt',
undefined,
false,
'raw input',
);
await collectEvents(session, { streamId: streamId ?? undefined });
});
@@ -195,9 +219,7 @@ describe('LegacyAgentSession', () => {
liveEvents.push(event);
});
const { streamId } = await session.send({
message: [{ type: 'text', text: 'hi' }],
});
const { streamId } = await session.send(makeMessageSend('hi'));
expect(streamId).toBe('test-stream');
expect(liveEvents.some((event) => event.type === 'agent_start')).toBe(
@@ -235,14 +257,12 @@ describe('LegacyAgentSession', () => {
);
const session = new LegacyAgentSession(deps);
const { streamId } = await session.send({
message: [{ type: 'text', text: 'first' }],
});
const { streamId } = await session.send(makeMessageSend('first'));
await vi.advanceTimersByTimeAsync(0);
await expect(
session.send({ message: [{ type: 'text', text: 'second' }] }),
).rejects.toThrow('cannot be called while a stream is active');
await expect(session.send(makeMessageSend('second'))).rejects.toThrow(
'cannot be called while a stream is active',
);
resolveHang?.();
await collectEvents(session, { streamId: streamId ?? undefined });
@@ -273,16 +293,12 @@ describe('LegacyAgentSession', () => {
);
const session = new LegacyAgentSession(deps);
const first = await session.send({
message: [{ type: 'text', text: 'first' }],
});
const first = await session.send(makeMessageSend('first'));
const firstEvents = await collectEvents(session, {
streamId: first.streamId ?? undefined,
});
const second = await session.send({
message: [{ type: 'text', text: 'second' }],
});
const second = await session.send(makeMessageSend('second'));
const secondEvents = await collectEvents(session, {
streamId: second.streamId ?? undefined,
});
@@ -330,7 +346,7 @@ describe('LegacyAgentSession', () => {
);
const session = new LegacyAgentSession(deps);
await session.send({ message: [{ type: 'text', text: 'hi' }] });
await session.send(makeMessageSend('hi'));
const events = await collectEvents(session);
const types = events.map((e) => e.type);
@@ -387,7 +403,7 @@ describe('LegacyAgentSession', () => {
]);
const session = new LegacyAgentSession(deps);
await session.send({ message: [{ type: 'text', text: 'read a file' }] });
await session.send(makeMessageSend('read a file'));
const events = await collectEvents(session);
const types = events.map((e) => e.type);
@@ -455,9 +471,7 @@ describe('LegacyAgentSession', () => {
scheduleMock.mockResolvedValueOnce([errorToolCall]);
const session = new LegacyAgentSession(deps);
await session.send({
message: [{ type: 'text', text: 'write file' }],
});
await session.send(makeMessageSend('write file'));
const events = await collectEvents(session);
const toolResp = events.find(
@@ -506,9 +520,7 @@ describe('LegacyAgentSession', () => {
scheduleMock.mockResolvedValueOnce([stopToolCall]);
const session = new LegacyAgentSession(deps);
await session.send({
message: [{ type: 'text', text: 'do something' }],
});
await session.send(makeMessageSend('do something'));
const events = await collectEvents(session);
const streamEnd = events.find(
@@ -552,9 +564,7 @@ describe('LegacyAgentSession', () => {
scheduleMock.mockResolvedValueOnce([fatalToolCall]);
const session = new LegacyAgentSession(deps);
await session.send({
message: [{ type: 'text', text: 'write file' }],
});
await session.send(makeMessageSend('write file'));
const events = await collectEvents(session);
const toolResp = events.find(
@@ -592,7 +602,7 @@ describe('LegacyAgentSession', () => {
);
const session = new LegacyAgentSession(deps);
await session.send({ message: [{ type: 'text', text: 'hi' }] });
await session.send(makeMessageSend('hi'));
const events = await collectEvents(session);
const streamEnd = events.find(
@@ -621,7 +631,7 @@ describe('LegacyAgentSession', () => {
);
const session = new LegacyAgentSession(deps);
await session.send({ message: [{ type: 'text', text: 'hi' }] });
await session.send(makeMessageSend('hi'));
const events = await collectEvents(session);
const blocked = events.find(
@@ -663,7 +673,7 @@ describe('LegacyAgentSession', () => {
);
const session = new LegacyAgentSession(deps);
await session.send({ message: [{ type: 'text', text: 'hi' }] });
await session.send(makeMessageSend('hi'));
const events = await collectEvents(session);
const err = events.find(
@@ -690,7 +700,7 @@ describe('LegacyAgentSession', () => {
);
const session = new LegacyAgentSession(deps);
await session.send({ message: [{ type: 'text', text: 'hi' }] });
await session.send(makeMessageSend('hi'));
const events = await collectEvents(session);
const warning = events.find(
@@ -738,7 +748,7 @@ describe('LegacyAgentSession', () => {
);
const session = new LegacyAgentSession(deps);
await session.send({ message: [{ type: 'text', text: 'hi' }] });
await session.send(makeMessageSend('hi'));
const events = await collectEvents(session);
const streamEnd = events.find(
@@ -762,7 +772,7 @@ describe('LegacyAgentSession', () => {
);
const session = new LegacyAgentSession(deps);
await session.send({ message: [{ type: 'text', text: 'hi' }] });
await session.send(makeMessageSend('hi'));
const events = await collectEvents(session);
const errorEvents = events.filter(
@@ -799,9 +809,7 @@ describe('LegacyAgentSession', () => {
);
const session = new LegacyAgentSession(deps);
const { streamId } = await session.send({
message: [{ type: 'text', text: 'hi' }],
});
const { streamId } = await session.send(makeMessageSend('hi'));
await vi.advanceTimersByTimeAsync(0);
await session.abort();
@@ -847,7 +855,7 @@ describe('LegacyAgentSession', () => {
);
const session = new LegacyAgentSession(deps);
await session.send({ message: [{ type: 'text', text: 'hi' }] });
await session.send(makeMessageSend('hi'));
// Give the loop time to start processing
await new Promise((r) => setTimeout(r, 50));
@@ -891,9 +899,7 @@ describe('LegacyAgentSession', () => {
);
const session = new LegacyAgentSession(deps);
const { streamId } = await session.send({
message: [{ type: 'text', text: 'hi' }],
});
const { streamId } = await session.send(makeMessageSend('hi'));
await new Promise((resolve) => setTimeout(resolve, 25));
await session.abort();
@@ -935,7 +941,7 @@ describe('LegacyAgentSession', () => {
);
const session = new LegacyAgentSession(deps);
await session.send({ message: [{ type: 'text', text: 'hi' }] });
await session.send(makeMessageSend('hi'));
await collectEvents(session);
expect(session.events.length).toBeGreaterThan(0);
@@ -964,9 +970,7 @@ describe('LegacyAgentSession', () => {
liveEvents.push(event);
});
const { streamId } = await session.send({
message: [{ type: 'text', text: 'hi' }],
});
const { streamId } = await session.send(makeMessageSend('hi'));
await collectEvents(session, { streamId: streamId ?? undefined });
unsubscribe();
@@ -1002,9 +1006,7 @@ describe('LegacyAgentSession', () => {
);
const session = new LegacyAgentSession(deps);
const first = await session.send({
message: [{ type: 'text', text: 'first request' }],
});
const first = await session.send(makeMessageSend('first request'));
await collectEvents(session, { streamId: first.streamId ?? undefined });
const liveEvents: AgentEvent[] = [];
@@ -1012,9 +1014,7 @@ describe('LegacyAgentSession', () => {
liveEvents.push(event);
});
const second = await session.send({
message: [{ type: 'text', text: 'second request' }],
});
const second = await session.send(makeMessageSend('second request'));
await collectEvents(session, { streamId: second.streamId ?? undefined });
unsubscribe();
@@ -1058,14 +1058,10 @@ describe('LegacyAgentSession', () => {
);
const session = new LegacyAgentSession(deps);
const first = await session.send({
message: [{ type: 'text', text: 'first request' }],
});
const first = await session.send(makeMessageSend('first request'));
await collectEvents(session, { streamId: first.streamId ?? undefined });
const second = await session.send({
message: [{ type: 'text', text: 'second request' }],
});
const second = await session.send(makeMessageSend('second request'));
await collectEvents(session, { streamId: second.streamId ?? undefined });
const firstStreamEvents = await collectEvents(session, {
@@ -1120,14 +1116,10 @@ describe('LegacyAgentSession', () => {
);
const session = new LegacyAgentSession(deps);
const first = await session.send({
message: [{ type: 'text', text: 'first request' }],
});
const first = await session.send(makeMessageSend('first request'));
await collectEvents(session, { streamId: first.streamId ?? undefined });
await session.send({
message: [{ type: 'text', text: 'second request' }],
});
await session.send(makeMessageSend('second request'));
await collectEvents(session);
const firstAgentMessage = session.events.find(
@@ -1175,7 +1167,7 @@ describe('LegacyAgentSession', () => {
);
const session = new LegacyAgentSession(deps);
await session.send({ message: [{ type: 'text', text: 'hi' }] });
await session.send(makeMessageSend('hi'));
const events = await collectEvents(session);
expect(events.length).toBeGreaterThan(0);
@@ -1196,7 +1188,7 @@ describe('LegacyAgentSession', () => {
);
const session = new LegacyAgentSession(deps);
await session.send({ message: [{ type: 'text', text: 'hi' }] });
await session.send(makeMessageSend('hi'));
const events = await collectEvents(session);
expect(events[events.length - 1]?.type).toBe('agent_end');
@@ -1244,7 +1236,7 @@ describe('LegacyAgentSession', () => {
]);
const session = new LegacyAgentSession(deps);
await session.send({ message: [{ type: 'text', text: 'do it' }] });
await session.send(makeMessageSend('do it'));
const events = await collectEvents(session);
// Only one agent_end at the very end
@@ -1291,7 +1283,7 @@ describe('LegacyAgentSession', () => {
]);
const session = new LegacyAgentSession(deps);
await session.send({ message: [{ type: 'text', text: 'go' }] });
await session.send(makeMessageSend('go'));
const events = await collectEvents(session);
// Should have at least one usage event from the intermediate Finished
@@ -1314,7 +1306,7 @@ describe('LegacyAgentSession', () => {
});
const session = new LegacyAgentSession(deps);
await session.send({ message: [{ type: 'text', text: 'hi' }] });
await session.send(makeMessageSend('hi'));
const events = await collectEvents(session);
const err = events.find(
@@ -1342,7 +1334,7 @@ describe('LegacyAgentSession', () => {
});
const session = new LegacyAgentSession(deps);
await session.send({ message: [{ type: 'text', text: 'hi' }] });
await session.send(makeMessageSend('hi'));
const events = await collectEvents(session);
const err = events.find(
@@ -1365,7 +1357,7 @@ describe('LegacyAgentSession', () => {
});
const session = new LegacyAgentSession(deps);
await session.send({ message: [{ type: 'text', text: 'hi' }] });
await session.send(makeMessageSend('hi'));
const events = await collectEvents(session);
const err = events.find(
@@ -1385,7 +1377,7 @@ describe('LegacyAgentSession', () => {
});
const session = new LegacyAgentSession(deps);
await session.send({ message: [{ type: 'text', text: 'hi' }] });
await session.send(makeMessageSend('hi'));
const events = await collectEvents(session);
const err = events.find(
@@ -1405,7 +1397,7 @@ describe('LegacyAgentSession', () => {
});
const session = new LegacyAgentSession(deps);
await session.send({ message: [{ type: 'text', text: 'hi' }] });
await session.send(makeMessageSend('hi'));
const events = await collectEvents(session);
const err = events.find(
packages/core/src/agent/legacy-agent-session.ts
+31 -9
View File
@@ -105,12 +105,16 @@ class LegacyAgentProtocol implements AgentProtocol {
this._beginNewStream();
const streamId = this._translationState.streamId;
const parts = contentPartsToGeminiParts(message);
const userMessage = this._makeUserMessageEvent(message, payload._meta);
const parts = contentPartsToGeminiParts(message.content);
const userMessage = this._makeUserMessageEvent(
message.content,
message.displayContent,
payload._meta,
);
this._emit([userMessage]);
this._scheduleRunLoop(parts);
this._scheduleRunLoop(parts, message.displayContent);
return { streamId };
}
@@ -119,18 +123,24 @@ class LegacyAgentProtocol implements AgentProtocol {
this._abortController.abort();
}
private _scheduleRunLoop(initialParts: Part[]): void {
private _scheduleRunLoop(
initialParts: Part[],
displayContent?: string,
): void {
// Use a macrotask so send() resolves with the streamId before agent_start
// is emitted and consumers can attach to the stream without racing startup.
setTimeout(() => {
void this._runLoopInBackground(initialParts);
void this._runLoopInBackground(initialParts, displayContent);
}, 0);
}
private async _runLoopInBackground(initialParts: Part[]): Promise<void> {
private async _runLoopInBackground(
initialParts: Part[],
displayContent?: string,
): Promise<void> {
this._ensureAgentStart();
try {
await this._runLoop(initialParts);
await this._runLoop(initialParts, displayContent);
} catch (err: unknown) {
if (this._abortController.signal.aborted || isAbortLikeError(err)) {
this._ensureAgentEnd('aborted');
@@ -141,8 +151,12 @@ class LegacyAgentProtocol implements AgentProtocol {
}
}
private async _runLoop(initialParts: Part[]): Promise<void> {
private async _runLoop(
initialParts: Part[],
initialDisplayContent?: string,
): Promise<void> {
let currentParts: Part[] = initialParts;
let currentDisplayContent = initialDisplayContent;
let turnCount = 0;
const maxTurns = this._config.getMaxSessionTurns();
@@ -162,7 +176,11 @@ class LegacyAgentProtocol implements AgentProtocol {
currentParts,
this._abortController.signal,
this._promptId,
undefined,
false,
currentDisplayContent,
);
currentDisplayContent = undefined;
for await (const event of responseStream) {
if (this._abortController.signal.aborted) {
@@ -383,13 +401,17 @@ class LegacyAgentProtocol implements AgentProtocol {
private _makeUserMessageEvent(
content: ContentPart[],
displayContent?: string,
meta?: Record<string, unknown>,
): AgentEvent<'message'> {
const eventContent: ContentPart[] = displayContent
? [{ type: 'text', text: displayContent }]
: content;
const event = {
...this._nextEventFields(),
type: 'message',
role: 'user',
content,
content: eventContent,
...(meta ? { _meta: meta } : {}),
} satisfies AgentEvent<'message'>;
return event;
packages/core/src/agent/mock.test.ts
+1 -1
View File
@@ -34,7 +34,7 @@ describe('MockAgentProtocol', () => {
const streamPromise = waitForStreamEnd(session);
const { streamId } = await session.send({
message: [{ type: 'text', text: 'hi' }],
message: { content: [{ type: 'text', text: 'hi' }] },
});
expect(streamId).toBeDefined();
packages/core/src/agent/mock.ts
+8 -1
View File
@@ -10,6 +10,7 @@ import type {
AgentEventData,
AgentProtocol,
AgentSend,
ContentPart,
Unsubscribe,
} from './types.js';
@@ -133,11 +134,17 @@ export class MockAgentProtocol implements AgentProtocol {
// 1. User/Update event (BEFORE agent_start)
if ('message' in payload && payload.message) {
const message = Array.isArray(payload.message)
? { content: payload.message, displayContent: undefined }
: payload.message;
const userContent: ContentPart[] = message.displayContent
? [{ type: 'text', text: message.displayContent }]
: message.content;
eventsToEmit.push(
normalize({
type: 'message',
role: 'user',
content: payload.message,
content: userContent,
_meta: payload._meta,
}),
);
packages/core/src/agent/types.ts
+4 -1
View File
@@ -46,7 +46,10 @@ type RequireExactlyOne<T> = {
}[keyof T];
interface AgentSendPayloads {
message: ContentPart[];
message: {
content: ContentPart[];
displayContent?: string;
};
elicitations: ElicitationResponse[];
update: { title?: string; model?: string; config?: Record<string, unknown> };
action: { type: string; data: unknown };
packages/core/src/output/json-formatter.ts
+2 -1
View File
@@ -6,6 +6,7 @@
import stripAnsi from 'strip-ansi';
import type { SessionMetrics } from '../telemetry/uiTelemetry.js';
import { getErrorType } from '../utils/errors.js';
import type { JsonError, JsonOutput } from './types.js';
export class JsonFormatter {
@@ -42,7 +43,7 @@ export class JsonFormatter {
sessionId?: string,
): string {
const jsonError: JsonError = {
type: error.constructor.name,
type: getErrorType(error),
message: stripAnsi(error.message),
...(code && { code }),
};