Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion packages/cli/src/config/config.ts
Original file line number Diff line number Diff line change
Expand Up @@ -242,7 +242,7 @@ export async function parseArguments(settings: Settings): Promise<CliArgs> {
.option('output-format', {
type: 'string',
description: 'The format of the CLI output.',
choices: ['text', 'json'],
choices: ['text', 'json', 'stream-json'],
})
.option('resume', {
alias: 'r',
Expand Down
27 changes: 27 additions & 0 deletions packages/cli/src/nonInteractiveCli.ts
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,9 @@ import {
promptIdContext,
OutputFormat,
JsonFormatter,
StreamJsonFormatter,
uiTelemetryService,
streamingTelemetryService,
} from '@blocksuser/gemini-cli-core';
import type { Content, Part } from '@google/genai';

Expand All @@ -41,6 +43,18 @@ export async function runNonInteractive(

try {
consolePatcher.patch();

// Set up streaming telemetry for stream-json format
const isStreamJsonFormat = config.getOutputFormat() === OutputFormat.STREAM_JSON;
let streamJsonFormatter: StreamJsonFormatter | undefined;

if (isStreamJsonFormat) {
streamJsonFormatter = new StreamJsonFormatter();
streamingTelemetryService.enable();
streamingTelemetryService.addTelemetryListener((event) => {
process.stdout.write(streamJsonFormatter!.formatTelemetryBlock(event) + '\n');
});
}
// 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') {
Expand Down Expand Up @@ -123,6 +137,11 @@ export async function runNonInteractive(
if (event.type === GeminiEventType.Content) {
if (config.getOutputFormat() === OutputFormat.JSON) {
responseText += event.value;
} else if (config.getOutputFormat() === OutputFormat.STREAM_JSON) {
responseText += event.value;
if (streamJsonFormatter) {
process.stdout.write(streamJsonFormatter.formatContentBlock(event.value) + '\n');
}
} else {
process.stdout.write(event.value);
}
Expand Down Expand Up @@ -162,6 +181,11 @@ export async function runNonInteractive(
const formatter = new JsonFormatter();
const stats = uiTelemetryService.getMetrics();
process.stdout.write(formatter.format(responseText, stats));
} else if (config.getOutputFormat() === OutputFormat.STREAM_JSON) {
if (streamJsonFormatter) {
const stats = uiTelemetryService.getMetrics();
process.stdout.write(streamJsonFormatter.formatFinalBlock(responseText, stats) + '\n');
}
} else {
process.stdout.write('\n'); // Ensure a final newline
}
Expand All @@ -172,6 +196,9 @@ export async function runNonInteractive(
handleError(error, config);
} finally {
consolePatcher.cleanup();
if (config.getOutputFormat() === OutputFormat.STREAM_JSON) {
streamingTelemetryService.disable();
}
if (isTelemetrySdkInitialized()) {
await shutdownTelemetry(config);
}
Expand Down
1 change: 1 addition & 0 deletions packages/core/src/index.ts
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@
export * from './config/config.js';
export * from './output/types.js';
export * from './output/json-formatter.js';
export * from './output/stream-json-formatter.js';

// Export Core Logic
export * from './core/client.js';
Expand Down
77 changes: 77 additions & 0 deletions packages/core/src/output/stream-json-formatter.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,77 @@
/**
* @license
* Copyright 2025 Google LLC
* SPDX-License-Identifier: Apache-2.0
*/

import stripAnsi from 'strip-ansi';
import type { SessionMetrics } from '../telemetry/uiTelemetry.js';
import type { JsonError } from './types.js';
import type { TelemetryEvent } from '../telemetry/types.js';

export interface StreamJsonTelemetryBlock {
type: 'telemetry';
event: TelemetryEvent;
}

export interface StreamJsonContentBlock {
type: 'content';
content: string;
}

export interface StreamJsonFinalBlock {
type: 'final';
response?: string;
stats?: SessionMetrics;
error?: JsonError;
}

export type StreamJsonBlock = StreamJsonTelemetryBlock | StreamJsonContentBlock | StreamJsonFinalBlock;

export class StreamJsonFormatter {
formatTelemetryBlock(event: TelemetryEvent): string {
const block: StreamJsonTelemetryBlock = {
type: 'telemetry',
event,
};
return JSON.stringify(block);
}

formatContentBlock(content: string): string {
const block: StreamJsonContentBlock = {
type: 'content',
content: stripAnsi(content),
};
return JSON.stringify(block);
}

formatFinalBlock(response?: string, stats?: SessionMetrics, error?: JsonError): string {
const block: StreamJsonFinalBlock = {
type: 'final',
};

if (response !== undefined) {
block.response = stripAnsi(response);
}

if (stats) {
block.stats = stats;
}

if (error) {
block.error = error;
}

return JSON.stringify(block);
}

formatError(error: Error, code?: string | number): string {
const jsonError: JsonError = {
type: error.constructor.name,
message: stripAnsi(error.message),
...(code && { code }),
};

return this.formatFinalBlock(undefined, undefined, jsonError);
}
}
1 change: 1 addition & 0 deletions packages/core/src/output/types.ts
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@ import type { SessionMetrics } from '../telemetry/uiTelemetry.js';
export enum OutputFormat {
TEXT = 'text',
JSON = 'json',
STREAM_JSON = 'stream-json',
}

export interface JsonError {
Expand Down
2 changes: 2 additions & 0 deletions packages/core/src/telemetry/index.ts
Original file line number Diff line number Diff line change
Expand Up @@ -54,3 +54,5 @@ export { SemanticAttributes } from '@opentelemetry/semantic-conventions';
export * from './uiTelemetry.js';
export { HighWaterMarkTracker } from './high-water-mark-tracker.js';
export { RateLimiter } from './rate-limiter.js';
export { streamingTelemetryService } from './streamingTelemetry.js';
export type { TelemetryStreamListener } from './streamingTelemetry.js';
4 changes: 4 additions & 0 deletions packages/core/src/telemetry/loggers.ts
Original file line number Diff line number Diff line change
Expand Up @@ -66,6 +66,7 @@ import {
import { isTelemetrySdkInitialized } from './sdk.js';
import type { UiEvent } from './uiTelemetry.js';
import { uiTelemetryService } from './uiTelemetry.js';
import { streamingTelemetryService } from './streamingTelemetry.js';
import { ClearcutLogger } from './clearcut-logger/clearcut-logger.js';
import { safeJsonStringify } from '../utils/safeJsonStringify.js';
import { UserAccountManager } from '../utils/userAccountManager.js';
Expand Down Expand Up @@ -119,6 +120,7 @@ export function logCliConfiguration(
}

export function logUserPrompt(config: Config, event: UserPromptEvent): void {
streamingTelemetryService.emitEvent(event);
ClearcutLogger.getInstance(config)?.logNewPromptEvent(event);
if (!isTelemetrySdkInitialized()) return;

Expand Down Expand Up @@ -147,6 +149,7 @@ export function logUserPrompt(config: Config, event: UserPromptEvent): void {
}

export function logToolCall(config: Config, event: ToolCallEvent): void {
streamingTelemetryService.emitEvent(event);
const uiEvent = {
...event,
'event.name': EVENT_TOOL_CALL,
Expand Down Expand Up @@ -359,6 +362,7 @@ export function logApiError(config: Config, event: ApiErrorEvent): void {
}

export function logApiResponse(config: Config, event: ApiResponseEvent): void {
streamingTelemetryService.emitEvent(event);
const uiEvent = {
...event,
'event.name': EVENT_API_RESPONSE,
Expand Down
44 changes: 44 additions & 0 deletions packages/core/src/telemetry/streamingTelemetry.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,44 @@
/**
* @license
* Copyright 2025 Google LLC
* SPDX-License-Identifier: Apache-2.0
*/

import { EventEmitter } from 'node:events';
import type { TelemetryEvent } from './types.js';

export interface TelemetryStreamListener {
(event: TelemetryEvent): void;
}

class StreamingTelemetryService extends EventEmitter {
private enabled = false;

enable(): void {
this.enabled = true;
}

disable(): void {
this.enabled = false;
}

isEnabled(): boolean {
return this.enabled;
}

addTelemetryListener(listener: TelemetryStreamListener): void {
this.on('telemetry', listener);
}

removeTelemetryListener(listener: TelemetryStreamListener): void {
this.off('telemetry', listener);
}

emitEvent(event: TelemetryEvent): void {
if (this.enabled) {
this.emit('telemetry', event);
}
}
}

export const streamingTelemetryService = new StreamingTelemetryService();