📊 Langfuse Tracing & Robust Ingestion#

TL;DR 🚀 Observe every agent turn safely without blocking execution or hitting body size limits. Langfuse traces are exported via OpenTelemetry (primary) with a REST fallback (backup). Large payloads (long prompt histories, git diffs, file reads) can aggregate to hit the self-hosted Next.js HTTP 413 “Body exceeded 4.5mb limit” block on batch ingestion. We fix this via OTel visibility buffers, payload truncation, and chunked batch flushes. 🔌

🧬 Instrumentation & Fallback Flow#

flowchart TD
  subgraph Agent["🤖 Pi Coding Agent Run"]
    P[Prompt Start] -->|Turn Start| T[Turn Span]
    T -->|Call LLM| G[Generation Observation]
    T -->|Call Tool| O[Tool Span]
    T -->|Turn End| E[Turn End]
  end

  subgraph Export["📡 Trace Export & Visibility Fallback"]
    E -->|agent_end| FLUSH["forceFlush() OTel Spans"]
    FLUSH -->|Wait up to 8s| POLL{"Trace visible in DB?<br/>(POLL_INTERVAL=500ms)"}
    
    POLL -->|🟢 Yes| DONE[OTel Succeeded]
    POLL -->|🔴 No| REST[REST Fallback Ingestion]
  end

  subgraph Fallback["✂️ REST Fallback Processing (pi-langfuse)"]
    REST -->|Tier 1| TRUNC["Truncate String values >200KB<br/>(Recursively prune deep objects)"]
    TRUNC -->|Tier 2| CHUNK["Slice batch into chunks of 15 items"]
    CHUNK -->|Tier 3| SEND["POST sequentially to /api/public/ingestion<br/>(Individual try/catches)"]
  end

  classDef main fill:#5865F2,stroke:#fff,stroke-width:2px,color:#fff;
  classDef helper fill:#2F3136,stroke:#7289DA,stroke-width:1px,color:#fff;
  classDef state fill:#43B581,stroke:#fff,stroke-width:1px,color:#fff;
  class E,FLUSH main;
  class POLL,REST,TRUNC,CHUNK,SEND helper;
  class DONE state;

🚨 The HTTP 413 Body Size Limit#

Self-hosted Langfuse servers use Next.js behind the hood, enforcing a default bodyParser: { sizeLimit: '4.5mb' } on the ingestion endpoint.

When a long-running agent session is flushed, the REST fallback bundles all observations (system prompt, conversation history, tool calls, and final answers) into a single batch JSON array. This payload easily aggregates to 5MB - 10MB, causing the server to reject the entire batch with statusCode: 413.

/* ❌ Next.js body-parser rejection raw body */
{
  "ok": false,
  "error": {
    "reason": "non-json",
    "statusCode": 413,
    "rawBody": "Body exceeded 4.5mb limit"
  }
}

🛡️ The Three-Tier Reliability Fix#

We implement three layers of defensive ingestion inside pi-langfuse’s REST fallback:

1️⃣ OpenTelemetry Visibility Buffer#

Self-hosted Langfuse ingests traces asynchronously (Ingestion queue ➔ Worker ➔ DB). Spans are rarely queryable within ~1.5s after flush.

• Fix: We increase OTEL_VISIBILITY_TIMEOUT_MS to 8_000ms (polling every 500ms) so the REST fallback only fires on genuine OTel exporter failures, preventing duplicate trace writes.

2️⃣ Recursive Payload Truncation (safeValue)#

No single trace input, output, or metadata object should choke the network. We recursively traverse payloads and truncate individual strings to a safe threshold.

// 🚀 why: prevents massive string values (like git diffs) from exceeding limits
function truncateString(str: string, limit = 200_000): string {
  if (str.length > limit) {
    return str.slice(0, limit) + "\n\n... [Truncated by pi-langfuse due to length limits]";
  }
  return str;
}

// 🚀 what: recursively prunes objects and limits nesting depth to avoid stack overflows
function safeValue(val: unknown, depth = 0): unknown {
  if (val === null || val === undefined) return val;
  if (typeof val === "string") return truncateString(val);
  
  if (Array.isArray(val)) {
    if (depth > 5) return "[Array truncated due to nesting depth]";
    return val.map((item) => safeValue(item, depth + 1));
  }
  if (typeof val === "object") {
    if (depth > 5) return "[Object truncated due to nesting depth]";
    const cleaned: Record<string, unknown> = {};
    for (const key of Object.keys(val)) {
      cleaned[key] = safeValue((val as any)[key], depth + 1);
    }
    return cleaned;
  }
  return val;
}

3️⃣ Chunked REST Ingestion (chunkSize = 15)#

Instead of sending the whole execution history in one request, we partition the batch.

// 🚀 how: slice the payload array and wrap each POST in an independent try/catch block
const chunkSize = 15;
for (let i = 0; i < batch.length; i += chunkSize) {
  const chunk = batch.slice(i, i + chunkSize);
  try {
    await rt.scoreClient.api.ingestion.batch({
      batch: chunk,
      metadata: {
        source: "pi-langfuse",
        fallback: "rest-ingestion",
        chunkIndex: Math.floor(i / chunkSize),
        totalChunks: Math.ceil(batch.length / chunkSize),
      },
    });
  } catch (e) {
    // 💡 non-blocking: one chunk failing doesn't abort the remaining queue flushes
    console.warn(`📊 Langfuse: Failed to ingest fallback batch chunk ${Math.floor(i / chunkSize) + 1}/${Math.ceil(batch.length / chunkSize)}`, e);
  }
}

💡 Best Practices Checklist#

• ⏳ Deferred Flush: Always defer the final Langfuse shutdown/flush using setTimeout(shutdownRuntime, 0) on agent_end. Never let telemetry block the main agent execution turn or increase user latency.
• 🛡️ PII & Credential Masking: Never log environment arrays or raw configs. Function arguments containing secret keys must be stripped before logging inputs.
• 🏷️ Clean Trace Naming: Choose descriptive names (code-generation, tool-execution) over generic IDs to ensure UI filtering is actually usable.
• 🔗 Session Attribution: Bind sessionId on traces using the Discord thread ID. This groups conversational multi-turn histories in the Langfuse Sessions view.