Group pipeline runs into Langfuse sessions, tagged by trigger source

Each pipeline invocation (runAsync/searchAsync) now gets a fresh
sessionId, wrapped via the new Tracer.runWithSession() (implemented
with @langfuse/tracing's propagateAttributes, since v5 has no direct
sessionId/tags option on startObservation) so it shows up in
Langfuse's Sessions view. The session is also tagged with what
triggered the run: "cron", "api-list", "api-trigger", "api-search".

Verified: full test suite (79/79) passing, tsc/eslint clean, real
docker compose build + container smoke test against /health.

src/app/controllers/RecommendationsController.ts

@@ -40,7 +40,7 @@ export class RecommendationsController extends Controller {
           : this.config.LLM_ENABLED;
 
     try {
-      const recommendations = await this.pipeline.runAsync({ count, useLlm });
+      const recommendations = await this.pipeline.runAsync({ count, useLlm }, { trigger: 'api-list' });
       reply.send(recommendations);
     } catch (err) {
       reply.code(500).send({ error: String(err) });
@@ -49,10 +49,10 @@ export class RecommendationsController extends Controller {
 
   private async handleTrigger(_request: FastifyRequest, reply: FastifyReply) {
     try {
-      const recommendations = await this.pipeline.runAsync({
+      const recommendations = await this.pipeline.runAsync(
-        count: this.config.RECOMMEND_COUNT,
+        { count: this.config.RECOMMEND_COUNT, useLlm: this.config.LLM_ENABLED },
-        useLlm: this.config.LLM_ENABLED,
+        { trigger: 'api-trigger' }
-      });
+      );
       if (this.config.RADARR_ENABLED) {
         await this.pipeline.autoRequest(recommendations);
       }
@@ -73,7 +73,7 @@ export class RecommendationsController extends Controller {
       return;
     }
     try {
-      const recommendations = await this.pipeline.searchAsync(query, this.config.RECOMMEND_COUNT);
+      const recommendations = await this.pipeline.searchAsync(query, this.config.RECOMMEND_COUNT, 'api-search');
       reply.send(recommendations);
     } catch (err) {
       reply.code(500).send({ error: String(err) });

src/app/cron.ts

@@ -9,10 +9,10 @@ export function scheduleCron(config: AppConfig, pipeline: RecommendationPipeline
   cron.schedule(config.CRON_SCHEDULE, async () => {
     logger.info('Cron: running recommendation pipeline');
     try {
-      const recommendations = await pipeline.runAsync({
+      const recommendations = await pipeline.runAsync(
-        count: config.RECOMMEND_COUNT,
+        { count: config.RECOMMEND_COUNT, useLlm: config.LLM_ENABLED },
-        useLlm: config.LLM_ENABLED,
+        { trigger: 'cron' }
-      });
+      );
       if (config.RADARR_ENABLED) {
         await pipeline.autoRequest(recommendations);
       }

src/app/pipelines/RecommendationPipeline.ts

@@ -1,3 +1,4 @@
+import { randomUUID } from 'node:crypto';
 import { LlmPipeline } from '../../core/llm/LlmPipeline';
 import type { RunContext } from '../../core/llm/RunContext';
 import type { JellyfinClient } from '../clients/jellyfin/JellyfinClient';
@@ -114,8 +115,16 @@ export class RecommendationPipeline extends LlmPipeline<RecommendationRequest, R
     }
   }
 
-  async searchAsync(query: string, count: number): Promise<Recommendation[]> {
+  async searchAsync(query: string, count: number, trigger?: string): Promise<Recommendation[]> {
-    const trace = this.tracer.startTrace(`${this.name}-search`);
+    const sessionId = randomUUID();
+    return this.tracer.runWithSession(
+      { sessionId, tags: trigger ? [trigger] : undefined },
+      () => this.runSearch(query, count, trigger)
+    );
+  }
+
+  private async runSearch(query: string, count: number, trigger?: string): Promise<Recommendation[]> {
+    const trace = this.tracer.startTrace(`${this.name}-search`, { trigger });
     const ctx: RunContext = { trace, logger: this.logger };
     try {
       let library: Movie[] = [];

src/core/langfuse/LangfuseTracer.ts

@@ -1,9 +1,15 @@
 import { NodeSDK } from '@opentelemetry/sdk-node';
 import { LangfuseSpanProcessor } from '@langfuse/otel';
-import { startObservation } from '@langfuse/tracing';
+import { propagateAttributes, startObservation } from '@langfuse/tracing';
 import type { LangfuseGeneration, LangfuseSpan } from '@langfuse/tracing';
 import { Tracer } from '../llm/Tracer';
-import type { GenerationHandle, GenerationInput, GenerationOutput, TraceHandle } from '../llm/Tracer';
+import type {
+  GenerationHandle,
+  GenerationInput,
+  GenerationOutput,
+  SessionInfo,
+  TraceHandle,
+} from '../llm/Tracer';
 
 export class LangfuseTracer extends Tracer {
   private readonly processor: LangfuseSpanProcessor;
@@ -48,6 +54,10 @@ export class LangfuseTracer extends Tracer {
     await this.processor.forceFlush();
   }
 
+  runWithSession<T>(session: SessionInfo, fn: () => Promise<T>): Promise<T> {
+    return propagateAttributes({ sessionId: session.sessionId, tags: session.tags }, fn);
+  }
+
   async shutdown(): Promise<void> {
     await this.sdk.shutdown();
   }

src/core/llm/LlmPipeline.ts

@@ -1,8 +1,14 @@
+import { randomUUID } from 'node:crypto';
 import type { Logger } from '../logging/logger';
 import type { LlmTool } from './LlmTool';
 import type { RunContext } from './RunContext';
 import type { Tracer } from './Tracer';
 
+export interface PipelineRunOptions {
+  /** What triggered this run (e.g. "cron", "api-list") — attached as a Langfuse tag. */
+  trigger?: string;
+}
+
 export abstract class LlmPipeline<TInput, TOutput> {
   constructor(
     protected readonly tracer: Tracer,
@@ -12,14 +18,20 @@ export abstract class LlmPipeline<TInput, TOutput> {
   protected abstract readonly name: string;
   protected abstract runInternal(input: TInput, ctx: RunContext): Promise<TOutput>;
 
-  async runAsync(input: TInput): Promise<TOutput> {
+  async runAsync(input: TInput, options: PipelineRunOptions = {}): Promise<TOutput> {
-    const trace = this.tracer.startTrace(this.name);
+    const sessionId = randomUUID();
-    const ctx: RunContext = { trace, logger: this.logger };
+    return this.tracer.runWithSession(
-    try {
+      { sessionId, tags: options.trigger ? [options.trigger] : undefined },
-      return await this.runInternal(input, ctx);
+      async () => {
-    } finally {
+        const trace = this.tracer.startTrace(this.name, { trigger: options.trigger });
-      await this.tracer.flush();
+        const ctx: RunContext = { trace, logger: this.logger };
-    }
+        try {
+          return await this.runInternal(input, ctx);
+        } finally {
+          await this.tracer.flush();
+        }
+      }
+    );
   }
 
   protected runTool<TToolInput, TToolOutput>(

src/core/llm/Tracer.ts

@@ -20,7 +20,18 @@ export interface TraceHandle {
   startGeneration(input: GenerationInput): GenerationHandle;
 }
 
+export interface SessionInfo {
+  sessionId: string;
+  tags?: string[];
+}
+
 export abstract class Tracer {
   abstract startTrace(name: string, metadata?: Record<string, unknown>): TraceHandle;
   abstract flush(): Promise<void>;
+
+  /**
+   * Runs `fn` with `session` (sessionId/tags) attached to every trace/span
+   * created inside it, so multiple runs can be grouped in the Langfuse UI.
+   */
+  abstract runWithSession<T>(session: SessionInfo, fn: () => Promise<T>): Promise<T>;
 }

test/app/controllers/RecommendationsController.test.ts

@@ -17,7 +17,7 @@ describe('RecommendationsController', () => {
 
     const res = await app.inject({ method: 'GET', url: '/recommendations' });
     expect(res.statusCode).toBe(200);
-    expect(runAsync).toHaveBeenCalledWith({ count: 10, useLlm: true });
+    expect(runAsync).toHaveBeenCalledWith({ count: 10, useLlm: true }, { trigger: 'api-list' });
   });
 
   it('GET /recommendations?count= and force_llm/force_tmdb override config', async () => {
@@ -25,7 +25,7 @@ describe('RecommendationsController', () => {
     const app = makeApp({ RECOMMEND_COUNT: 10, LLM_ENABLED: false }, { runAsync } as unknown as RecommendationPipeline);
 
     await app.inject({ method: 'GET', url: '/recommendations?count=3&force_llm=true' });
-    expect(runAsync).toHaveBeenCalledWith({ count: 3, useLlm: true });
+    expect(runAsync).toHaveBeenCalledWith({ count: 3, useLlm: true }, { trigger: 'api-list' });
   });
 
   it('falls back to the configured count for a malformed or non-positive count query param', async () => {
@@ -33,13 +33,13 @@ describe('RecommendationsController', () => {
     const app = makeApp({ RECOMMEND_COUNT: 10, LLM_ENABLED: true }, { runAsync } as unknown as RecommendationPipeline);
 
     await app.inject({ method: 'GET', url: '/recommendations?count=abc' });
-    expect(runAsync).toHaveBeenCalledWith({ count: 10, useLlm: true });
+    expect(runAsync).toHaveBeenCalledWith({ count: 10, useLlm: true }, { trigger: 'api-list' });
 
     await app.inject({ method: 'GET', url: '/recommendations?count=-5' });
-    expect(runAsync).toHaveBeenCalledWith({ count: 10, useLlm: true });
+    expect(runAsync).toHaveBeenCalledWith({ count: 10, useLlm: true }, { trigger: 'api-list' });
 
     await app.inject({ method: 'GET', url: '/recommendations?count=0' });
-    expect(runAsync).toHaveBeenCalledWith({ count: 10, useLlm: true });
+    expect(runAsync).toHaveBeenCalledWith({ count: 10, useLlm: true }, { trigger: 'api-list' });
   });
 
   it('POST /recommendations/trigger auto-requests via Radarr when enabled', async () => {
@@ -75,6 +75,6 @@ describe('RecommendationsController', () => {
       payload: { query: 'sci-fi like Blade Runner' },
     });
     expect(res.statusCode).toBe(200);
-    expect(searchAsync).toHaveBeenCalledWith('sci-fi like Blade Runner', 10);
+    expect(searchAsync).toHaveBeenCalledWith('sci-fi like Blade Runner', 10, 'api-search');
   });
 });

test/app/pipelines/RecommendationPipeline.test.ts

@@ -25,6 +25,7 @@ function makeTracer() {
   const tracer = {
     startTrace: vi.fn().mockReturnValue(trace),
     flush: vi.fn().mockResolvedValue(undefined),
+    runWithSession: vi.fn().mockImplementation((_session: unknown, fn: () => unknown) => fn()),
   } as unknown as Tracer;
   return tracer;
 }

test/core/langfuse/LangfuseTracer.test.ts

@@ -1,23 +1,33 @@
 import { describe, expect, it, vi } from 'vitest';
 
-const { forceFlush, shutdown, sdkStart, sdkShutdown, generationUpdate, generationEnd, childStartObservation, startObservation } =
+const {
-  vi.hoisted(() => {
+  forceFlush,
-    const generationUpdate = vi.fn();
+  shutdown,
-    const generationEnd = vi.fn();
+  sdkStart,
-    const generation = { update: generationUpdate, end: generationEnd };
+  sdkShutdown,
-    const childStartObservation = vi.fn().mockReturnValue(generation);
+  generationUpdate,
-    const rootSpan = { traceId: 'lf-trace-1', startObservation: childStartObservation };
+  generationEnd,
-    return {
+  childStartObservation,
-      forceFlush: vi.fn().mockResolvedValue(undefined),
+  startObservation,
-      shutdown: vi.fn().mockResolvedValue(undefined),
+  propagateAttributes,
-      sdkStart: vi.fn(),
+} = vi.hoisted(() => {
-      sdkShutdown: vi.fn().mockResolvedValue(undefined),
+  const generationUpdate = vi.fn();
-      generationUpdate,
+  const generationEnd = vi.fn();
-      generationEnd,
+  const generation = { update: generationUpdate, end: generationEnd };
-      childStartObservation,
+  const childStartObservation = vi.fn().mockReturnValue(generation);
-      startObservation: vi.fn().mockReturnValue(rootSpan),
+  const rootSpan = { traceId: 'lf-trace-1', startObservation: childStartObservation };
-    };
+  return {
-  });
+    forceFlush: vi.fn().mockResolvedValue(undefined),
+    shutdown: vi.fn().mockResolvedValue(undefined),
+    sdkStart: vi.fn(),
+    sdkShutdown: vi.fn().mockResolvedValue(undefined),
+    generationUpdate,
+    generationEnd,
+    childStartObservation,
+    startObservation: vi.fn().mockReturnValue(rootSpan),
+    propagateAttributes: vi.fn().mockImplementation((_attrs: unknown, fn: () => unknown) => fn()),
+  };
+});
 
 vi.mock('@langfuse/otel', () => ({
   LangfuseSpanProcessor: vi.fn().mockImplementation(() => ({ forceFlush, shutdown })),
@@ -27,7 +37,7 @@ vi.mock('@opentelemetry/sdk-node', () => ({
   NodeSDK: vi.fn().mockImplementation(() => ({ start: sdkStart, shutdown: sdkShutdown })),
 }));
 
-vi.mock('@langfuse/tracing', () => ({ startObservation }));
+vi.mock('@langfuse/tracing', () => ({ startObservation, propagateAttributes }));
 
 import { LangfuseTracer } from '../../../src/core/langfuse/LangfuseTracer';
 
@@ -74,4 +84,14 @@ describe('LangfuseTracer', () => {
     await tracer.shutdown();
     expect(sdkShutdown).toHaveBeenCalledTimes(1);
   });
+
+  it('runWithSession propagates sessionId/tags to the callback via propagateAttributes and returns its result', async () => {
+    const tracer = new LangfuseTracer({ publicKey: 'pk', secretKey: 'sk', baseUrl: 'http://lf' });
+    const fn = vi.fn().mockResolvedValue('done');
+
+    const result = await tracer.runWithSession({ sessionId: 'session-1', tags: ['cron'] }, fn);
+
+    expect(propagateAttributes).toHaveBeenCalledWith({ sessionId: 'session-1', tags: ['cron'] }, fn);
+    expect(result).toBe('done');
+  });
 });

test/core/llm/LlmPipeline.test.ts

@@ -23,24 +23,44 @@ class ThrowingPipeline extends LlmPipeline<void, void> {
 function makeTracer(trace: TraceHandle) {
   const flush = vi.fn().mockResolvedValue(undefined);
   const startTrace = vi.fn().mockReturnValue(trace);
-  const tracer = { startTrace, flush } as unknown as Tracer;
+  const runWithSession = vi.fn().mockImplementation((_session: unknown, fn: () => unknown) => fn());
-  return { tracer, startTrace, flush };
+  const tracer = { startTrace, flush, runWithSession } as unknown as Tracer;
+  return { tracer, startTrace, flush, runWithSession };
 }
 
 describe('LlmPipeline', () => {
-  it('opens one trace, runs runInternal with it, and flushes afterward', async () => {
+  it('runs the whole call inside a session scoped to a fresh session id, opens one trace, runs runInternal, and flushes afterward', async () => {
     const trace = { id: 'trace-1', startGeneration: vi.fn() } as unknown as TraceHandle;
-    const { tracer, startTrace, flush } = makeTracer(trace);
+    const { tracer, startTrace, flush, runWithSession } = makeTracer(trace);
     const logger = {} as unknown as Logger;
 
     const pipeline = new SumPipeline(tracer, logger);
     const result = await pipeline.runAsync([1, 2, 3]);
 
     expect(result).toBe(6);
-    expect(startTrace).toHaveBeenCalledWith('sum-pipeline');
+    expect(runWithSession).toHaveBeenCalledWith(
+      expect.objectContaining({ sessionId: expect.any(String), tags: undefined }),
+      expect.any(Function)
+    );
+    expect(startTrace).toHaveBeenCalledWith('sum-pipeline', { trigger: undefined });
     expect(flush).toHaveBeenCalledTimes(1);
   });
 
+  it('tags the session with the given trigger', async () => {
+    const trace = { id: 'trace-1', startGeneration: vi.fn() } as unknown as TraceHandle;
+    const { tracer, startTrace, runWithSession } = makeTracer(trace);
+    const logger = {} as unknown as Logger;
+
+    const pipeline = new SumPipeline(tracer, logger);
+    await pipeline.runAsync([1, 2, 3], { trigger: 'cron' });
+
+    expect(runWithSession).toHaveBeenCalledWith(
+      expect.objectContaining({ tags: ['cron'] }),
+      expect.any(Function)
+    );
+    expect(startTrace).toHaveBeenCalledWith('sum-pipeline', { trigger: 'cron' });
+  });
+
   it('still flushes the trace when runInternal throws', async () => {
     const trace = { id: 'trace-2', startGeneration: vi.fn() } as unknown as TraceHandle;
     const { tracer, flush } = makeTracer(trace);