diff --git a/README.md b/README.md index 8ee8de8..95f8dee 100644 --- a/README.md +++ b/README.md @@ -94,6 +94,11 @@ path is in [`pilot/README.md`](pilot/README.md). The model endpoint is supplied at runtime via `PRAGENT_MODEL_BASE_URL`; the committed `opencode.json` carries a placeholder. +Per-review token spend, latency and equivalent cost are shipped to a +self-hosted Langfuse, split into `ollama` and `claude` environments so the two +spend stories stay separate: [`pilot/README-langfuse.md`](pilot/README-langfuse.md). +Emission is a silent no-op unless `LANGFUSE_HOST` and the key pair are set. + ## Extending it The review "factory" is [`.opencode/`](.opencode/README.md) — agent definitions diff --git a/pilot/README-langfuse.md b/pilot/README-langfuse.md new file mode 100644 index 0000000..3602e41 --- /dev/null +++ b/pilot/README-langfuse.md @@ -0,0 +1,122 @@ +# pragent → Langfuse + +Every review the pilot runs ships one **trace** to a self-hosted Langfuse. The +review body already prints a usage table, but that table lives and dies inside +one Gitea PR. Langfuse is where the same numbers become a trend: tokens per +review, latency per model, equivalent cost per repo, and how those move when +the model or the tiering changes. + +## The ollama / claude split + +Both paths route through the same headroom proxy, so the provider prefix does +not distinguish them — `headroom/claude-sonnet-5` is Claude spend, +`headroom/glm-5.2:cloud` is not. The split is keyed off the **bare model name** +and lands on the trace's `environment`: + +| resolved model | environment | +| --------------------------- | ----------- | +| `headroom/claude-sonnet-5` | `claude` | +| `claude-opus-5` | `claude` | +| `headroom/glm-5.2:cloud` | `ollama` | +| `headroom/MiniMax-M2.7` | `ollama` | +| `vllm-qwen38/qwen3.8-27b` | `ollama` | + +Langfuse takes an environment selector on every dashboard, filter and cost +breakdown, so the two spend stories stay separate inside one project — one key +pair to rotate instead of two. Tags carry the finer cut: +`provider:headroom`, `model:`, `engine:opencode`, `repo:`, +`lens:` per fan-out lens. + +To split into two *projects* later, point `LANGFUSE_PUBLIC_KEY` / +`LANGFUSE_SECRET_KEY` at the second project on whichever deployment runs the +Claude path. Nothing in the code needs to change. + +## What a trace carries + +- **trace** `pr-review` — `sessionId` = `owner/repo#index`, so every push to one + PR groups together. Input is the PR identity; output is the summary + finding + count; metadata carries steps, duration, severity counts and the provider's + own reported cost. +- **generation** `opencode-review` — `model`, `usageDetails`, `costDetails`. + +`usageDetails.input` is the **uncached** input. opencode reports `cache_read` +*inside* `input`, and Langfuse sums the keys it is given, so passing both +verbatim would bill the resent prefix twice. + +### How cost is priced + +Langfuse has no price table of its own here — we compute the number and ship it +as `costDetails.total`, so what Langfuse charts is exactly what +`cost_model.PRICES` says. + +A model that genuinely bills (`claude-*`, `gpt-*`, `gemini-*`, `grok-*`) is +priced **as itself**: basis `actual`. + +A model that costs nothing through the headroom proxy is priced against a +**comparison target** instead: basis `equivalent:`. That covers the +models absent from `PRICES` (`MiniMax-M2.7` — which is what the webhook +actually runs — and `glm-5.2:cloud`) as well as entries priced at all zeros +(the self-hosted vLLM `qwen3.8-27b`). Without this the dashboard would be a +flat $0.00 line, since the pilot's own path is free. + +The target follows the same precedence as the review body, so the PR and the +dashboard never disagree: + + .pr-review.json:cost_target > PRAGENT_PRICE_TARGET > claude-sonnet-5 + +An equivalent cost is a hypothetical, not money spent, so every trace is tagged +`cost:actual` or `cost:equivalent:` and the generation metadata carries +`cost_basis`. Filter on it before reading any cost chart as spend. + +If the comparison target itself is unknown, the trace ships usage with **no** +cost block — better no number than a wrong one. + +Anthropic prices in `cost_model.PRICES` were fetched 2026-08-18; re-check them +before quoting anything externally. + +## Configuration + +| env | meaning | +| --------------------- | --------------------------------------------------------- | +| `LANGFUSE_HOST` | `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 | + +Unset host or either key ⇒ emission is a silent no-op. That is the default, so +a checkout without Langfuse behaves exactly as before. + +## Fail-open + +`langfuse_trace` is stdlib-only (`urllib`) and every entry point swallows its +own exceptions; `_emit_langfuse` in `ai_review.py` wraps even the import. A +Langfuse outage cannot fail, delay past `LANGFUSE_TIMEOUT`, or alter a review. + +Both token-spending exit paths emit — the normal post **and** the salvage path +where the agent produced unparseable output. That run cost the same as a clean +one, and is precisely the failure worth trending. + +## Deployment + +Cluster side lives outside this repo: `~/k8s/langfuse.yaml` (ClickHouse + +web + worker, reusing the gitea postgres, gitea valkey and minio), +`~/k8s/oauth2-proxy-langfuse.yaml` (the Logto gate), and +`~/k8s/langfuse-setup.sh`, which provisions the database, the bucket, the +secrets, and wires `pragent-webhook` with the three env vars above. + +The UI is at **https://langfuse.marcospaulo.dev.br**: + + browser -> Caddy (VPS, TLS, DNS-01) -> tailscale + -> 100.74.17.70:30361 -> oauth2-proxy (Logto, email allowlist) + -> langfuse-web (ClusterIP) + +Logto sits at *both* layers off one app (`langfuse`, two redirect URIs): the +proxy gates the domain, and Langfuse's own NextAuth uses the same Logto as a +custom OIDC provider, so the inner login is a silent redirect rather than a +second password. + +pragent does **not** go through any of that. It posts to +`langfuse-web.langfuse.svc.cluster.local:3000` from inside the cluster, on +API-key auth — putting ingestion behind an interactive SSO gate would break it +on the first review. diff --git a/pilot/ai_review.py b/pilot/ai_review.py index 9e82231..c0d6289 100644 --- a/pilot/ai_review.py +++ b/pilot/ai_review.py @@ -2068,6 +2068,49 @@ def _need(name: str) -> str: return v +def _emit_langfuse( + *, + repo: str, + index: str, + sha: str, + title: str, + model: str, + usage: dict | None, + findings: list[dict], + summary: str, + engine: str, + config: dict | None = None, +) -> None: + """Ship this review's usage to Langfuse, if one is configured. + + Called on both exit paths that spent tokens — the normal post and the + salvage path — because an unparseable run costs the same as a clean one and + is exactly the kind of thing worth trending. + + Local import + blanket except: `langfuse_trace` is stdlib-only but optional, + and telemetry is never allowed to fail a review (see the fail-open contract + in `review_pr`). The trace's `environment` is `claude` or `ollama`, so the + two spend stories stay separated in every Langfuse view. + """ + try: + import langfuse_trace + + # Same comparison model the review body prices against, so the number + # in Langfuse and the number in the PR agree. Free/unknown models + # (MiniMax, glm, self-hosted qwen) are priced against it; a paid model + # is priced as itself. + price_target, _err = _resolve_price_target(config) + + langfuse_trace.emit_review_trace( + repo=repo, index=index, sha=sha, title=title, model=model, + usage=usage, findings=findings, summary=summary or "", + engine=engine, lenses=(usage or {}).get("lenses"), + price_target=price_target, + ) + except Exception as e: + print(f"pragent: langfuse emit skipped: {e}", file=sys.stderr) + + def review_pr( api: str, repo: str, @@ -2208,6 +2251,11 @@ def review_pr( salvaged or "AI review produced no parseable output.", display_model, sha, usage_section=usage_section, static_message=(config or {}).get("static_message", ""))) + _emit_langfuse( + repo=repo, index=index, sha=sha, title=title, + model=display_model, usage=usage, findings=[], + summary=salvaged, engine=engine, config=config, + ) return True else: user_prompt = build_user_prompt(title, body + compression_note, diff, config, prior, additional_context) @@ -2291,6 +2339,11 @@ def review_pr( ) post_inline_review(api, repo, index, token, summary_body, anchored) + _emit_langfuse( + repo=repo, index=index, sha=sha, title=title, + model=display_model, usage=usage, findings=findings, + summary=review_summary, engine=engine, config=config, + ) print( f"pragent: reviewed {repo}#{index} sha={sha[:8]} " f"engine={engine} findings={len(findings)} inline={len(anchored)}", diff --git a/pilot/langfuse_trace.py b/pilot/langfuse_trace.py new file mode 100644 index 0000000..484fc7e --- /dev/null +++ b/pilot/langfuse_trace.py @@ -0,0 +1,364 @@ +#!/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, +) -> list[dict]: + """The ingestion batch for one review: a trace plus one generation. + + Split out from `emit_review_trace` so the shape is testable without a + Langfuse to POST to. + """ + 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, + } + ) + + return events + + +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: + return resp.status + + +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 diff --git a/tests/pilot/test_langfuse_trace.py b/tests/pilot/test_langfuse_trace.py new file mode 100644 index 0000000..1cd4634 --- /dev/null +++ b/tests/pilot/test_langfuse_trace.py @@ -0,0 +1,245 @@ +"""Unit tests for Langfuse trace emission. No network. + +`_post` is monkeypatched everywhere a POST would happen; a test that reaches +the real network is a bug in the test, not a slow test. +""" +import json +import os +import sys + +HERE = os.path.dirname(os.path.abspath(__file__)) +ROOT = os.path.abspath(os.path.join(HERE, "..", "..")) +sys.path.insert(0, os.path.join(ROOT, "pilot")) + +import langfuse_trace as lt # noqa: E402 + + +USAGE = { + "input": 2_000_000, + "output": 17_000, + "reasoning": 500, + "cache_read": 400_000, + "cache_write": 50_000, + "total": 2_017_000, + "cost": 0.0, + "steps": 28, + "duration_s": 348.3, +} + +BASE = dict( + repo="techspark/pragent", + index="42", + sha="2613b3e1122334455", + title="Harden the review path", + usage=USAGE, + findings=[ + {"severity": "critical", "path": "a.py"}, + {"severity": "minor", "path": "b.py"}, + {"severity": "minor", "path": "c.py"}, + ], + summary="Three findings.", +) + + +# --------------------------------------------------------------------------- +# model -> environment split (the whole point of the integration) +# --------------------------------------------------------------------------- + + +def test_claude_models_land_in_the_claude_environment(): + assert lt.resolve_environment("headroom/claude-sonnet-5") == "claude" + assert lt.resolve_environment("claude-opus-5") == "claude" + + +def test_everything_else_lands_in_the_ollama_environment(): + for m in ( + "headroom/glm-5.2:cloud", + "headroom/MiniMax-M2.7", + "vllm-qwen38/qwen3.8-27b", + "gpt-5", + ): + assert lt.resolve_environment(m) == "ollama", m + + +def test_provider_and_bare_model_are_split_on_the_first_slash_only(): + assert lt.provider_of("vllm-qwen38/qwen3.8-27b") == "vllm-qwen38" + assert lt.strip_provider("headroom/glm-5.2:cloud") == "glm-5.2:cloud" + # A bare name has no provider prefix; default to the pilot's proxy. + assert lt.provider_of("glm-5.2:cloud") == "headroom" + assert lt.strip_provider("glm-5.2:cloud") == "glm-5.2:cloud" + + +# --------------------------------------------------------------------------- +# usage accounting +# --------------------------------------------------------------------------- + + +def test_cache_reads_are_subtracted_from_input_not_added(): + # Langfuse sums usageDetails keys; opencode reports cache_read *inside* + # input, so reporting both raw would bill the prefix twice. + d = lt._usage_details(USAGE) + assert d["input"] == 2_000_000 - 400_000 + assert d["cache_read_input_tokens"] == 400_000 + assert d["cache_write_input_tokens"] == 50_000 + assert d["output"] == 17_000 + assert d["reasoning"] == 500 + + +def test_zero_cache_fields_are_omitted_rather_than_sent_as_zero(): + d = lt._usage_details({"input": 100, "output": 10}) + assert d == {"input": 100, "output": 10} + + +def test_a_paid_model_is_priced_as_itself(): + costs, basis = lt._cost_details(USAGE, "headroom/claude-sonnet-5") + assert costs["total"] > 0 + assert basis == "actual" + + +def test_minimax_is_priced_against_the_comparison_target_not_zero(): + # MiniMax-M2.7 is the model the webhook actually runs and it is absent from + # PRICES; charting it at $0 would make the whole dashboard a flat line. + costs, basis = lt._cost_details(USAGE, "headroom/MiniMax-M2.7") + assert costs["total"] > 0 + assert basis == "equivalent:claude-sonnet-5" + + +def test_glm_is_priced_against_the_comparison_target(): + costs, basis = lt._cost_details(USAGE, "headroom/glm-5.2:cloud") + assert costs["total"] > 0 + assert basis.startswith("equivalent:") + + +def test_an_all_zero_price_entry_counts_as_free_not_as_priced(): + # The self-hosted vLLM qwen IS in PRICES, at 0.00 across the board. + costs, basis = lt._cost_details(USAGE, "vllm-qwen38/qwen3.8-27b") + assert costs["total"] > 0 + assert basis.startswith("equivalent:") + + +def test_explicit_price_target_wins_over_the_default(): + costs, basis = lt._cost_details(USAGE, "headroom/MiniMax-M2.7", "claude-opus-5") + assert basis == "equivalent:claude-opus-5" + sonnet, _ = lt._cost_details(USAGE, "headroom/MiniMax-M2.7", "claude-sonnet-5") + assert costs["total"] > sonnet["total"] + + +def test_env_overrides_the_default_target(monkeypatch): + monkeypatch.setenv("PRAGENT_PRICE_TARGET", "claude-haiku-4-5") + assert lt.resolve_price_target() == "claude-haiku-4-5" + # An explicit argument still beats the env. + assert lt.resolve_price_target("gpt-5") == "gpt-5" + + +def test_unknown_comparison_target_yields_no_cost_block_rather_than_a_wrong_one(): + costs, basis = lt._cost_details(USAGE, "headroom/MiniMax-M2.7", "not-a-real-model") + assert costs == {} + assert basis == "" + + +# --------------------------------------------------------------------------- +# batch shape +# --------------------------------------------------------------------------- + + +def test_batch_has_a_trace_and_a_generation_linked_by_trace_id(): + batch = lt.build_batch(model="headroom/claude-sonnet-5", **BASE) + types = [e["type"] for e in batch] + assert types == ["trace-create", "generation-create"] + trace, gen = batch + assert gen["body"]["traceId"] == trace["body"]["id"] + assert trace["body"]["environment"] == gen["body"]["environment"] == "claude" + + +def test_batch_without_usage_is_trace_only(): + batch = lt.build_batch(model="headroom/glm-5.2:cloud", **{**BASE, "usage": None}) + assert [e["type"] for e in batch] == ["trace-create"] + + +def test_trace_carries_repo_pr_session_and_severity_counts(): + batch = lt.build_batch(model="headroom/glm-5.2:cloud", **BASE) + body = batch[0]["body"] + assert body["sessionId"] == "techspark/pragent#42" + assert body["metadata"]["severities"] == {"critical": 1, "minor": 2} + assert body["metadata"]["findings"] == 3 + assert "provider:headroom" in body["tags"] + assert "model:glm-5.2:cloud" in body["tags"] + + +def test_lens_names_become_tags(): + batch = lt.build_batch( + model="headroom/glm-5.2:cloud", lenses=["security", "tests"], **BASE + ) + assert "lens:security" in batch[0]["body"]["tags"] + assert "lens:tests" in batch[0]["body"]["tags"] + + +def test_cost_basis_is_tagged_so_equivalent_is_never_read_as_spend(): + batch = lt.build_batch(model="headroom/MiniMax-M2.7", **BASE) + trace = batch[0]["body"] + assert "cost:equivalent:claude-sonnet-5" in trace["tags"] + assert trace["metadata"]["cost_basis"] == "equivalent:claude-sonnet-5" + + paid = lt.build_batch(model="headroom/claude-sonnet-5", **BASE) + assert "cost:actual" in paid[0]["body"]["tags"] + + +def test_minimax_generation_carries_a_nonzero_cost(): + batch = lt.build_batch(model="headroom/MiniMax-M2.7", **BASE) + assert batch[1]["body"]["costDetails"]["total"] > 0 + + +def test_batch_is_json_serializable(): + batch = lt.build_batch(model="headroom/claude-sonnet-5", **BASE) + json.dumps({"batch": batch}) + + +# --------------------------------------------------------------------------- +# emit_review_trace — config gate and fail-open +# --------------------------------------------------------------------------- + + +def _configure(monkeypatch): + monkeypatch.setenv("LANGFUSE_HOST", "http://langfuse.test:3000/") + monkeypatch.setenv("LANGFUSE_PUBLIC_KEY", "pk-lf-test") + monkeypatch.setenv("LANGFUSE_SECRET_KEY", "sk-lf-test") + + +def test_no_config_means_no_post_and_no_error(monkeypatch): + for k in ("LANGFUSE_HOST", "LANGFUSE_PUBLIC_KEY", "LANGFUSE_SECRET_KEY"): + monkeypatch.delenv(k, raising=False) + calls = [] + monkeypatch.setattr(lt, "_post", lambda *a, **k: calls.append(a) or 200) + assert lt.emit_review_trace(model="headroom/glm-5.2:cloud", **BASE) is False + assert calls == [] + + +def test_configured_emit_posts_to_the_ingestion_endpoint(monkeypatch): + _configure(monkeypatch) + seen = {} + + def fake_post(host, pk, sk, batch, timeout): + seen.update(host=host, pk=pk, sk=sk, batch=batch, timeout=timeout) + return 207 + + monkeypatch.setattr(lt, "_post", fake_post) + assert lt.emit_review_trace(model="headroom/claude-sonnet-5", **BASE) is True + # Trailing slash stripped so the path is not doubled. + assert seen["host"] == "http://langfuse.test:3000" + assert len(seen["batch"]) == 2 + + +def test_transport_failure_is_swallowed(monkeypatch): + _configure(monkeypatch) + + def boom(*a, **k): + raise OSError("connection refused") + + monkeypatch.setattr(lt, "_post", boom) + assert lt.emit_review_trace(model="headroom/glm-5.2:cloud", **BASE) is False + + +def test_non_success_status_reports_failure_without_raising(monkeypatch): + _configure(monkeypatch) + monkeypatch.setattr(lt, "_post", lambda *a, **k: 401) + assert lt.emit_review_trace(model="headroom/glm-5.2:cloud", **BASE) is False