feat(langfuse): add opt-in Langfuse observability integration

Add Langfuse as an OTel trace transport for MCP request lifecycle tracking.

Changes:
- New src/lib/langfuse-transport.ts: initializes LangfuseSpanProcessor when
  ASK262_LANGFUSE_ENABLED=true with valid credentials
- src/lib/tracing.ts: add metadata to TraceContext, propagate it to child spans,
  use startActiveSpan for proper OTel context, add setupTracing() entry point,
  add getSessionMetadata() for version/transport metadata
- src/lib/logger.ts: include TraceContext.metadata in JSON log mixin
- src/mcp-server-{http,stdio}.ts: call setupTracing() at startup, pass
  version/transport metadata through withSpan/withSpanContext
- src/lib/fireworks-embeddings.ts: wrap API calls in withSpan() with gen_ai
  attributes for Langfuse generation/cost tracking
- .env.example: add LANGFUSE_* configuration variables
- AGENTS.md: document Langfuse integration
- package.json: add @langfuse/tracing, @langfuse/otel, @opentelemetry/sdk-node

src/lib/fireworks-embeddings.ts

@@ -1,5 +1,7 @@
 import { Embeddings, type EmbeddingsParams } from "@langchain/core/embeddings";
+import { trace } from "@opentelemetry/api";
 import { LogOperation, logger } from "./logger.js";
+import { withSpan } from "./tracing.js";
 
 /**
  * Interface for FireworksEmbeddings parameters.
@@ -180,37 +182,59 @@ export class FireworksEmbeddings extends Embeddings {
 
   /**
    * Make the actual API call to Fireworks for embeddings.
+   * Wrapped in an OTel span with GenAI attributes for Langfuse cost tracking.
    */
   private async embedBatch(documents: string[]): Promise<number[][]> {
-    const url = `${this.baseUrl}/embeddings`;
+    return await withSpan(
+      LogOperation.PROCESSING_EMBEDDING_BATCH,
+      { model: this.modelName, batch_size: documents.length },
+      async () => {
+        const url = `${this.baseUrl}/embeddings`;
 
-    const response = await fetch(url, {
-      method: "POST",
-      headers: {
-        Authorization: `Bearer ${this.apiKey}`,
-        "Content-Type": "application/json",
-        Accept: "application/json",
+        const response = await fetch(url, {
+          method: "POST",
+          headers: {
+            Authorization: `Bearer ${this.apiKey}`,
+            "Content-Type": "application/json",
+            Accept: "application/json",
+          },
+          body: JSON.stringify({
+            model: this.modelName,
+            input: documents,
+          }),
+        });
+
+        if (!response.ok) {
+          const errorText = await response.text();
+          throw new Error(
+            `Fireworks API error: ${response.status} ${response.statusText} - ${errorText}`,
+          );
+        }
+
+        const data = (await response.json()) as FireworksEmbeddingResponse;
+
+        // Extract embeddings from response
+        // Fireworks returns embeddings in the same order as input
+        const embeddings = data.data.map((item) => item.embedding);
+
+        // Set GenAI attributes on the active span for Langfuse generation tracking
+        const activeSpan = trace.getActiveSpan();
+        if (activeSpan) {
+          activeSpan.setAttribute("gen_ai.system", "fireworks");
+          activeSpan.setAttribute("gen_ai.request.model", this.modelName);
+          activeSpan.setAttribute(
+            "gen_ai.usage.input_tokens",
+            data.usage.prompt_tokens,
+          );
+          activeSpan.setAttribute(
+            "gen_ai.usage.output_tokens",
+            data.usage.completion_tokens,
+          );
+        }
+
+        return embeddings;
       },
-      body: JSON.stringify({
-        model: this.modelName,
-        input: documents,
-      }),
-    });
-
-    if (!response.ok) {
-      const errorText = await response.text();
-      throw new Error(
-        `Fireworks API error: ${response.status} ${response.statusText} - ${errorText}`,
-      );
-    }
-
-    const data = (await response.json()) as FireworksEmbeddingResponse;
-
-    // Extract embeddings from response
-    // Fireworks returns embeddings in the same order as input
-    const embeddings = data.data.map((item) => item.embedding);
-
-    return embeddings;
+    );
   }
 }
 
