fix: removed langfuse telemetry memory leaks

Author: jahilldev

api/src/agent/agent.module.ts

@@ -1,7 +1,9 @@
 import { Module } from '@nestjs/common';
+import { LangfuseModule } from '../langfuse/langfuse.module.js';
 import { HermesAgentService } from './agent.service.js';
 
 @Module({
+  imports: [LangfuseModule],
   providers: [HermesAgentService],
   exports: [HermesAgentService],
 })

api/src/agent/agent.service.ts

@@ -6,6 +6,7 @@ import { AppConfigService } from '../config/config.service.js';
 import { MetricsService } from '../metrics/metrics.service.js';
 import { type AgentRunOptions, type AgentRunResult, type AgentRunStatus } from './agent.model.js';
 import { buildSpawnSpec, spawnProcess, prepareHermesMemoryPaths } from './agent.utility.js';
+import { LangfuseService } from '../langfuse/langfuse.service.js';
 
 /**
  * Drives the Hermes Agent CLI. Each call runs `hermes -z --yolo --accept-hooks …` headless in the
@@ -20,6 +21,7 @@ export class HermesAgentService {
     private readonly prisma: PrismaService,
     private readonly config: AppConfigService,
     private readonly metrics: MetricsService,
+    private readonly langfuse: LangfuseService,
   ) {}
 
   /**
@@ -132,6 +134,7 @@ export class HermesAgentService {
       },
     });
 
+    this.langfuse.complete(run.id);
     this.metrics.recordAgentRun(opts.phase, status, raw.durationMs);
 
     this.logger.log(

api/src/langfuse/langfuse.controller.ts

@@ -118,7 +118,7 @@ export class LangfuseController {
       map((r) =>
         encode({ type: 'done', status: r.status, exitCode: r.exitCode, durationMs: r.durationMs }),
       ),
-      shareReplay({ bufferSize: 1, refCount: false }),
+      shareReplay({ bufferSize: 1, refCount: true }),
     );
 
     return concat(history$, merge(live$.pipe(takeUntil(done$)), done$));

api/src/langfuse/langfuse.service.ts

@@ -52,6 +52,7 @@ export class LangfuseService {
   complete(sessionId: string): void {
     this.subjects.get(sessionId)?.complete();
     this.subjects.delete(sessionId);
+    this.buffers.delete(sessionId);
   }
 
   observe(sessionId: string): Observable<LangfuseEvent> | null {