feat(telemetry): flag synchronous code blocking the event loop

.changeset/slow-loops-report.md

@@ -0,0 +1,5 @@
+---
+'@livekit/agents': minor
+---
+
+Flag synchronous code that blocks agent event loops in telemetry and logs.

agents/package.json

@@ -62,11 +62,13 @@
     "@opentelemetry/api-logs": "^0.220.0",
     "@opentelemetry/core": "^2.8.0",
     "@opentelemetry/exporter-logs-otlp-proto": "^0.220.0",
+    "@opentelemetry/exporter-metrics-otlp-proto": "^0.220.0",
     "@opentelemetry/exporter-trace-otlp-proto": "^0.220.0",
     "@opentelemetry/instrumentation-pino": "^0.66.0",
     "@opentelemetry/otlp-exporter-base": "^0.220.0",
     "@opentelemetry/resources": "^2.8.0",
     "@opentelemetry/sdk-logs": "^0.220.0",
+    "@opentelemetry/sdk-metrics": "^2.8.0",
     "@opentelemetry/sdk-trace-base": "^2.8.0",
     "@opentelemetry/sdk-trace-node": "^2.8.0",
     "@opentelemetry/semantic-conventions": "^1.28.0",

agents/src/inference/stt.test.ts

@@ -1075,7 +1075,9 @@ describeLiveKitInference('LiveKit Inference STT integration', agents, async (har
     'assemblyai/universal-streaming',
     'xai/stt-1',
   ] as const) {
-    describe(model, { retry: 1 }, async () => {
+    // each model is an independent gateway session: run the models, and both sample rates of
+    // each, at the same time instead of one 50 s clip after another
+    describe(model, { retry: 1, concurrent: true }, async () => {
       const stt =
         model === 'assemblyai/universal-streaming'
           ? new STT({ model, modelOptions: { format_turns: true } })

agents/src/inference/test_utils.ts

@@ -13,7 +13,7 @@ interface InferenceTestHarness {
   stt: (
     model: STT,
     vad: VAD,
-    supports?: Partial<{ streaming: boolean; nonStreaming: boolean }>,
+    supports?: Partial<{ streaming: boolean; nonStreaming: boolean; streamSpeed: number }>,
   ) => Promise<void>;
   tts: (
     model: TTS,

agents/src/ipc/job_proc_lazy_main.ts

@@ -3,21 +3,30 @@
 // SPDX-License-Identifier: Apache-2.0
 import { Room, RoomEvent, dispose } from '@livekit/rtc-node';
 import { ThrowsPromise } from '@livekit/throws-transformer/throws';
+import { context as otelContext } from '@opentelemetry/api';
 import { EventEmitter, once } from 'node:events';
 import { pathToFileURL } from 'node:url';
 import type { Logger } from 'pino';
 import { type Agent, isAgent } from '../generator.js';
-import { JobContext, JobProcess, type RunningJobInfo, runWithJobContextAsync } from '../job.js';
+import {
+  JobContext,
+  JobProcess,
+  type RunningJobInfo,
+  runWithJobContext,
+  runWithJobContextAsync,
+} from '../job.js';
 import {
   finalizeSession,
   flushJobLogs,
+  flushJobMetrics,
   runShutdownCallbacks,
   validateSessionEndTimeout,
   waitForEntrypointShutdown,
 } from '../job_lifecycle.js';
 import { initializeLogger, log } from '../log.js';
 import { loggerOptions, setLoggerState } from '../log_core.js';
 import type { SimulationContext } from '../simulation.js';
+import { getMonitor, startMonitoring, stopMonitoring } from '../telemetry/loop_monitor.js';
 import { Future, shortuuid } from '../utils.js';
 import { defaultInitializeProcessFunc } from '../worker.js';
 import type { InferenceExecutor } from './inference_executor.js';
@@ -166,6 +175,9 @@ const startJob = (
             span.setAttribute(traceTypes.ATTR_JOB_ID, info.job.id);
             span.setAttribute(traceTypes.ATTR_AGENT_NAME, info.job.agentName);
             span.setAttribute(traceTypes.ATTR_ROOM_NAME, info.job.room?.name ?? '');
+            getMonitor()?.setReportContext(otelContext.active(), (fn) =>
+              runWithJobContext(ctx, fn),
+            );
             return func(ctx);
           },
           { name: 'job_entrypoint' },
@@ -261,6 +273,7 @@ const startJob = (
     logger.debug('initializing job runner');
     await agent.prewarm(proc);
     logger.debug('job runner initialized');
+    const loopMonitor = startMonitoring({ name: 'job' });
     safeSend({ case: 'initializeResponse', value: undefined });
 
     let job: JobTask | undefined = undefined;
@@ -320,6 +333,10 @@ const startJob = (
     process.on('message', messageHandler);
 
     await join.await;
+    // stop the monitor first so a stall from the shutdown callbacks is recorded, then export:
+    // the periodic reader gets no further turn before process.exit() below
+    if (loopMonitor) stopMonitoring(loopMonitor);
+    await flushJobMetrics(logger);
     clearTimeout(orphanedTimeout);
     process.off('message', messageHandler);
 

agents/src/job_lifecycle.test.ts

@@ -7,17 +7,20 @@ import { type JobContext, getJobContext } from './job.js';
 import {
   finalizeSession,
   flushJobLogs,
+  flushJobMetrics,
   runShutdownCallbacks,
   waitForEntrypointShutdown,
 } from './job_lifecycle.js';
-import { flushOtelLogs } from './telemetry/index.js';
+import { flushCloudMetrics, flushOtelLogs } from './telemetry/index.js';
 import type { AgentSession } from './voice/agent_session.js';
 
 vi.mock('./telemetry/index.js', () => ({
   flushOtelLogs: vi.fn(),
+  flushCloudMetrics: vi.fn(),
 }));
 
 const flushOtelLogsMock = vi.mocked(flushOtelLogs);
+const flushCloudMetricsMock = vi.mocked(flushCloudMetrics);
 
 function createLogger(): Logger {
   return {
@@ -339,3 +342,36 @@ describe('flushJobLogs', () => {
     expect(logger.error).toHaveBeenCalledWith({ error }, 'Failed to flush OTEL logs');
   });
 });
+
+describe('flushJobMetrics', () => {
+  it('exports the pending cloud metrics', async () => {
+    flushCloudMetricsMock.mockResolvedValue();
+    await flushJobMetrics(createLogger());
+    expect(flushCloudMetricsMock).toHaveBeenCalledOnce();
+  });
+
+  it('stops waiting after the metric flush timeout', async () => {
+    vi.useFakeTimers();
+    flushCloudMetricsMock.mockReturnValue(new Promise<void>(() => {}));
+    const logger = createLogger();
+    const completion = flushJobMetrics(logger);
+
+    await vi.advanceTimersByTimeAsync(10_000);
+    await completion;
+
+    expect(logger.error).toHaveBeenCalledWith(
+      { timeout: 10_000 },
+      'OTEL metric flush timed out; proceeding with job shutdown',
+    );
+  });
+
+  it('continues after logging metric exporter errors', async () => {
+    const error = new Error('exporter failed');
+    const logger = createLogger();
+    flushCloudMetricsMock.mockRejectedValue(error);
+
+    await flushJobMetrics(logger);
+
+    expect(logger.error).toHaveBeenCalledWith({ error }, 'Failed to flush OTEL metrics');
+  });
+});

agents/src/job_lifecycle.ts

@@ -3,13 +3,14 @@
 // SPDX-License-Identifier: Apache-2.0
 import type { Logger } from 'pino';
 import { type JobContext, runWithJobContextAsync } from './job.js';
-import { flushOtelLogs } from './telemetry/index.js';
+import { flushCloudMetrics, flushOtelLogs } from './telemetry/index.js';
 import { IdleTimeoutError, waitUntilTimeout } from './utils.js';
 
 export const DEFAULT_SESSION_END_TIMEOUT = 300 * 1000;
 const ENTRYPOINT_SHUTDOWN_TIMEOUT = 15 * 1000;
 const SESSION_CLOSE_TIMEOUT = 60 * 1000;
 const OTEL_LOG_FLUSH_TIMEOUT = 10 * 1000;
+const OTEL_METRIC_FLUSH_TIMEOUT = 10 * 1000;
 const MAX_TIMER_TIMEOUT = 2_147_483_647;
 
 type SessionEndCallback = (ctx: JobContext) => unknown;
@@ -124,3 +125,12 @@ export async function flushJobLogs(logger: Logger): Promise<void> {
     lateReject: 'OTEL log flush rejected after shutdown timeout',
   });
 }
+
+/** Export the last metrics of the job. Run it after everything that can still record one. */
+export async function flushJobMetrics(logger: Logger): Promise<void> {
+  await waitOrContinue(() => flushCloudMetrics(), OTEL_METRIC_FLUSH_TIMEOUT, logger, {
+    timeout: 'OTEL metric flush timed out; proceeding with job shutdown',
+    error: 'Failed to flush OTEL metrics',
+    lateReject: 'OTEL metric flush rejected after shutdown timeout',
+  });
+}

agents/src/telemetry/index.ts

@@ -25,9 +25,11 @@ export {
 } from './pino_otel_transport.js';
 export * as genAI from './gen_ai.js';
 export { REDACTED_EXCEPTION_MESSAGE } from './redaction.js';
+export * as loopMonitor from './loop_monitor.js';
 export * as traceTypes from './trace_types.js';
 export {
   FanoutSpanProcessor,
+  flushCloudMetrics,
   flushOtelLogs,
   setTracerProvider,
   setupCloudTracer,