#!/usr/bin/env python3 """pragent pilot — Langfuse trace emission. Ships one trace per PR review to a self-hosted Langfuse (v3) so the reviewer's token spend, latency and per-model behaviour are queryable outside the review body. The review body already renders a usage table; that table is per-PR and disappears into Gitea. This is the same numbers, aggregated. Why hand-rolled instead of the `langfuse` SDK: the pilot image is stdlib-only (see pilot/Dockerfile — no requirements.txt anywhere in the repo), and the ingestion API is a single authenticated POST of a JSON batch. Pulling an SDK plus its otel dependency tree into a fail-open telemetry side-path is a bad trade. Provider split -------------- `environment` on every trace is either `ollama` or `claude`, derived from the resolved display model (`resolve_environment`). That is what keeps the two spend stories separate in Langfuse: every dashboard, filter and cost breakdown takes an environment selector, so "what did the local/self-hosted path cost" and "what did the Claude path cost" are two views of one project rather than two projects with two key pairs to rotate. Tags carry the finer split (`provider:headroom`, `model:...`, `engine:opencode`). Cost ---- The pilot's own path bills $0 (headroom proxy, no per-token charge), so the `cost` reported to Langfuse is the *equivalent* cost from `cost_model` — what the same tokens would bill on the comparison model. That is the number worth trending; a chart of $0.00 is not. A model is "free" when `cost_model.PRICES` has no entry for it (MiniMax-M2.7, glm-5.2:cloud) or when its entry is all zeros (the self-hosted vLLM qwen). In both cases the reported cost is priced against the comparison target instead — same precedence the review body uses: `.pr-review.json:cost_target` > `PRAGENT_PRICE_TARGET` > `claude-sonnet-5`. A paid model is priced as itself. Because a hypothetical and a real charge must never be read as the same number, every trace is tagged `cost:actual` or `cost:equivalent:`, and the generation's metadata carries `cost_basis`. Fail-open: every entry point swallows its own exceptions. Telemetry must never cost a review. Env: LANGFUSE_HOST e.g. http://langfuse-web.langfuse.svc.cluster.local:3000 LANGFUSE_PUBLIC_KEY pk-lf-... LANGFUSE_SECRET_KEY sk-lf-... LANGFUSE_TIMEOUT seconds, default 5 LANGFUSE_DEBUG 1 to log ingestion failures to stderr Disabled (silently) when host or either key is unset. """ from __future__ import annotations import base64 import json import os import sys import time import urllib.error import urllib.request import uuid from datetime import datetime, timezone INGESTION_PATH = "/api/public/ingestion" # Model-key prefixes that mean "this review ran against Anthropic-shaped # billing". Everything else (glm, MiniMax, qwen, local vLLM) is the ollama / # self-hosted side of the split. _CLAUDE_PREFIXES = ("claude-", "anthropic/") def _now_iso() -> str: return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") def _enabled() -> tuple[str, str, str] | None: host = (os.environ.get("LANGFUSE_HOST") or "").strip().rstrip("/") pk = (os.environ.get("LANGFUSE_PUBLIC_KEY") or "").strip() sk = (os.environ.get("LANGFUSE_SECRET_KEY") or "").strip() if not host or not pk or not sk: return None return host, pk, sk def _debug(msg: str) -> None: if os.environ.get("LANGFUSE_DEBUG"): print(f"pragent/langfuse: {msg}", file=sys.stderr, flush=True) def strip_provider(model: str) -> str: """`headroom/claude-sonnet-5` -> `claude-sonnet-5`. Bare names pass through.""" return model.split("/", 1)[1] if "/" in model else model def provider_of(model: str) -> str: """The opencode provider block a display model routes through.""" return model.split("/", 1)[0] if "/" in model else "headroom" def resolve_environment(model: str) -> str: """Which spend story this review belongs to: `claude` or `ollama`. Keyed off the bare model name, not the provider, because both paths route through the same `headroom` proxy — `headroom/claude-sonnet-5` is Claude spend, `headroom/glm-5.2:cloud` is not. """ bare = strip_provider(model).lower() return "claude" if bare.startswith(_CLAUDE_PREFIXES) else "ollama" def _usage_details(usage: dict) -> dict: """opencode's usage dict -> Langfuse `usageDetails`. Langfuse sums every key except the ones it knows are derived, so `input` here is the *uncached* portion: reporting both `input` (which opencode reports as the full input, cache included) and `cache_read_input_tokens` would double-count. """ inp = int(usage.get("input") or 0) cache_read = int(usage.get("cache_read") or 0) cache_write = int(usage.get("cache_write") or 0) details = { "input": max(0, inp - cache_read), "output": int(usage.get("output") or 0), } if cache_read: details["cache_read_input_tokens"] = cache_read if cache_write: details["cache_write_input_tokens"] = cache_write reasoning = int(usage.get("reasoning") or 0) if reasoning: details["reasoning"] = reasoning return details DEFAULT_PRICE_TARGET = "claude-sonnet-5" def resolve_price_target(price_target: str | None = None) -> str: """The model to price free/unknown runs against. Mirrors `ai_review._resolve_price_target`: an explicit target (which the caller reads from `.pr-review.json:cost_target`) wins, then `PRAGENT_PRICE_TARGET`, then Sonnet. """ if price_target and price_target.strip(): return price_target.strip() env = os.environ.get("PRAGENT_PRICE_TARGET", "").strip() return env or DEFAULT_PRICE_TARGET def _is_free(price) -> bool: """A price entry that charges nothing — self-hosted or proxied at no cost.""" return price.input == 0 and price.output == 0 def _cost_details(usage: dict, model: str, price_target: str | None = None) -> tuple[dict, str]: """USD for this usage plus the basis it was computed on. Returns `({"total": …}, basis)` where basis is `actual` for a model that genuinely bills, or `equivalent:` for one that does not. `({}, "")` when nothing can be priced at all — better no number than a wrong one. Local import + broad except: `cost_model` is only present on the opencode path, and an unknown model key must not break telemetry. """ try: from cost_model import PRICES, Usage, cost bare = strip_provider(model) price = PRICES.get(bare) basis = "actual" if price is None or _is_free(price): # MiniMax / glm / self-hosted qwen: $0 through the proxy, so the # useful number is what these tokens would have billed elsewhere. target = resolve_price_target(price_target) price = PRICES.get(target) if price is None: _debug(f"comparison target {target!r} not in PRICES") return {}, "" basis = f"equivalent:{target}" u = Usage( uncached_input=max(0, int(usage.get("input") or 0) - int(usage.get("cache_read") or 0)), cached_input=int(usage.get("cache_read") or 0), cache_writes=int(usage.get("cache_write") or 0), output=int(usage.get("output") or 0), ) return {"total": round(cost(u, price), 6)}, basis except Exception as e: # pragma: no cover - defensive _debug(f"cost lookup failed for {model!r}: {e}") return {}, "" def _severity_counts(findings: list[dict] | None) -> dict: counts: dict[str, int] = {} for f in findings or []: sev = str(f.get("severity") or "unknown").lower() counts[sev] = counts.get(sev, 0) + 1 return counts def build_batch( *, repo: str, index: str, sha: str, title: str, model: str, usage: dict | None, findings: list[dict] | None = None, summary: str = "", engine: str = "opencode", tier: str = "", lenses: list[str] | None = None, trace_id: str | None = None, release: str = "", price_target: str | None = None, dropped_count: float | None = None, ) -> list[dict]: """The ingestion batch for one review: a trace, a generation, and scores. Split out from `emit_review_trace` so the shape is testable without a Langfuse to POST to. `dropped_count` is how many findings the parser rejected for an unusable `path`/`line`, measured where the model output was parsed. Passing it turns on the `dropped_findings` score; leaving it `None` omits that score rather than reporting a zero the caller never measured. """ usage = usage or {} tid = trace_id or str(uuid.uuid4()) ts = _now_iso() env = resolve_environment(model) duration = float(usage.get("duration_s") or 0.0) started = datetime.fromtimestamp( time.time() - duration, tz=timezone.utc ).isoformat().replace("+00:00", "Z") tags = [ f"provider:{provider_of(model)}", f"model:{strip_provider(model)}", f"engine:{engine}", f"repo:{repo}", ] if tier: tags.append(f"tier:{tier}") for lens in lenses or []: tags.append(f"lens:{lens}") costs, cost_basis = _cost_details(usage, model, price_target) if usage else ({}, "") if cost_basis: # Filterable in Langfuse, so an equivalent-cost chart can never be # mistaken for money actually spent. tags.append(f"cost:{cost_basis}") metadata = { "repo": repo, "pr": index, "sha": sha, "engine": engine, "steps": usage.get("steps"), "duration_s": duration or None, "findings": len(findings or []), "severities": _severity_counts(findings), "provider_cost_usd": usage.get("cost"), "cost_basis": cost_basis or None, } if lenses: metadata["lenses"] = lenses if tier: metadata["tier"] = tier metadata = {k: v for k, v in metadata.items() if v not in (None, {}, [])} trace_body = { "id": tid, "name": "pr-review", "timestamp": ts, "environment": env, "sessionId": f"{repo}#{index}", "input": {"repo": repo, "pr": index, "sha": sha, "title": title}, "output": {"summary": summary[:2000], "findings": len(findings or [])}, "metadata": metadata, "tags": tags, } if release: trace_body["release"] = release events = [ { "id": str(uuid.uuid4()), "type": "trace-create", "timestamp": ts, "body": trace_body, } ] if usage: gen_body = { "id": str(uuid.uuid4()), "traceId": tid, "type": "GENERATION", "name": f"{engine}-review", "environment": env, "startTime": started, "endTime": ts, "model": strip_provider(model), "usageDetails": _usage_details(usage), "metadata": metadata, "level": "DEFAULT", } if costs: gen_body["costDetails"] = costs events.append( { "id": str(uuid.uuid4()), "type": "generation-create", "timestamp": ts, "body": gen_body, } ) events.extend( _score_events( trace_id=tid, findings=findings, environment=env, cost_usd=costs.get("total"), dropped_count=dropped_count, timestamp=ts, cost_basis=cost_basis, ) ) return events def _score_events(*, cost_basis: str, **kwargs) -> list[dict]: """Deterministic scores for this review, or [] if the scorer is missing. Local import + blanket except for the same reason the rest of this module swallows: `eval_scores` is optional, and a scoring bug must not cost the trace it was supposed to annotate. """ try: import eval_scores # The cost score is only meaningful next to its basis — a $/finding # figure computed from an equivalent price is not money that was spent. comment = f"cost basis: {cost_basis}" if cost_basis else "" return eval_scores.build_scores(comment=comment, **kwargs) except Exception as e: # pragma: no cover - defensive _debug(f"scoring failed: {e}") return [] def _post(host: str, pk: str, sk: str, batch: list[dict], timeout: float) -> int: payload = json.dumps({"batch": batch}).encode("utf-8") auth = base64.b64encode(f"{pk}:{sk}".encode("utf-8")).decode("ascii") req = urllib.request.Request( host + INGESTION_PATH, data=payload, headers={ "Content-Type": "application/json", "Authorization": f"Basic {auth}", "User-Agent": "pragent-pilot/1.0", }, method="POST", ) with urllib.request.urlopen(req, timeout=timeout) as resp: _warn_on_rejected_events(resp.read()) return resp.status def _warn_on_rejected_events(raw: bytes) -> None: """Surface per-event rejections hiding inside a 207. The ingestion endpoint answers 207 Multi-Status when *some* events failed, so a caller that only checks the status code reads a batch where every single event was rejected as a success. That failure mode is invisible exactly when it matters — the traces simply never appear. """ try: body = json.loads(raw or b"{}") errors = body.get("errors") or [] if errors: first = errors[0] _debug( f"{len(errors)} event(s) rejected by ingestion; " f"first: status={first.get('status')} {first.get('error')}" ) except Exception: # pragma: no cover - never let logging break emission pass def emit_review_trace(**kwargs) -> bool: """Ship one review's trace. Returns True if Langfuse accepted it. No-op (False) when Langfuse is unconfigured. Never raises — a telemetry outage must not turn into a failed review. """ conf = _enabled() if conf is None: return False host, pk, sk = conf try: timeout = float(os.environ.get("LANGFUSE_TIMEOUT", "5")) except ValueError: timeout = 5.0 try: batch = build_batch(**kwargs) status = _post(host, pk, sk, batch, timeout) if status not in (200, 201, 207): _debug(f"ingestion returned HTTP {status}") return False return True except urllib.error.HTTPError as e: _debug(f"ingestion HTTP {e.code}: {e.read()[:300]!r}") except Exception as e: _debug(f"ingestion failed: {e}") return False