diff --git a/.fullsend/scripts/langfuse-post-run.py b/.fullsend/scripts/langfuse-post-run.py new file mode 100644 index 000000000..99bc558eb --- /dev/null +++ b/.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()) diff --git a/.github/workflows/fullsend-langfuse.yaml b/.github/workflows/fullsend-langfuse.yaml new file mode 100644 index 000000000..60f5cac2f --- /dev/null +++ b/.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