@@ -235,5 +259,9 @@ interface FireworksEmbeddingResponse {
   usage: {
     prompt_tokens: number;
     total_tokens: number;
+    completion_tokens: number;
+    prompt_tokens_details: {
+      cached_tokens: number;
+    };
   };
 }

src/lib/langfuse-transport.ts

@@ -0,0 +1,32 @@
+import { NodeTracerProvider } from "@opentelemetry/sdk-trace-node";
+import { LangfuseSpanProcessor } from "@langfuse/otel";
+
+let provider: NodeTracerProvider | null = null;
+
+/**
+ * Initialize Langfuse OTel span processor.
+ * Called once at server startup before any spans are created.
+ */
+export function initializeLangfuseTransport(): void {
+  if (process.env.ASK262_LANGFUSE_ENABLED !== "true") return;
+
+  const publicKey = process.env.LANGFUSE_PUBLIC_KEY;
+  const secretKey = process.env.LANGFUSE_SECRET_KEY;
+
+  if (!publicKey || !secretKey) {
+    console.error(
+      "[LANGFUSE] Missing LANGFUSE_PUBLIC_KEY or LANGFUSE_SECRET_KEY, skipping",
+    );
+    return;
+  }
+
+  provider = new NodeTracerProvider({
+    spanProcessors: [
+      new LangfuseSpanProcessor({
+        // Export all spans, not just LLM-relevant ones
+        shouldExportSpan: () => true,
+      }),
+    ],
+  });
+  provider.register();
+}

src/lib/logger.ts

@@ -208,6 +208,7 @@ async function createRootLogger(): Promise<pino.Logger> {
             trace_id: traceCtx.traceId,
             span_id: traceCtx.spanId,
             parent_span_id: traceCtx.parentSpanId,
+            ...(traceCtx.metadata && { metadata: traceCtx.metadata }),
           };
         }
         return {};

src/lib/tracing.ts

@@ -10,6 +10,30 @@
 import { AsyncLocalStorage } from "node:async_hooks";
 import { randomUUID } from "node:crypto";
 import { SpanStatusCode, trace } from "@opentelemetry/api";
+import pkg from "../../package.json" with { type: "json" };
+import { initializeLangfuseTransport } from "./langfuse-transport.js";
+
+/**
+ * One-time setup for the tracing system.
+ * Registers the Langfuse OTel processor when enabled.
+ * Must be called before any spans are created.
+ */
+export function setupTracing(): void {
+  initializeLangfuseTransport();
+}
+
+/**
+ * Build session metadata for a transport.
+ * Included in both JSON logs (via TraceContext) and Langfuse traces.
+ */
+export function getSessionMetadata(
+  transport: "http" | "stdio",
+): Record<string, unknown> {
+  return {
+    version: pkg.version,
+    transport,
+  };
+}
 
 /**
  * Trace context stored in AsyncLocalStorage.
@@ -23,6 +47,8 @@ interface TraceContext {
   parentSpanId: string | null;
   /** Span depth level (0 = root) */
   depth: number;
+  /** Optional metadata propagated to JSON logs and Langfuse traces */
+  metadata?: Record<string, unknown>;
 }
 
 // AsyncLocalStorage for automatic context propagation
