pragent/pilot/langfuse_trace.py

#!/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:<target>`, 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:<target>` 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