LCORE-5254: Langfuse fullsend setup

.fullsend/scripts/langfuse-post-run.py

@@ -0,0 +1,268 @@
+# Generated by uie-fullsend langfuse-setup. Uses trusted default-branch code.
+"""Add generation observations to native Fullsend traces; Python standard library.
+
+Derived from fullsend-component/scripts/lib/langfuse.sh's generation bridge.
+Only model/usage and CI metadata are exported; artifacts never supply code,
+credentials, an endpoint, or prompt/response content.
+"""
+import argparse
+import hashlib
+import json
+import math
+import os
+from pathlib import Path
+import re
+import secrets
+import sys
+import time
+import urllib.error
+import urllib.parse
+import urllib.request
+
+
+def attributes(entries):
+    if isinstance(entries, dict):
+        return dict(entries)
+    result = {}
+    for entry in entries or []:
+        value = entry.get("value", {})
+        for kind in ("stringValue", "intValue", "doubleValue", "boolValue"):
+            if kind in value:
+                result[entry["key"]] = value[kind]
+                break
+    return result
+
+
+def otlp_attributes(values):
+    result = []
+    for key, value in values.items():
+        if value is None:
+            continue
+        if isinstance(value, bool):
+            encoded = {"boolValue": value}
+        elif isinstance(value, int):
+            encoded = {"intValue": str(value)}
+        elif isinstance(value, float):
+            encoded = {"doubleValue": value}
+        elif isinstance(value, list):
+            encoded = {"arrayValue": {"values": [{"stringValue": str(item)} for item in value]}}
+        else:
+            encoded = {"stringValue": str(value)}
+        result.append({"key": key, "value": encoded})
+    return result
+
+
+def add_filterable_context(values):
+    for key in ("project_key", "cost_center", "github.repository", "github.run_id"):
+        if values.get(key):
+            name = key.replace(".", "_")
+            for level in ("trace", "observation"):
+                values[f"langfuse.{level}.metadata.{name}"] = values[key]
+    tags = [f"{label}:{values[key]}" for key, label in (
+        ("gen_ai.agent.name", "stage"),
+        ("project_key", "project_key"), ("cost_center", "cost_center"),
+        ("github.repository", "repo"),
+    ) if values.get(key)]
+    if tags:
+        values["langfuse.trace.tags"] = tags
+
+
+def number(value, integer=False):
+    if value is None:
+        return None
+    parsed = float(value)
+    if not math.isfinite(parsed) or parsed < 0:
+        raise ValueError("Invalid usage value")
+    return int(parsed) if integer else parsed
+
+
+def read_spans(path):
+    # The exporter writes one OTLP resourceSpans envelope per JSONL line.
+    if path.stat().st_size > 32 * 1024 * 1024:
+        raise ValueError("Telemetry exceeds size limit")
+    spans = {}
+    with path.open(encoding="utf-8") as stream:
+        for line in stream:
+            if not line.strip():
+                continue
+            record = json.loads(line)
+            for resource in record.get("resourceSpans", []):
+                for scope in resource.get("scopeSpans", []):
+                    for span in scope.get("spans", []):
+                        spans[(span.get("traceId"), span.get("spanId"))] = span
+    return list(spans.values())
+
+
+def build_payload(path, context):
+    spans = read_spans(path)
+    agents = [span for span in spans if attributes(span.get("attributes")).get("gen_ai.operation.name") == "invoke_agent"]
+    metrics = {}
+    metrics_path = path.with_name("metrics.json")
+    if metrics_path.is_file() and metrics_path.stat().st_size < 1024 * 1024:
+        metrics = json.loads(metrics_path.read_text(encoding="utf-8"))
+        if not isinstance(metrics, dict):
+            metrics = {}
+    # Use metrics only for a single agent; don't multiply whole-run totals across agents.
+    if len(agents) != 1:
+        metrics = {}
+    output = []
+    for agent in agents:
+        source = attributes(agent.get("attributes"))
+        if any(span.get("parentSpanId") == agent.get("spanId") and
+               attributes(span.get("attributes")).get("langfuse.observation.type") == "generation" for span in spans):
+            continue
+        model = source.get("gen_ai.request.model") or metrics.get("model")
+        if not model:
+            continue  # Never invent a model (the GitLab reference has a Gemini fallback).
+        trace_id, parent_id = agent.get("traceId") or "", agent.get("spanId") or ""
+        if not re.fullmatch(r"[0-9a-fA-F]{32}", trace_id) or not re.fullmatch(r"[0-9a-fA-F]{16}", parent_id):
+            raise ValueError("Invalid trace identity")
+        start, end = int(agent["startTimeUnixNano"]), int(agent["endTimeUnixNano"])
+        if start < 0 or end < start:
+            raise ValueError("Invalid span timestamps")
+        model = str(model).removeprefix("google-vertex/")
+        values = dict(context)
+        values.update({"langfuse.observation.type": "generation", "gen_ai.operation.name": "generate_content",
+                       "gen_ai.request.model": model, "langfuse.observation.model.name": model,
+                       "gen_ai.system": source.get("gen_ai.system")})
+        usage = metrics.get("token_usage") or {}
+        for attr, fallback in (("gen_ai.usage.input_tokens", "input"), ("gen_ai.usage.output_tokens", "output"),
+                               ("gen_ai.usage.cache_read.input_tokens", "cache_read"),
+                               ("gen_ai.usage.reasoning.output_tokens", "reasoning")):
+            value = source.get(attr)
+            if value is None and attr.endswith("reasoning.output_tokens"):
+                value = source.get("gen_ai.usage.reasoning_tokens")
+            if value is None:
+                value = usage.get(fallback)
+            if value is not None:
+                values[attr] = number(value, integer=True)
+        cost = source.get("fullsend.cost_usd")
+        if cost is None:
+            cost = metrics.get("total_cost_usd")
+        if cost is not None:
+            values["gen_ai.usage.cost"] = number(cost)
+            values["langfuse.observation.cost_details"] = json.dumps({"total": number(cost)})
+        for attr, fallback in (("fullsend.num_turns", "iterations"), ("fullsend.tool_calls", None)):
+            value = source.get(attr)
+            if value is None and fallback is not None:
+                value = metrics.get(fallback)
+            if value is not None:
+                values[attr] = number(value, integer=True)
+        stage = source.get("gen_ai.agent.name") or next(
+            (parent.name.removeprefix("fullsend-") for parent in path.parents
+             if parent.name.startswith("fullsend-")), "run")
+        values["gen_ai.agent.name"] = stage
+        trace_name = f"{stage}: {context.get('github.repository', 'fullsend')}"
+        if context.get("github.run_id"):
+            trace_name += f"#{context['github.run_id']}"
+        values["langfuse.trace.name"] = trace_name
+        span_id = hashlib.sha256(f"fullsend-langfuse-generation-v1:{trace_id}:{parent_id}".encode()).hexdigest()[:16]
+        add_filterable_context(values)
+        output.append({"traceId": trace_id, "spanId": span_id, "parentSpanId": parent_id,
+                       "name": f"{stage.capitalize()} Generation: {model}", "kind": 1,
+                       "startTimeUnixNano": str(start), "endTimeUnixNano": str(end),
+                       "attributes": otlp_attributes(values), "status": {"code": (agent.get("status") or {}).get("code", 1)}})
+    return {"resourceSpans": [{"resource": {"attributes": otlp_attributes(context)},
+            "scopeSpans": [{"scope": {"name": "fullsend-generation-bridge", "version": "1.0.0"}, "spans": output}]}]}
+
+
+def verification_payload(trace_id, context):
+    if not re.fullmatch(r"[0-9a-fA-F]{32}", trace_id):
+        raise ValueError("Invalid verification trace ID")
+    if not context.get("cost_center") or context["cost_center"].lower() == "unassigned":
+        raise ValueError("Missing configured cost_center resource attribute")
+    now = time.time_ns()
+    values = dict(context)
+    values["langfuse.trace.name"] = "Fullsend Langfuse verification"
+    add_filterable_context(values)
+    span = {"traceId": trace_id.lower(), "spanId": secrets.token_hex(8),
+            "name": "Fullsend Langfuse verification", "kind": 1,
+            "startTimeUnixNano": str(now), "endTimeUnixNano": str(now + 1_000_000),
+            "attributes": otlp_attributes(values), "status": {"code": 1}}
+    return {"resourceSpans": [{"resource": {"attributes": otlp_attributes(context)},
+            "scopeSpans": [{"scope": {"name": "fullsend-generation-bridge", "version": "1.0.0"},
+                            "spans": [span]}]}]}
+
+
+def export(payload, env):
+    endpoint = env.get("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", "")
+    url = urllib.parse.urlsplit(endpoint)
+    is_loopback_http = url.scheme == "http" and url.hostname in ("localhost", "127.0.0.1", "::1")
+    if (url.scheme != "https" and not is_loopback_http) or not url.hostname or url.username or url.password or url.query or url.fragment:
+        raise ValueError("Invalid configured export endpoint")
+    headers = {"Content-Type": "application/json", "x-langfuse-ingestion-version": "4"}
+    for item in env.get("OTEL_EXPORTER_OTLP_TRACES_HEADERS", "").split(","):
+        if "=" in item:
+            key, value = item.split("=", 1)
+            headers[key.strip()] = urllib.parse.unquote(value.strip())
+    if not any(key.lower() == "authorization" for key in headers):
+        raise ValueError("Missing configured export authorization")
+    request = urllib.request.Request(endpoint, data=json.dumps(payload).encode(), headers=headers, method="POST")
+    # Don't forward authentication to an unexpected redirect destination.
+    class NoRedirect(urllib.request.HTTPRedirectHandler):
+        def redirect_request(self, *args, **kwargs):
+            return None
+    with urllib.request.build_opener(NoRedirect).open(request, timeout=30) as response:
+        result = json.loads(response.read() or b"{}")
+    if int((result.get("partialSuccess") or {}).get("rejectedSpans", 0)):
+        raise ValueError("Backend rejected generation spans")
+
+
+def main(argv=None):
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("artifacts", nargs="?", type=Path)
+    parser.add_argument("--dry-run", action="store_true", help="Build and count spans without contacting Langfuse")
+    parser.add_argument("--verify-trace-id", help="Export one diagnostic trace with this 32-character hex ID")
+    args = parser.parse_args(argv)
+    if args.artifacts is None and args.verify_trace_id is None:
+        parser.error("artifacts or --verify-trace-id is required")
+    if args.artifacts is not None and args.verify_trace_id is not None:
+        parser.error("artifacts and --verify-trace-id cannot be combined")
+    context = {}
+    for item in os.environ.get("OTEL_RESOURCE_ATTRIBUTES", "").split(","):
+        if "=" in item:
+            key, value = item.split("=", 1)
+            context[key.strip()] = urllib.parse.unquote(value.strip())
+    context["github.repository"] = os.environ.get("GITHUB_REPOSITORY", context.get("github.repository", "fullsend"))
+    for env_key, attr in (("SOURCE_RUN_ID", "github.run_id"), ("SOURCE_RUN_ATTEMPT", "github.run_attempt"),
+                          ("SOURCE_ACTOR", "user.id")):
+        if os.environ.get(env_key):
+            context[attr] = os.environ[env_key]
+    if context.get("github.run_id"):
+        context["session.id"] = context["langfuse.session.id"] = context["github.run_id"]
+    if context.get("user.id"):
+        context["langfuse.user.id"] = context["user.id"]
+    if args.verify_trace_id is not None:
+        try:
+            payload = verification_payload(args.verify_trace_id, context)
+            if not args.dry_run:
+                export(payload, os.environ)
+            print(f"{'Prepared' if args.dry_run else 'Exported'} verification trace {args.verify_trace_id.lower()}.")
+            return 0
+        except Exception as error:
+            status = f" (HTTP {error.code})" if isinstance(error, urllib.error.HTTPError) else ""
+            print(f"Langfuse verification failed: {type(error).__name__}{status}", file=sys.stderr)
+            return 1
+    files = sorted(args.artifacts.rglob("run-telemetry.jsonl"))
+    if not files:
+        print("No Fullsend telemetry artifacts found; nothing to export.")
+    failed = False
+    for path in files:
+        try:
+            payload = build_payload(path, context)
+            count = len(payload["resourceSpans"][0]["scopeSpans"][0]["spans"])
+            if count and not args.dry_run:
+                export(payload, os.environ)
+            print(f"{'Prepared' if args.dry_run else 'Exported'} {count} generation spans.")
+            if not count:
+                print("No agent spans with known models need generation export.")
+        except Exception as error:
+            # Never print URLs, headers, artifact content, or upstream error bodies.
+            status = f" (HTTP {error.code})" if isinstance(error, urllib.error.HTTPError) else ""
+            print(f"Langfuse generation export failed: {type(error).__name__}{status}", file=sys.stderr)
+            failed = True
+    return int(failed)
+
+
+if __name__ == "__main__":
+    sys.exit(main())