@@ -70,12 +96,16 @@ function generateTraceId(): string {
  * const traceCtx = createTraceContext(req.headers['x-request-id'] as string);
  * ```
  */
-export function createTraceContext(traceId?: string): TraceContext {
+export function createTraceContext(
+  traceId?: string,
+  metadata?: Record<string, unknown>,
+): TraceContext {
   return {
     traceId: traceId ?? generateTraceId(),
     spanId: generateSpanId(),
     parentSpanId: null,
     depth: 0,
+    metadata,
   };
 }
 
@@ -108,6 +138,7 @@ export async function withSpan<T>(
   attributes: Record<string, unknown> = {},
   fn: () => Promise<T>,
   traceId?: string,
+  metadata?: Record<string, unknown>,
 ): Promise<T> {
   const parentContext = getCurrentContext();
   const tracer = trace.getTracer("ask262");
@@ -119,33 +150,39 @@ export async function withSpan<T>(
         spanId: generateSpanId(),
         parentSpanId: parentContext.spanId,
         depth: parentContext.depth + 1,
+        metadata: parentContext.metadata,
       }
-    : createTraceContext(traceId);
-
-  // Create OTel span for context tracking
-  const span = tracer.startSpan(operation, {
-    attributes: {
-      ...attributes,
-      "span.depth": spanContext.depth,
-    },
-  });
+    : createTraceContext(traceId, metadata);
 
   // Store in AsyncLocalStorage for nested calls
   return traceStorage.run(spanContext, async () => {
-    try {
-      const result = await fn();
-      span.setStatus({ code: SpanStatusCode.OK });
-      return result;
-    } catch (err) {
-      span.setStatus({
-        code: SpanStatusCode.ERROR,
-        message: err instanceof Error ? err.message : String(err),
-      });
-      span.recordException(err instanceof Error ? err : new Error(String(err)));
-      throw err;
-    } finally {
-      span.end();
-    }
+    return tracer.startActiveSpan(
+      operation,
+      {
+        attributes: {
+          ...attributes,
+          "span.depth": spanContext.depth,
+        },
+      },
+      async (span) => {
+        try {
+          const result = await fn();
+          span.setStatus({ code: SpanStatusCode.OK });
+          return result;
+        } catch (err) {
+          span.setStatus({
+            code: SpanStatusCode.ERROR,
+            message: err instanceof Error ? err.message : String(err),
+          });
+          span.recordException(
+            err instanceof Error ? err : new Error(String(err)),
+          );
+          throw err;
+        } finally {
+          span.end();
+        }
+      },
+    );
   });
 }
 
@@ -216,6 +253,7 @@ export async function withSpanContext<T>(
   operation: string,
   attributes: Record<string, unknown> = {},
   fn: () => Promise<T>,
+  metadata?: Record<string, unknown>,
 ): Promise<T> {
   // Check if we're already in a context
   const existingContext = getCurrentContext();
@@ -226,36 +264,37 @@ export async function withSpanContext<T>(
   }
 
   // Create new root span with this trace ID
-  const spanContext: TraceContext = {
-    traceId,
-    spanId: generateSpanId(),
-    parentSpanId: null,
-    depth: 0,
-  };
-
+  const spanContext = createTraceContext(traceId, metadata);
   const tracer = trace.getTracer("ask262");
-  const span = tracer.startSpan(operation, {
-    attributes: {
-      ...attributes,
-      "span.depth": 0,
-    },
-  });
 
   return traceStorage.run(spanContext, async () => {
-    try {
-      const result = await fn();
-      span.setStatus({ code: SpanStatusCode.OK });
-      return result;
-    } catch (err) {
-      span.setStatus({
-        code: SpanStatusCode.ERROR,
-        message: err instanceof Error ? err.message : String(err),
-      });
-      span.recordException(err instanceof Error ? err : new Error(String(err)));
-      throw err;
-    } finally {
-      span.end();
-    }
+    return tracer.startActiveSpan(
+      operation,
+      {
+        attributes: {
+          ...attributes,
+          "span.depth": 0,
+        },
+      },
+      async (span) => {
+        try {
+          const result = await fn();
+          span.setStatus({ code: SpanStatusCode.OK });
+          return result;
+        } catch (err) {
+          span.setStatus({
+            code: SpanStatusCode.ERROR,
+            message: err instanceof Error ? err.message : String(err),
+          });
+          span.recordException(
+            err instanceof Error ? err : new Error(String(err)),
+          );
+          throw err;
+        } finally {
+          span.end();
+        }
+      },
+    );
   });
 }