refactor: organize pilot packages
Group review, feedback, evaluation, observability, and entrypoint code into packages. Keep thin top-level compatibility shims for existing scripts and imports, and mirror the structure in the tests.
This commit is contained in:
Claude
@@ -0,0 +1 @@
|
||||
"""Cost modeling and Langfuse telemetry."""
|
||||
@@ -0,0 +1,434 @@
|
||||
#!/usr/bin/env python3
|
||||
"""pragent pilot — per-review cost model.
|
||||
|
||||
Answers "what would this cost on a paid API?" for the pilot's agent loop. The
|
||||
pilot currently runs on `glm-5.2:cloud` through the on-network headroom proxy at
|
||||
no per-token charge, so every review's measured usage is *free but real*: it
|
||||
tells us exactly what the same work would bill on Claude or GPT.
|
||||
|
||||
The model is deliberately explicit rather than a single fudge factor, because
|
||||
the dominant cost in an agent loop is not the diff — it is **resending the
|
||||
conversation on every step**. A 12-step review re-reads its own prefix 12 times.
|
||||
Prompt caching is what makes that affordable, and whether caching is on changes
|
||||
the answer by ~3x, so it's a parameter, not an assumption.
|
||||
|
||||
Token accounting per review:
|
||||
|
||||
step 1 input = prefix + brief
|
||||
step k input = prefix + brief + (tool results accumulated through k-1)
|
||||
total input = sum over steps
|
||||
cached = the prefix + brief part of steps 2..n (stable, byte-identical)
|
||||
uncached = step 1 in full + the growing tool-result tail
|
||||
|
||||
`prefix` = system + tool schemas + agent definition + the skills this tier loads.
|
||||
Those sizes are MEASURED from the files in this repo (see `measure_factory`),
|
||||
not guessed. Diff size, file reads, and step count are per-tier assumptions from
|
||||
the `attention-tiering` skill's budgets — override them on the CLI to fit your
|
||||
own repos.
|
||||
|
||||
Prices are per million tokens, from the providers' published pricing pages
|
||||
(fetched 2026-08-18 — re-check before quoting):
|
||||
https://platform.claude.com/docs/en/about-claude/pricing
|
||||
https://developers.openai.com/api/docs/pricing
|
||||
|
||||
Usage:
|
||||
python3 pilot/cost_model.py # all tiers, all models
|
||||
python3 pilot/cost_model.py --prs-per-month 350
|
||||
python3 pilot/cost_model.py --mix 5,35,55,5 # trivial,lite,full,oversized %
|
||||
python3 pilot/cost_model.py --no-cache # what caching is worth
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import os
|
||||
from dataclasses import dataclass, field
|
||||
|
||||
CHARS_PER_TOKEN = 4 # English prose/code rule of thumb; ±15% is normal
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Prices — USD per million tokens
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Price:
|
||||
"""Per-MTok prices. `cache_write` and `cache_read` are absolute rates, not
|
||||
multipliers, so providers with different cache economics stay comparable.
|
||||
|
||||
`provider` is the opencode provider name (`headroom`, `vllm-qwen38`, ...). It
|
||||
doubles as the dispatch key for `.pr-review.json:model` overrides — when
|
||||
a per-repo override is set, `_resolve_display_model` returns
|
||||
`f"{provider}/{key}"` so the opencode subprocess routes correctly.
|
||||
Default `headroom` preserved for the existing roster."""
|
||||
|
||||
name: str
|
||||
input: float
|
||||
output: float
|
||||
cache_write: float
|
||||
cache_read: float
|
||||
provider: str = "headroom"
|
||||
|
||||
@property
|
||||
def batch_input(self) -> float:
|
||||
return self.input / 2
|
||||
|
||||
@property
|
||||
def batch_output(self) -> float:
|
||||
return self.output / 2
|
||||
|
||||
|
||||
# Anthropic: cache write = 1.25x input (5-minute TTL), cache read = 0.1x input.
|
||||
# OpenAI: cached input is a published rate (0.1x input); there is no separate
|
||||
# cache-write charge — writes are billed as ordinary input.
|
||||
PRICES: dict[str, Price] = {
|
||||
"claude-opus-5": Price("Claude Opus 5", 5.00, 25.00, 6.25, 0.50),
|
||||
"claude-sonnet-5": Price("Claude Sonnet 5", 2.00, 10.00, 2.50, 0.20),
|
||||
"claude-haiku-4-5": Price("Claude Haiku 4.5", 1.00, 5.00, 1.25, 0.10),
|
||||
"gpt-5.6-sol": Price("GPT-5.6 Sol", 5.00, 30.00, 5.00, 0.50),
|
||||
"gpt-5.6-terra": Price("GPT-5.6 Terra", 2.00, 12.00, 2.00, 0.20),
|
||||
"gpt-5.6-luna": Price("GPT-5.6 Luna", 0.20, 1.20, 0.20, 0.02),
|
||||
# OpenAI — cached_input 0.1x, no separate cache_write
|
||||
"gpt-5": Price("GPT-5", 1.25, 10.00, 1.25, 0.125),
|
||||
"gpt-5-mini": Price("GPT-5 mini", 0.25, 2.00, 0.25, 0.025),
|
||||
# Google Gemini — cache_write = input
|
||||
"gemini-2.5-pro": Price("Gemini 2.5 Pro", 1.875, 12.50, 1.875, 0.1875),
|
||||
"gemini-2.5-flash": Price("Gemini 2.5 Flash", 0.30, 2.50, 0.30, 0.03),
|
||||
# xAI Grok — cache_write = input
|
||||
"grok-4.5": Price("Grok 4.5", 2.00, 6.00, 2.00, 0.30),
|
||||
"grok-4.3": Price("Grok 4.3", 1.25, 2.50, 1.25, 0.20),
|
||||
# Self-hosted — AI workstation RTX 3090, vLLM + DFlash2 spec-decode, no
|
||||
# per-token charge. provider="vllm-qwen38" so the opencode subprocess
|
||||
# routes via the matching provider block in opencode.json
|
||||
# (baseURL=http://192.168.1.79:18020/v1). Equivalent-cost column reads $0
|
||||
# — the cost-comparison signal is that the same work would bill $X on a
|
||||
# paid model.
|
||||
"qwen3.8-27b": Price("Qwen3.8-27B (vLLM, MTP, 150k ctx)", 0.0, 0.0, 0.0, 0.0, provider="vllm-qwen38"),
|
||||
}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Factory footprint — measured from this repo
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||||
|
||||
# Skills the primary always loads, and the conditional ones per tier. Mirrors
|
||||
# the load table in .opencode/agents/pragent.md.
|
||||
ALWAYS_SKILLS = ("review-methodology", "findings-schema", "attention-tiering")
|
||||
TIER_SKILLS: dict[str, tuple[str, ...]] = {
|
||||
"trivial": (),
|
||||
"lite": ("comment-craft",),
|
||||
"full": ("linter-playbook", "security-lens", "comment-craft"),
|
||||
"oversized": ("linter-playbook", "security-lens", "comment-craft", "malicious-change"),
|
||||
}
|
||||
|
||||
# opencode's own system prompt + the JSON tool schemas it sends (read, grep,
|
||||
# glob, bash, webfetch, skill, task, …). Not in this repo, so this is the one
|
||||
# component that is an estimate rather than a measurement.
|
||||
HARNESS_TOKENS = 3500
|
||||
|
||||
|
||||
def _tok(path: str) -> int:
|
||||
try:
|
||||
with open(path, "rb") as f:
|
||||
return len(f.read()) // CHARS_PER_TOKEN
|
||||
except OSError:
|
||||
return 0
|
||||
|
||||
|
||||
def measure_factory(root: str = _ROOT) -> dict[str, int]:
|
||||
"""Token size of each prompt component, measured from the files on disk."""
|
||||
out = {"agent": _tok(os.path.join(root, ".opencode", "agents", "pragent.md"))}
|
||||
skills_dir = os.path.join(root, ".opencode", "skills")
|
||||
if os.path.isdir(skills_dir):
|
||||
for name in sorted(os.listdir(skills_dir)):
|
||||
p = os.path.join(skills_dir, name, "SKILL.md")
|
||||
if os.path.isfile(p):
|
||||
out[f"skill:{name}"] = _tok(p)
|
||||
for lens in ("security", "tests", "perf"):
|
||||
out[f"subagent:{lens}"] = _tok(os.path.join(root, ".opencode", "agents", f"{lens}.md"))
|
||||
return out
|
||||
|
||||
|
||||
def prefix_tokens(tier: str, factory: dict[str, int]) -> int:
|
||||
"""Stable per-step prefix: harness + agent definition + loaded skills."""
|
||||
total = HARNESS_TOKENS + factory.get("agent", 0)
|
||||
for s in ALWAYS_SKILLS + TIER_SKILLS.get(tier, ()):
|
||||
total += factory.get(f"skill:{s}", 0)
|
||||
return total
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Per-tier workload assumptions
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@dataclass
|
||||
class Tier:
|
||||
"""One tier's workload. Defaults follow the `attention-tiering` budgets."""
|
||||
|
||||
name: str
|
||||
diff_tokens: int # the diff as it lands in the brief
|
||||
steps: int # model turns in the agent loop
|
||||
file_reads: int # files read from the checkout
|
||||
tokens_per_read: int # avg tokens returned per read/grep/linter result
|
||||
output_tokens: int # assistant output across all steps (incl. reasoning)
|
||||
subagents: int = 0 # lens subagents spawned
|
||||
brief_fixed: int = 600 # brief template + PR meta + prior reviews
|
||||
share: float = 0.0 # fraction of PRs at this tier (for the monthly mix)
|
||||
_factory: dict = field(default_factory=dict, repr=False)
|
||||
|
||||
|
||||
DEFAULT_TIERS = [
|
||||
# diff_tok steps reads tok/read output subs share
|
||||
Tier("trivial", 400, 2, 0, 0, 600, 0, share=0.05),
|
||||
Tier("lite", 1500, 6, 4, 2000, 2500, 0, share=0.35),
|
||||
Tier("full", 6000, 24, 20, 3300, 12000, 0, share=0.55),
|
||||
Tier("oversized", 25000, 35, 30, 3500, 20000, 2, share=0.05),
|
||||
]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Observed runs — the calibration anchor
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
# Real usage reported by opencode's step_finish events. Keep this list
|
||||
# events. Keep this list append-only: it is the only thing separating this model
|
||||
# from a guess, and the first entry corrected the tier assumptions by ~15x.
|
||||
OBSERVED_RUNS: list[dict] = [
|
||||
{
|
||||
"label": "internal/hardening-PR (16 files, 1020 insertions / 91 deletions)",
|
||||
"date": "2026-08-18",
|
||||
"tier": "full",
|
||||
"diff_tokens": 17_600, # 16 files, 1020 insertions / 91 deletions
|
||||
"steps": 28,
|
||||
"duration_s": 348.3,
|
||||
"input": 2_071_025,
|
||||
"output": 17_303,
|
||||
"cache_read": 0,
|
||||
"cache_write": 0,
|
||||
"subagents": 0,
|
||||
},
|
||||
{
|
||||
"label": "internal/hardening-PR (same PR, two commits later)",
|
||||
"date": "2026-08-18",
|
||||
"tier": "full",
|
||||
"diff_tokens": 21_000, # same PR, two commits later
|
||||
"steps": 31,
|
||||
"duration_s": 189.8,
|
||||
"input": 2_213_077,
|
||||
"output": 9_058,
|
||||
"cache_read": 0,
|
||||
"cache_write": 0,
|
||||
"subagents": 0,
|
||||
},
|
||||
# A third run of the same PR (sha 2613b3e, 31 steps' worth of work in 330s)
|
||||
# ended without a parseable findings block and so reported no usage at all —
|
||||
# the reason `salvage_summary` now keeps the usage section on that path.
|
||||
]
|
||||
|
||||
|
||||
def observed_usage(run: dict) -> Usage:
|
||||
return Usage(
|
||||
uncached_input=run["input"] - run.get("cache_read", 0),
|
||||
cached_input=run.get("cache_read", 0),
|
||||
cache_writes=run.get("cache_write", 0),
|
||||
output=run["output"],
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class Usage:
|
||||
uncached_input: int = 0
|
||||
cached_input: int = 0
|
||||
cache_writes: int = 0
|
||||
output: int = 0
|
||||
|
||||
@property
|
||||
def total_input(self) -> int:
|
||||
return self.uncached_input + self.cached_input
|
||||
|
||||
|
||||
def tier_usage(tier: Tier, factory: dict[str, int], caching: bool = True) -> Usage:
|
||||
"""Token usage for one review at this tier.
|
||||
|
||||
The agent loop resends the whole conversation each step. The prefix + brief
|
||||
are byte-identical across steps, so with caching they are written once and
|
||||
read back on every later step; the tool-result tail grows and is charged as
|
||||
ordinary input. Without caching every step pays full input price for
|
||||
everything it has accumulated — which is the quadratic term that makes an
|
||||
uncached agent loop expensive.
|
||||
"""
|
||||
prefix = prefix_tokens(tier.name, factory)
|
||||
stable = prefix + tier.brief_fixed + tier.diff_tokens
|
||||
|
||||
# Tool results arrive one per step, after the first.
|
||||
result_steps = max(0, min(tier.file_reads, tier.steps - 1))
|
||||
per_result = tier.tokens_per_read
|
||||
|
||||
u = Usage(output=tier.output_tokens)
|
||||
|
||||
if caching:
|
||||
u.cache_writes = stable
|
||||
u.cached_input = stable * max(0, tier.steps - 1)
|
||||
u.uncached_input = 0
|
||||
else:
|
||||
u.uncached_input = stable * tier.steps
|
||||
|
||||
# The growing tail of tool results: a result produced at step i is resent on
|
||||
# every step after it, so it is counted (steps - i) times.
|
||||
tail = 0
|
||||
for i in range(1, result_steps + 1):
|
||||
tail += per_result * (tier.steps - i)
|
||||
u.uncached_input += tail
|
||||
|
||||
# Each lens subagent is its own loop: its own prefix, the diff, a few reads.
|
||||
for _ in range(tier.subagents):
|
||||
sub_prefix = HARNESS_TOKENS + factory.get("subagent:security", 600)
|
||||
sub_stable = sub_prefix + tier.diff_tokens
|
||||
sub_steps = 6
|
||||
if caching:
|
||||
u.cache_writes += sub_stable
|
||||
u.cached_input += sub_stable * (sub_steps - 1)
|
||||
else:
|
||||
u.uncached_input += sub_stable * sub_steps
|
||||
for i in range(1, 4):
|
||||
u.uncached_input += per_result * (sub_steps - i)
|
||||
u.output += 1500
|
||||
|
||||
return u
|
||||
|
||||
|
||||
def cost(u: Usage, price: Price, batch: bool = False) -> float:
|
||||
"""USD for one review's usage at these prices."""
|
||||
inp = price.batch_input if batch else price.input
|
||||
out = price.batch_output if batch else price.output
|
||||
cw = price.cache_write / 2 if batch else price.cache_write
|
||||
cr = price.cache_read / 2 if batch else price.cache_read
|
||||
return (
|
||||
u.uncached_input * inp
|
||||
+ u.cached_input * cr
|
||||
+ u.cache_writes * cw
|
||||
+ u.output * out
|
||||
) / 1_000_000
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Reporting
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def blended_cost(tiers: list[Tier], factory: dict, price: Price, caching: bool) -> float:
|
||||
"""Weighted cost of one average PR across the tier mix."""
|
||||
total_share = sum(t.share for t in tiers) or 1.0
|
||||
return sum(
|
||||
cost(tier_usage(t, factory, caching), price) * (t.share / total_share)
|
||||
for t in tiers
|
||||
)
|
||||
|
||||
|
||||
def report(tiers: list[Tier], prs_per_month: int, caching: bool, models: list[str]) -> str:
|
||||
factory = measure_factory()
|
||||
lines: list[str] = []
|
||||
|
||||
lines.append(f"Factory footprint (measured, {CHARS_PER_TOKEN} chars/token):")
|
||||
for k, v in sorted(factory.items()):
|
||||
lines.append(f" {k:<34} {v:>6,} tok")
|
||||
lines.append(f" {'harness (opencode + tool schemas, est.)':<34} {HARNESS_TOKENS:>6,} tok")
|
||||
lines.append("")
|
||||
|
||||
lines.append(f"Per-review tokens (prompt caching: {'on' if caching else 'OFF'})")
|
||||
lines.append(f" {'tier':<11} {'prefix':>8} {'uncached':>10} {'cached':>10} {'cwrite':>8} {'output':>8}")
|
||||
for t in tiers:
|
||||
u = tier_usage(t, factory, caching)
|
||||
lines.append(
|
||||
f" {t.name:<11} {prefix_tokens(t.name, factory):>8,} {u.uncached_input:>10,} "
|
||||
f"{u.cached_input:>10,} {u.cache_writes:>8,} {u.output:>8,}"
|
||||
)
|
||||
lines.append("")
|
||||
|
||||
lines.append("Cost per review (USD)")
|
||||
header = f" {'model':<18}" + "".join(f"{t.name:>12}" for t in tiers) + f"{'blended':>12}"
|
||||
lines.append(header)
|
||||
for key in models:
|
||||
p = PRICES[key]
|
||||
row = f" {p.name:<18}"
|
||||
for t in tiers:
|
||||
row += f"{cost(tier_usage(t, factory, caching), p):>12.4f}"
|
||||
row += f"{blended_cost(tiers, factory, p, caching):>12.4f}"
|
||||
lines.append(row)
|
||||
lines.append("")
|
||||
|
||||
mix = ", ".join(f"{t.name} {t.share:.0%}" for t in tiers)
|
||||
lines.append(f"Monthly at {prs_per_month} PRs/month (mix: {mix})")
|
||||
lines.append(f" {'model':<18} {'per PR':>10} {'per month':>12} {'batch -50%':>12}")
|
||||
for key in models:
|
||||
p = PRICES[key]
|
||||
per_pr = blended_cost(tiers, factory, p, caching)
|
||||
lines.append(
|
||||
f" {p.name:<18} {per_pr:>10.4f} {per_pr * prs_per_month:>12.2f}"
|
||||
f" {per_pr * prs_per_month / 2:>12.2f}"
|
||||
)
|
||||
lines.append("")
|
||||
lines.append("Batch column applies the 50% async discount; it is shown for scale only —")
|
||||
lines.append("PR review is latency-sensitive and a stateful agent loop is not batchable.")
|
||||
lines.append("")
|
||||
lines.append(observed_report(models))
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def observed_report(models: list[str]) -> str:
|
||||
"""Price the runs actually measured through the opencode usage telemetry."""
|
||||
if not OBSERVED_RUNS:
|
||||
return "No observed runs recorded yet."
|
||||
lines = ["Observed runs (measured via opencode step_finish events)"]
|
||||
for run in OBSERVED_RUNS:
|
||||
u = observed_usage(run)
|
||||
lines.append(
|
||||
f" {run['label']} — tier {run['tier']}, {run['steps']} steps, "
|
||||
f"{run['duration_s']:.0f}s, {run['input']:,} in / {run['output']:,} out, "
|
||||
f"cache {run['cache_read']:,} read / {run['cache_write']:,} write"
|
||||
)
|
||||
row = " "
|
||||
for key in models:
|
||||
p = PRICES[key]
|
||||
row += f" {p.name}: ${cost(u, p):.2f} "
|
||||
lines.append(row)
|
||||
lines.append("")
|
||||
lines.append(" NOTE: the pilot's headroom/glm-5.2 path reports zero cache read and zero")
|
||||
lines.append(" cache write, i.e. prompt caching is NOT in play today. On a provider where")
|
||||
lines.append(" it is, the stable prefix (agent + skills + brief + diff, resent every step)")
|
||||
lines.append(" drops to 0.1x — worth roughly a third of the bill on a run like the one")
|
||||
lines.append(" above. Budget with caching OFF until the measured cache columns are nonzero.")
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def main(argv: list[str] | None = None) -> int:
|
||||
ap = argparse.ArgumentParser(description="pragent per-review cost model")
|
||||
ap.add_argument("--prs-per-month", type=int, default=350)
|
||||
ap.add_argument("--mix", default="", help="trivial,lite,full,oversized as percentages")
|
||||
ap.add_argument("--no-cache", action="store_true", help="model without prompt caching")
|
||||
ap.add_argument("--models", default=",".join(PRICES))
|
||||
args = ap.parse_args(argv)
|
||||
|
||||
tiers = DEFAULT_TIERS
|
||||
if args.mix:
|
||||
shares = [float(x) for x in args.mix.split(",")]
|
||||
if len(shares) != len(tiers):
|
||||
ap.error(f"--mix needs {len(tiers)} comma-separated values")
|
||||
for t, s in zip(tiers, shares):
|
||||
t.share = s / 100.0
|
||||
|
||||
models = [m.strip() for m in args.models.split(",") if m.strip()]
|
||||
unknown = [m for m in models if m not in PRICES]
|
||||
if unknown:
|
||||
ap.error(f"unknown model(s): {', '.join(unknown)}")
|
||||
|
||||
print(report(tiers, args.prs_per_month, not args.no_cache, models))
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -0,0 +1,468 @@
|
||||
#!/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 view, 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": _review_input(repo, index, sha, title),
|
||||
"output": _review_output(summary, findings),
|
||||
"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",
|
||||
# Repeated from the trace on purpose: an evaluator's variable
|
||||
# mapping reads the *observation's* input/output, so a generation
|
||||
# left blank cannot be judged at all.
|
||||
"input": _review_input(repo, index, sha, title),
|
||||
"output": _review_output(summary, findings),
|
||||
}
|
||||
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
|
||||
|
||||
|
||||
MAX_JUDGED_FINDINGS = 25
|
||||
_FIELD_CAP = 600
|
||||
|
||||
|
||||
def _review_input(repo: str, index, sha: str, title: str) -> dict:
|
||||
return {"repo": repo, "pr": index, "sha": sha, "title": title}
|
||||
|
||||
|
||||
def _review_output(summary: str, findings) -> dict:
|
||||
"""What the reviewer actually said, in a shape an evaluator can read.
|
||||
|
||||
The findings themselves are included, not just their count. A judge given
|
||||
only `{"summary": ..., "findings": 3}` can say nothing about whether those
|
||||
three findings are specific, actionable, or consistent with the summary —
|
||||
which is the whole question worth asking of a reviewer that has no ground
|
||||
truth to check against.
|
||||
|
||||
Capped rather than complete: this rides in every ingestion batch, and a
|
||||
review with 80 findings would push the payload past what is reasonable to
|
||||
store per trace. `finding_count` stays exact so nothing reading the count
|
||||
is misled by the cap.
|
||||
"""
|
||||
items = list(findings or [])
|
||||
return {
|
||||
"summary": summary[:2000],
|
||||
"finding_count": len(items),
|
||||
"findings_truncated": len(items) > MAX_JUDGED_FINDINGS,
|
||||
"findings": [
|
||||
{
|
||||
"path": f.get("path"),
|
||||
"line": f.get("line"),
|
||||
"severity": f.get("severity"),
|
||||
"problem": str(f.get("problem") or "")[:_FIELD_CAP],
|
||||
"fix": str(f.get("fix") or "")[:_FIELD_CAP],
|
||||
}
|
||||
for f in items[:MAX_JUDGED_FINDINGS]
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
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
|
||||
Reference in New Issue
Block a user