force flush and 1 sec interval (#468)

* force flush

* changeset

* changes

Author: shagun-singh-inkeep

.changeset/violet-carrots-change.md

@@ -0,0 +1,5 @@
+---
+"@inkeep/agents-run-api": patch
+---
+
+force flush

agents-run-api/src/agents/Agent.ts

@@ -60,6 +60,7 @@ import { toolSessionManager } from './ToolSessionManager';
 import type { SystemPromptV1 } from './types';
 import { Phase1Config } from './versions/v1/Phase1Config';
 import { Phase2Config } from './versions/v1/Phase2Config';
+import { defaultBatchProcessor } from '../instrumentation';
 
 /**
  * Creates a stopWhen condition that stops when any tool call name starts with the given prefix
@@ -749,6 +750,7 @@ export class Agent {
         requestContext: requestContext || {},
         tenantId: this.config.tenantId,
       });
+      await defaultBatchProcessor.forceFlush();
 
       // Add built-in variables to resolved context
       const contextWithBuiltins = {
@@ -1471,6 +1473,7 @@ export class Agent {
               throw err;
             } finally {
               childSpan.end();
+              await defaultBatchProcessor.forceFlush();
             }
           }
         );
@@ -2041,6 +2044,7 @@ ${output}${structureHintsFormatted}`;
         // Mark span as successful
         span.setStatus({ code: SpanStatusCode.OK });
         span.end();
+        await defaultBatchProcessor.forceFlush();
 
         // Format response - handle object vs text responses differently
         // Only format if we don't already have formattedContent from streaming

agents-run-api/src/handlers/executionHandler.ts

@@ -20,6 +20,7 @@ import { agentInitializingOp, completionOp, errorOp } from '../utils/agent-opera
 import type { StreamHelper } from '../utils/stream-helpers.js';
 import { MCPStreamHelper } from '../utils/stream-helpers.js';
 import { registerStreamHelper, unregisterStreamHelper } from '../utils/stream-registry.js';
+import { defaultBatchProcessor } from '../instrumentation';
 
 const logger = getLogger('ExecutionHandler');
 
@@ -453,6 +454,7 @@ export class ExecutionHandler {
               throw error;
             } finally {
               span.end();
+              await defaultBatchProcessor.forceFlush();
             }
           });
         }

agents-run-api/src/instrumentation.ts

@@ -18,7 +18,9 @@ import { ATTR_SERVICE_NAME } from '@opentelemetry/semantic-conventions';
 
 const otlpExporter = new OTLPTraceExporter();
 
-export const defaultBatchProcessor = new BatchSpanProcessor(otlpExporter);
+export const defaultBatchProcessor = new BatchSpanProcessor(otlpExporter, {
+  scheduledDelayMillis: 1000,
+});
 
 export const defaultResource = resourceFromAttributes({
   [ATTR_SERVICE_NAME]: 'inkeep-agents-run-api',

agents-run-api/src/routes/chatDataStream.ts

@@ -22,6 +22,7 @@ import { ExecutionHandler } from '../handlers/executionHandler';
 import { getLogger } from '../logger';
 import { errorOp } from '../utils/agent-operations';
 import { createVercelStreamHelper } from '../utils/stream-helpers';
+import { defaultBatchProcessor } from '../instrumentation';
 
 type AppVariables = {
   credentialStores: CredentialStoreRegistry;
@@ -169,6 +170,7 @@ app.openapi(chatDataStreamRoute, async (c) => {
       dbClient,
       credentialStores,
     });
+    await defaultBatchProcessor.forceFlush();
 
     // Store last user message
     const lastUserMessage = body.messages.filter((m: any) => m.role === 'user').slice(-1)[0];

agents-run-api/src/services/GraphSession.ts

@@ -19,6 +19,7 @@ import { getStreamHelper } from '../utils/stream-registry';
 import { setSpanWithError, tracer } from '../utils/tracer';
 import { ArtifactParser } from './ArtifactParser';
 import { ArtifactService } from './ArtifactService';
+import { defaultBatchProcessor } from '../instrumentation';
 
 const logger = getLogger('GraphSession');
 
@@ -958,6 +959,7 @@ ${this.statusUpdateState?.config.prompt?.trim() || ''}`;
           return { summaries: [] };
         } finally {
           span.end();
+          await defaultBatchProcessor.forceFlush();
         }
       }
     );
@@ -1472,6 +1474,7 @@ Make it specific and relevant.`;
         } finally {
           // Always end the main span
           span.end();
+          await defaultBatchProcessor.forceFlush();
         }
       }
     );

agents-run-api/src/services/ResponseFormatter.ts

@@ -3,6 +3,7 @@ import { getLogger } from '../logger';
 import { ArtifactParser, type StreamPart } from '../services/ArtifactParser';
 import { tracer, setSpanWithError } from '../utils/tracer';
 import { graphSessionManager } from '../services/GraphSession';
+import { defaultBatchProcessor } from '../instrumentation';
 
 const logger = getLogger('ResponseFormatter');
 
@@ -83,6 +84,7 @@ export class ResponseFormatter {
         };
       } finally {
         span.end();
+        await defaultBatchProcessor.forceFlush();
       }
     });
   }
@@ -149,6 +151,7 @@ export class ResponseFormatter {
         return { text: responseText };
       } finally {
         span.end();
+        await defaultBatchProcessor.forceFlush();
       }
     });
   }