feat(hooks): Hook Session Lifecycle & Compression Integration (#14151)

packages/cli/src/gemini.test.tsx

@@ -257,6 +257,7 @@ describe('gemini.tsx main function', () => {
         getMessageBus: () => ({
           subscribe: vi.fn(),
         }),
+        getEnableHooks: () => false,
         getToolRegistry: vi.fn(),
         getContentGeneratorConfig: vi.fn(),
         getModel: () => 'gemini-pro',
@@ -489,6 +490,7 @@ describe('gemini.tsx main function kitty protocol', () => {
       getMessageBus: () => ({
         subscribe: vi.fn(),
       }),
+      getEnableHooks: () => false,
       getToolRegistry: vi.fn(),
       getContentGeneratorConfig: vi.fn(),
       getModel: () => 'gemini-pro',
@@ -588,6 +590,7 @@ describe('gemini.tsx main function kitty protocol', () => {
       getExtensions: () => [{ name: 'ext1' }],
       getPolicyEngine: vi.fn(),
       getMessageBus: () => ({ subscribe: vi.fn() }),
+      getEnableHooks: () => false,
       initialize: vi.fn(),
       getContentGeneratorConfig: vi.fn(),
       getMcpServers: () => ({}),
@@ -668,6 +671,7 @@ describe('gemini.tsx main function kitty protocol', () => {
       getExtensions: () => [],
       getPolicyEngine: vi.fn(),
       getMessageBus: () => ({ subscribe: vi.fn() }),
+      getEnableHooks: () => false,
       initialize: vi.fn(),
       getContentGeneratorConfig: vi.fn(),
       getMcpServers: () => ({}),
@@ -733,6 +737,7 @@ describe('gemini.tsx main function kitty protocol', () => {
       getDebugMode: () => false,
       getPolicyEngine: vi.fn(),
       getMessageBus: () => ({ subscribe: vi.fn() }),
+      getEnableHooks: () => false,
       initialize: vi.fn(),
       getContentGeneratorConfig: vi.fn(),
       getMcpServers: () => ({}),
@@ -814,6 +819,7 @@ describe('gemini.tsx main function kitty protocol', () => {
       getDebugMode: () => false,
       getPolicyEngine: vi.fn(),
       getMessageBus: () => ({ subscribe: vi.fn() }),
+      getEnableHooks: () => false,
       initialize: vi.fn(),
       getContentGeneratorConfig: vi.fn(),
       getMcpServers: () => ({}),
@@ -890,6 +896,7 @@ describe('gemini.tsx main function kitty protocol', () => {
       getDebugMode: () => false,
       getPolicyEngine: vi.fn(),
       getMessageBus: () => ({ subscribe: vi.fn() }),
+      getEnableHooks: () => false,
       initialize: vi.fn(),
       getContentGeneratorConfig: vi.fn(),
       getMcpServers: () => ({}),
@@ -961,6 +968,7 @@ describe('gemini.tsx main function kitty protocol', () => {
       getDebugMode: () => false,
       getPolicyEngine: vi.fn(),
       getMessageBus: () => ({ subscribe: vi.fn() }),
+      getEnableHooks: () => false,
       initialize: vi.fn(),
       getContentGeneratorConfig: vi.fn(),
       getMcpServers: () => ({}),
@@ -1130,6 +1138,7 @@ describe('gemini.tsx main function exit codes', () => {
       getGeminiMdFileCount: () => 0,
       getPolicyEngine: vi.fn(),
       getMessageBus: () => ({ subscribe: vi.fn() }),
+      getEnableHooks: () => false,
       getToolRegistry: vi.fn(),
       getContentGeneratorConfig: vi.fn(),
       getModel: () => 'gemini-pro',
@@ -1191,6 +1200,7 @@ describe('gemini.tsx main function exit codes', () => {
       getGeminiMdFileCount: () => 0,
       getPolicyEngine: vi.fn(),
       getMessageBus: () => ({ subscribe: vi.fn() }),
+      getEnableHooks: () => false,
       getToolRegistry: vi.fn(),
       getContentGeneratorConfig: vi.fn(),
       getModel: () => 'gemini-pro',
@@ -1302,6 +1312,7 @@ describe('startInteractiveUI', () => {
     registerCleanup: vi.fn(),
     runExitCleanup: vi.fn(),
     registerSyncCleanup: vi.fn(),
+    registerTelemetryConfig: vi.fn(),
   }));
 
   afterEach(() => {

packages/cli/src/gemini.tsx

@@ -31,6 +31,7 @@ import {
   registerCleanup,
   registerSyncCleanup,
   runExitCleanup,
+  registerTelemetryConfig,
 } from './utils/cleanup.js';
 import { getCliVersion } from './utils/version.js';
 import {
@@ -58,6 +59,10 @@ import {
   shouldEnterAlternateScreen,
   startupProfiler,
   ExitCodes,
+  SessionStartSource,
+  SessionEndReason,
+  fireSessionStartHook,
+  fireSessionEndHook,
 } from '@google/gemini-cli-core';
 import {
   initializeApp,
@@ -459,10 +464,22 @@ export async function main() {
     const config = await loadCliConfig(settings.merged, sessionId, argv);
     loadConfigHandle?.end();
 
+    // Register config for telemetry shutdown
+    // This ensures telemetry (including SessionEnd hooks) is properly flushed on exit
+    registerTelemetryConfig(config);
+
     const policyEngine = config.getPolicyEngine();
     const messageBus = config.getMessageBus();
     createPolicyUpdater(policyEngine, messageBus);
 
+    // Register SessionEnd hook to fire on graceful exit
+    // This runs before telemetry shutdown in runExitCleanup()
+    if (config.getEnableHooks() && messageBus) {
+      registerCleanup(async () => {
+        await fireSessionEndHook(messageBus, SessionEndReason.Exit);
+      });
+    }
+
     // Cleanup sessions after config initialization
     try {
       await cleanupExpiredSessions(config, settings.merged);
@@ -586,6 +603,22 @@ export async function main() {
     await config.initialize();
     startupProfiler.flush(config);
 
+    // Fire SessionStart hook through MessageBus (only if hooks are enabled)
+    // Must be called AFTER config.initialize() to ensure HookRegistry is loaded
+    const hooksEnabled = config.getEnableHooks();
+    const hookMessageBus = config.getMessageBus();
+    if (hooksEnabled && hookMessageBus) {
+      const sessionStartSource = resumedSessionData
+        ? SessionStartSource.Resume
+        : SessionStartSource.Startup;
+      await fireSessionStartHook(hookMessageBus, sessionStartSource);
+
+      // Register SessionEnd hook for graceful exit
+      registerCleanup(async () => {
+        await fireSessionEndHook(hookMessageBus, SessionEndReason.Exit);
+      });
+    }
+
     // If not a TTY, read from stdin
     // This is for cases where the user pipes input directly into the command
     if (!process.stdin.isTTY) {

packages/cli/src/gemini_cleanup.test.tsx

@@ -186,6 +186,7 @@ describe('gemini.tsx main function cleanup', () => {
       getDebugMode: vi.fn(() => false),
       getPolicyEngine: vi.fn(),
       getMessageBus: () => ({ subscribe: vi.fn() }),
+      getEnableHooks: vi.fn(() => false),
       initialize: vi.fn(),
       getContentGeneratorConfig: vi.fn(),
       getMcpServers: () => ({}),

packages/cli/src/nonInteractiveCli.test.ts

@@ -16,7 +16,6 @@ import type {
 import {
   executeToolCall,
   ToolErrorType,
-  shutdownTelemetry,
   GeminiEventType,
   OutputFormat,
   uiTelemetryService,
@@ -61,7 +60,6 @@ vi.mock('@google/gemini-cli-core', async (importOriginal) => {
   return {
     ...original,
     executeToolCall: vi.fn(),
-    shutdownTelemetry: vi.fn(),
     isTelemetrySdkInitialized: vi.fn().mockReturnValue(true),
     ChatRecordingService: MockChatRecordingService,
     uiTelemetryService: {
@@ -91,7 +89,6 @@ describe('runNonInteractive', () => {
   let mockSettings: LoadedSettings;
   let mockToolRegistry: ToolRegistry;
   let mockCoreExecuteToolCall: Mock;
-  let mockShutdownTelemetry: Mock;
   let consoleErrorSpy: MockInstance;
   let processStdoutSpy: MockInstance;
   let processStderrSpy: MockInstance;
@@ -123,7 +120,6 @@ describe('runNonInteractive', () => {
 
   beforeEach(async () => {
     mockCoreExecuteToolCall = vi.mocked(executeToolCall);
-    mockShutdownTelemetry = vi.mocked(shutdownTelemetry);
 
     mockCommandServiceCreate.mockResolvedValue({
       getCommands: mockGetCommands,
@@ -247,7 +243,8 @@ describe('runNonInteractive', () => {
       'prompt-id-1',
     );
     expect(getWrittenOutput()).toBe('Hello World\n');
-    expect(mockShutdownTelemetry).toHaveBeenCalled();
+    // Note: Telemetry shutdown is now handled in runExitCleanup() in cleanup.ts
+    // so we no longer expect shutdownTelemetry to be called directly here
   });
 
   it('should handle a single tool call and respond', async () => {

packages/cli/src/nonInteractiveCli.ts

@@ -15,8 +15,6 @@ import { isSlashCommand } from './ui/utils/commandUtils.js';
 import type { LoadedSettings } from './config/settings.js';
 import {
   executeToolCall,
-  shutdownTelemetry,
-  isTelemetrySdkInitialized,
   GeminiEventType,
   FatalInputError,
   promptIdContext,
@@ -445,9 +443,6 @@ export async function runNonInteractive({
 
       consolePatcher.cleanup();
       coreEvents.off(CoreEvent.UserFeedback, handleUserFeedback);
-      if (isTelemetrySdkInitialized()) {
-        await shutdownTelemetry(config);
-      }
     }
 
     if (errorToHandle) {

packages/cli/src/ui/AppContainer.tsx

@@ -59,6 +59,10 @@ import {
   disableLineWrapping,
   shouldEnterAlternateScreen,
   startupProfiler,
+  SessionStartSource,
+  SessionEndReason,
+  fireSessionStartHook,
+  fireSessionEndHook,
 } from '@google/gemini-cli-core';
 import { validateAuthMethod } from '../config/auth.js';
 import process from 'node:process';
@@ -284,14 +288,32 @@ export const AppContainer = (props: AppContainerProps) => {
       await config.initialize();
       setConfigInitialized(true);
       startupProfiler.flush(config);
+
+      // Fire SessionStart hook through MessageBus (only if hooks are enabled)
+      // Must be called AFTER config.initialize() to ensure HookRegistry is loaded
+      const hooksEnabled = config.getEnableHooks();
+      const hookMessageBus = config.getMessageBus();
+      if (hooksEnabled && hookMessageBus) {
+        const sessionStartSource = resumedSessionData
+          ? SessionStartSource.Resume
+          : SessionStartSource.Startup;
+        await fireSessionStartHook(hookMessageBus, sessionStartSource);
+      }
     })();
     registerCleanup(async () => {
       // Turn off mouse scroll.
       disableMouseEvents();
       const ideClient = await IdeClient.getInstance();
       await ideClient.disconnect();
+
+      // Fire SessionEnd hook on cleanup (only if hooks are enabled)
+      const hooksEnabled = config.getEnableHooks();
+      const hookMessageBus = config.getMessageBus();
+      if (hooksEnabled && hookMessageBus) {
+        await fireSessionEndHook(hookMessageBus, SessionEndReason.Exit);
+      }
     });
-  }, [config]);
+  }, [config, resumedSessionData]);
 
   useEffect(
     () => setUpdateHandler(historyManager.addItem, setUpdateInfo),

packages/cli/src/ui/commands/clearCommand.test.ts

@@ -44,6 +44,8 @@ describe('clearCommand', () => {
               }),
             }) as unknown as GeminiClient,
           setSessionId: vi.fn(),
+          getEnableHooks: vi.fn().mockReturnValue(false),
+          getMessageBus: vi.fn().mockReturnValue(undefined),
         },
       },
     });

packages/cli/src/ui/commands/clearCommand.ts

@@ -4,7 +4,14 @@
  * SPDX-License-Identifier: Apache-2.0
  */
 
-import { uiTelemetryService } from '@google/gemini-cli-core';
+import {
+  uiTelemetryService,
+  fireSessionEndHook,
+  fireSessionStartHook,
+  SessionEndReason,
+  SessionStartSource,
+  flushTelemetry,
+} from '@google/gemini-cli-core';
 import type { SlashCommand } from './types.js';
 import { CommandKind } from './types.js';
 import { randomUUID } from 'node:crypto';
@@ -21,6 +28,12 @@ export const clearCommand: SlashCommand = {
       ?.getGeminiClient()
       ?.getChat()
       .getChatRecordingService();
+    const messageBus = config?.getMessageBus();
+
+    // Fire SessionEnd hook before clearing
+    if (config?.getEnableHooks() && messageBus) {
+      await fireSessionEndHook(messageBus, SessionEndReason.Clear);
+    }
 
     if (geminiClient) {
       context.ui.setDebugMessage('Clearing terminal and resetting chat.');
@@ -38,6 +51,21 @@ export const clearCommand: SlashCommand = {
       chatRecordingService.initialize();
     }
 
+    // Fire SessionStart hook after clearing
+    if (config?.getEnableHooks() && messageBus) {
+      await fireSessionStartHook(messageBus, SessionStartSource.Clear);
+    }
+
+    // Give the event loop a chance to process any pending telemetry operations
+    // This ensures logger.emit() calls have fully propagated to the BatchLogRecordProcessor
+    await new Promise((resolve) => setImmediate(resolve));
+
+    // Flush telemetry to ensure hooks are written to disk immediately
+    // This is critical for tests and environments with I/O latency
+    if (config) {
+      await flushTelemetry(config);
+    }
+
     uiTelemetryService.setLastPromptTokenCount(0);
     context.ui.clear();
   },

packages/cli/src/utils/cleanup.ts

@@ -6,10 +6,16 @@
 
 import { promises as fs } from 'node:fs';
 import { join } from 'node:path';
-import { Storage } from '@google/gemini-cli-core';
+import {
+  Storage,
+  shutdownTelemetry,
+  isTelemetrySdkInitialized,
+} from '@google/gemini-cli-core';
+import type { Config } from '@google/gemini-cli-core';
 
 const cleanupFunctions: Array<(() => void) | (() => Promise<void>)> = [];
 const syncCleanupFunctions: Array<() => void> = [];
+let configForTelemetry: Config | null = null;
 
 export function registerCleanup(fn: (() => void) | (() => Promise<void>)) {
   cleanupFunctions.push(fn);
@@ -30,6 +36,14 @@ export function runSyncCleanup() {
   syncCleanupFunctions.length = 0;
 }
 
+/**
+ * Register the config instance for telemetry shutdown.
+ * This must be called early in the application lifecycle.
+ */
+export function registerTelemetryConfig(config: Config) {
+  configForTelemetry = config;
+}
+
 export async function runExitCleanup() {
   runSyncCleanup();
   for (const fn of cleanupFunctions) {
@@ -40,6 +54,16 @@ export async function runExitCleanup() {
     }
   }
   cleanupFunctions.length = 0; // Clear the array
+
+  // IMPORTANT: Shutdown telemetry AFTER all other cleanup functions have run
+  // This ensures SessionEnd hooks and other telemetry are properly flushed
+  if (configForTelemetry && isTelemetrySdkInitialized()) {
+    try {
+      await shutdownTelemetry(configForTelemetry);
+    } catch (_) {
+      // Ignore errors during telemetry shutdown
+    }
+  }
 }
 
 export async function cleanupCheckpoints() {

packages/core/src/core/sessionHookTriggers.ts

@@ -0,0 +1,96 @@
+/**
+ * @license
+ * Copyright 2025 Google LLC
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+import type { MessageBus } from '../confirmation-bus/message-bus.js';
+import {
+  MessageBusType,
+  type HookExecutionRequest,
+  type HookExecutionResponse,
+} from '../confirmation-bus/types.js';
+import type {
+  SessionStartSource,
+  SessionEndReason,
+  PreCompressTrigger,
+} from '../hooks/types.js';
+import { debugLogger } from '../utils/debugLogger.js';
+
+/**
+ * Fires the SessionStart hook.
+ *
+ * @param messageBus The message bus to use for hook communication
+ * @param source The source/trigger of the session start
+ */
+export async function fireSessionStartHook(
+  messageBus: MessageBus,
+  source: SessionStartSource,
+): Promise<void> {
+  try {
+    await messageBus.request<HookExecutionRequest, HookExecutionResponse>(
+      {
+        type: MessageBusType.HOOK_EXECUTION_REQUEST,
+        eventName: 'SessionStart',
+        input: {
+          source,
+        },
+      },
+      MessageBusType.HOOK_EXECUTION_RESPONSE,
+    );
+  } catch (error) {
+    debugLogger.warn(`SessionStart hook failed:`, error);
+  }
+}
+
+/**
+ * Fires the SessionEnd hook.
+ *
+ * @param messageBus The message bus to use for hook communication
+ * @param reason The reason for the session end
+ */
+export async function fireSessionEndHook(
+  messageBus: MessageBus,
+  reason: SessionEndReason,
+): Promise<void> {
+  try {
+    await messageBus.request<HookExecutionRequest, HookExecutionResponse>(
+      {
+        type: MessageBusType.HOOK_EXECUTION_REQUEST,
+        eventName: 'SessionEnd',
+        input: {
+          reason,
+        },
+      },
+      MessageBusType.HOOK_EXECUTION_RESPONSE,
+    );
+  } catch (error) {
+    debugLogger.warn(`SessionEnd hook failed:`, error);
+  }
+}
+
+/**
+ * Fires the PreCompress hook.
+ *
+ * @param messageBus The message bus to use for hook communication
+ * @param trigger The trigger type (manual or auto)
+ */
+export async function firePreCompressHook(
+  messageBus: MessageBus,
+  trigger: PreCompressTrigger,
+): Promise<void> {
+  try {
+    await messageBus.request<HookExecutionRequest, HookExecutionResponse>(
+      {
+        type: MessageBusType.HOOK_EXECUTION_REQUEST,
+        eventName: 'PreCompress',
+        input: {
+          trigger,
+        },
+      },
+      MessageBusType.HOOK_EXECUTION_RESPONSE,
+    );
+  } catch (error) {
+    debugLogger.warn(`PreCompress hook failed:`, error);
+  }
+}

packages/core/src/hooks/hookEventHandler.ts

@@ -220,6 +220,57 @@ function validateNotificationInput(input: Record<string, unknown>): {
   };
 }
 
+/**
+ * Validates SessionStart input fields
+ */
+function validateSessionStartInput(input: Record<string, unknown>): {
+  source: SessionStartSource;
+} {
+  const source = input['source'];
+  if (typeof source !== 'string') {
+    throw new Error(
+      'Invalid input for SessionStart hook event: source must be a string',
+    );
+  }
+  return {
+    source: source as SessionStartSource,
+  };
+}
+
+/**
+ * Validates SessionEnd input fields
+ */
+function validateSessionEndInput(input: Record<string, unknown>): {
+  reason: SessionEndReason;
+} {
+  const reason = input['reason'];
+  if (typeof reason !== 'string') {
+    throw new Error(
+      'Invalid input for SessionEnd hook event: reason must be a string',
+    );
+  }
+  return {
+    reason: reason as SessionEndReason,
+  };
+}
+
+/**
+ * Validates PreCompress input fields
+ */
+function validatePreCompressInput(input: Record<string, unknown>): {
+  trigger: PreCompressTrigger;
+} {
+  const trigger = input['trigger'];
+  if (typeof trigger !== 'string') {
+    throw new Error(
+      'Invalid input for PreCompress hook event: trigger must be a string',
+    );
+  }
+  return {
+    trigger: trigger as PreCompressTrigger,
+  };
+}
+
 /**
  * Hook event bus that coordinates hook execution across the system
  */
@@ -704,6 +755,21 @@ export class HookEventHandler {
           );
           break;
         }
+        case HookEventName.SessionStart: {
+          const { source } = validateSessionStartInput(enrichedInput);
+          result = await this.fireSessionStartEvent(source);
+          break;
+        }
+        case HookEventName.SessionEnd: {
+          const { reason } = validateSessionEndInput(enrichedInput);
+          result = await this.fireSessionEndEvent(reason);
+          break;
+        }
+        case HookEventName.PreCompress: {
+          const { trigger } = validatePreCompressInput(enrichedInput);
+          result = await this.firePreCompressEvent(trigger);
+          break;
+        }
         default:
           throw new Error(`Unsupported hook event: ${request.eventName}`);
       }

packages/core/src/hooks/hookRunner.ts

@@ -238,8 +238,18 @@ export class HookRunner {
             debugLogger.warn(`Hook stdin error: ${err}`);
           }
         });
-        child.stdin.write(JSON.stringify(input));
-        child.stdin.end();
+
+        // Wrap write operations in try-catch to handle synchronous EPIPE errors
+        // that occur when the child process exits before we finish writing
+        try {
+          child.stdin.write(JSON.stringify(input));
+          child.stdin.end();
+        } catch (err) {
+          // Ignore EPIPE errors which happen when the child process closes stdin early
+          if (err instanceof Error && 'code' in err && err.code !== 'EPIPE') {
+            debugLogger.warn(`Hook stdin write error: ${err}`);
+          }
+        }
       }
 
       // Collect stdout

packages/core/src/hooks/index.ts

@@ -19,3 +19,10 @@ export { HookEventHandler } from './hookEventHandler.js';
 export type { HookRegistryEntry, ConfigSource } from './hookRegistry.js';
 export type { AggregatedHookResult } from './hookAggregator.js';
 export type { HookEventContext } from './hookPlanner.js';
+
+// Export hook trigger functions
+export {
+  fireSessionStartHook,
+  fireSessionEndHook,
+  firePreCompressHook,
+} from '../core/sessionHookTriggers.js';

packages/core/src/hooks/types.ts

@@ -463,7 +463,6 @@ export enum SessionStartSource {
   Startup = 'startup',
   Resume = 'resume',
   Clear = 'clear',
-  Compress = 'compress',
 }
 
 /**

packages/core/src/services/chatCompressionService.test.ts

@@ -157,6 +157,8 @@ describe('ChatCompressionService', () => {
       getContentGenerator: vi.fn().mockReturnValue({
         countTokens: vi.fn().mockResolvedValue({ totalTokens: 100 }),
       }),
+      getEnableHooks: vi.fn().mockReturnValue(false),
+      getMessageBus: vi.fn().mockReturnValue(undefined),
     } as unknown as Config;
 
     vi.mocked(tokenLimit).mockReturnValue(1000);

packages/core/src/services/chatCompressionService.ts

@@ -21,6 +21,8 @@ import {
   DEFAULT_GEMINI_MODEL,
   PREVIEW_GEMINI_MODEL,
 } from '../config/models.js';
+import { firePreCompressHook } from '../core/sessionHookTriggers.js';
+import { PreCompressTrigger } from '../hooks/types.js';
 
 /**
  * Default threshold for compression token count as a fraction of the model's
@@ -123,6 +125,17 @@ export class ChatCompressionService {
       };
     }
 
+    // Fire PreCompress hook before compression (only if hooks are enabled)
+    // This fires for both manual and auto compression attempts
+    const hooksEnabled = config.getEnableHooks();
+    const messageBus = config.getMessageBus();
+    if (hooksEnabled && messageBus) {
+      const trigger = force
+        ? PreCompressTrigger.Manual
+        : PreCompressTrigger.Auto;
+      await firePreCompressHook(messageBus, trigger);
+    }
+
     const originalTokenCount = chat.getLastPromptTokenCount();
 
     // Don't compress if not forced and we are under the limit.

packages/core/src/telemetry/index.ts

@@ -16,6 +16,7 @@ export { DEFAULT_TELEMETRY_TARGET, DEFAULT_OTLP_ENDPOINT };
 export {
   initializeTelemetry,
   shutdownTelemetry,
+  flushTelemetry,
   isTelemetrySdkInitialized,
 } from './sdk.js';
 export {

packages/core/src/telemetry/sdk.ts

@@ -80,6 +80,8 @@ class DiagLoggerAdapter {
 diag.setLogger(new DiagLoggerAdapter(), DiagLogLevel.INFO);
 
 let sdk: NodeSDK | undefined;
+let spanProcessor: BatchSpanProcessor | undefined;
+let logRecordProcessor: BatchLogRecordProcessor | undefined;
 let telemetryInitialized = false;
 let callbackRegistered = false;
 let authListener: ((newCredentials: JWTInput) => Promise<void>) | undefined =
@@ -273,10 +275,14 @@ export async function initializeTelemetry(
     });
   }
 
+  // Store processor references for manual flushing
+  spanProcessor = new BatchSpanProcessor(spanExporter);
+  logRecordProcessor = new BatchLogRecordProcessor(logExporter);
+
   sdk = new NodeSDK({
     resource,
-    spanProcessors: [new BatchSpanProcessor(spanExporter)],
-    logRecordProcessors: [new BatchLogRecordProcessor(logExporter)],
+    spanProcessors: [spanProcessor],
+    logRecordProcessors: [logRecordProcessor],
     metricReader,
     instrumentations: [new HttpInstrumentation()],
   });
@@ -293,15 +299,37 @@ export async function initializeTelemetry(
     console.error('Error starting OpenTelemetry SDK:', error);
   }
 
+  // Note: We don't use process.on('exit') here because that callback is synchronous
+  // and won't wait for the async shutdownTelemetry() to complete.
+  // Instead, telemetry shutdown is handled in runExitCleanup() in cleanup.ts
   process.on('SIGTERM', () => {
     shutdownTelemetry(config);
   });
   process.on('SIGINT', () => {
     shutdownTelemetry(config);
   });
-  process.on('exit', () => {
-    shutdownTelemetry(config);
-  });
+}
+
+/**
+ * Force flush all pending telemetry data to disk.
+ * This is useful for ensuring telemetry is written before critical operations like /clear.
+ */
+export async function flushTelemetry(config: Config): Promise<void> {
+  if (!telemetryInitialized || !spanProcessor || !logRecordProcessor) {
+    return;
+  }
+  try {
+    // Force flush all pending telemetry to disk
+    await Promise.all([
+      spanProcessor.forceFlush(),
+      logRecordProcessor.forceFlush(),
+    ]);
+    if (config.getDebugMode()) {
+      debugLogger.log('OpenTelemetry SDK flushed successfully.');
+    }
+  } catch (error) {
+    console.error('Error flushing SDK:', error);
+  }
 }
 
 export async function shutdownTelemetry(