.github/workflows/fullsend-langfuse.yaml

@@ -0,0 +1,63 @@
+# Generated by uie-fullsend langfuse-setup. Uses trusted default-branch code.
+name: Fullsend Langfuse
+on:
+  workflow_run:
+    workflows: [fullsend]
+    types: [completed]
+  workflow_dispatch:
+    inputs:
+      trace_id:
+        description: OTLP trace ID for verification (32 hexadecimal characters)
+        required: true
+        type: string
+permissions:
+  actions: read
+  contents: read
+concurrency:
+  group: langfuse-${{ github.event.workflow_run.id || github.run_id }}-${{ github.event.workflow_run.run_attempt || github.run_attempt }}
+  cancel-in-progress: false
+jobs:
+  generation:
+    runs-on: ubuntu-24.04
+    timeout-minutes: 10
+    steps:
+      - uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2
+        with:
+          ref: ${{ github.event.repository.default_branch }}
+          persist-credentials: false
+      - name: Download Fullsend output as data
+        if: github.event_name == 'workflow_run'
+        id: artifacts
+        env:
+          GH_TOKEN: ${{ github.token }}
+          GH_REPO: ${{ github.repository }}
+          RUN_ID: ${{ github.event.workflow_run.id }}
+        run: |
+          artifact_names="$(gh api --paginate "repos/$GH_REPO/actions/runs/$RUN_ID/artifacts?per_page=100" --jq '.artifacts[].name | select(startswith("fullsend-"))')"
+          if [ -z "$artifact_names" ]; then
+            echo "No Fullsend artifacts were published by run $RUN_ID; skipping trace export."
+            echo "found=false" >> "$GITHUB_OUTPUT"
+            exit 0
+          fi
+          if ! gh run download "$RUN_ID" --pattern 'fullsend-*' --dir "$RUNNER_TEMP/langfuse-artifacts"; then
+            echo "::error::Failed to download Fullsend artifacts from run $RUN_ID."
+            exit 1
+          fi
+          echo "found=true" >> "$GITHUB_OUTPUT"
+      - name: Export Langfuse trace
+        if: github.event_name == 'workflow_dispatch' || steps.artifacts.outputs.found == 'true'
+        env:
+          OTEL_EXPORTER_OTLP_TRACES_ENDPOINT: ${{ vars.OTEL_EXPORTER_OTLP_TRACES_ENDPOINT }}
+          OTEL_EXPORTER_OTLP_TRACES_HEADERS: ${{ secrets.OTEL_EXPORTER_OTLP_TRACES_HEADERS }}
+          OTEL_RESOURCE_ATTRIBUTES: ${{ vars.OTEL_RESOURCE_ATTRIBUTES }}
+          SOURCE_RUN_ID: ${{ github.event.workflow_run.id || github.run_id }}
+          SOURCE_RUN_ATTEMPT: ${{ github.event.workflow_run.run_attempt || github.run_attempt }}
+          SOURCE_ACTOR: ${{ github.event.workflow_run.actor.login || github.actor }}
+          VERIFY_TRACE_ID: ${{ inputs.trace_id }}
+          EVENT_NAME: ${{ github.event_name }}
+        run: |
+          if [ "$EVENT_NAME" = workflow_dispatch ]; then
+            python3 .fullsend/scripts/langfuse-post-run.py --verify-trace-id "$VERIFY_TRACE_ID"
+          else
+            python3 .fullsend/scripts/langfuse-post-run.py "$RUNNER_TEMP/langfuse-artifacts"
+          fi