Merge pull request 'feat(pilot): emit per-review Langfuse traces' (#11) from feat/langfuse-tracing into main

Ship token spend, latency and equivalent cost for every review to the
self-hosted Langfuse so per-model behaviour is queryable as a trend rather
than one PR comment at a time.

langfuse_trace.py is stdlib-only and emits via the public ingestion API.
Traces split into `ollama` and `claude` environments keyed off the bare model
name, not the provider: both paths go through the same headroom proxy, so the
provider prefix says nothing about which spend story a review belongs to. The
pilot's own path bills $0, so the reported cost is the equivalent price from
cost_model.PRICES.

ai_review.py calls _emit_langfuse on both token-spending exit paths (the
normal post and the salvage path). Import and emission are wrapped in a
blanket except: with no LANGFUSE_HOST or key pair the whole thing is a silent
no-op, and a telemetry failure must never fail a review.

These files were previously deployed only by way of the image build's
`COPY . /app`, so a clean checkout would have silently dropped tracing.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

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

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:<bare>`, `engine:opencode`, `repo:<owner/name>`,
+`lens:<id>` 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:<target>`. 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:<target>` 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.

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)}",

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:<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,
+) -> 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

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