coulomb/llm-connect
4
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Implement LLM-WP-0007: Kimi K3 default and EUR spend reporting
Some checks failed
CI Smoke / host-smoke (push) Successful in 0s
Details
CI Smoke / container-smoke (push) Has been cancelled
Details

Browse source 
Add moonshotai/kimi-k3 as OpenRouter basemodel default after live smoke,
USD→EUR cost conversion, append-only usage ledger, and CLI run/cost/spend
week commands with token and euro reporting.
This commit is contained in:
tegwick 2026-08-03 22:00:34 +02:00
parent f3121c3f1f
commit 09aa1f3604
21 changed files with 1396 additions and 103 deletions
Download patch file Download diff file Expand all files Collapse all files

147
README.md
View file

@ -35,7 +35,8 @@ pip install llm-connect
```python
from llm_connect import create_adapter
# OpenRouter
# OpenRouter (default model: moonshotai/kimi-k3)
adapter = create_adapter("openrouter")
adapter = create_adapter("openrouter", model="anthropic/claude-sonnet-4")
# Gemini (uses GEMINI_API_KEY env var or apikey-geminifree.txt)
@ -48,6 +49,34 @@ adapter = create_adapter("openai", model="gpt-4.1-mini")
adapter = create_adapter("claude-code")
```
## CLI
```bash
# Run a prompt; content on stdout, tokens + €/USD cost on stderr
llm-connect run "Summarise the value chain concept."
llm-connect run "Hello" --provider mock --model moonshotai/kimi-k3 --json
# Estimate cost without calling a provider
llm-connect cost estimate --model moonshotai/kimi-k3 \
--prompt-tokens 1000 --completion-tokens 500
# Weekly spend (Mon-based; default timezone Europe/Berlin)
llm-connect spend week # current week: Mon 00:00 → now
llm-connect spend week --last # previous week: Mon → Sun
llm-connect spend week --json
```
Usage events append to a JSONL ledger (default
`~/.local/share/llm-connect/usage.jsonl`; override with `--ledger` or
`LLM_CONNECT_USAGE_LEDGER`). Costs are list-price estimates (USD rate table +
EUR via snapshot or `LLM_CONNECT_EUR_PER_USD`).
| Env | Purpose |
|---|---|
| `LLM_CONNECT_USAGE_LEDGER` | Usage JSONL path; enables library/server auto-recording when set |
| `LLM_CONNECT_EUR_PER_USD` | Euros per one USD (FX override) |
| `LLM_CONNECT_TZ` | Timezone for weekly spend windows (default `Europe/Berlin`) |
## API keys
Keys are resolved in this order (first found wins):
@ -73,15 +102,15 @@ config = RunConfig(
)
```
| Field | Default | Description |
|---|---|---|
| `model_name` | `"gpt-4"` | Model identifier (adapter may override) |
| `temperature` | `0.7` | Sampling temperature |
| `max_tokens` | `2000` | Maximum output tokens |
| `model_params` | `{}` | Portable extras translated by each adapter; see `docs/adapter-model-params.md` |
| `max_depth` | `3` | Max nesting depth for recursive calls |
| `skip_if_exists` | `True` | Skip if identical input hash already processed |
| `timeout_seconds` | `300` | Request timeout |
| Field | Default | Description |
|---|---|---|
| `model_name` | `"gpt-4"` | Model identifier (adapter may override) |
| `temperature` | `0.7` | Sampling temperature |
| `max_tokens` | `2000` | Maximum output tokens |
| `model_params` | `{}` | Portable extras translated by each adapter; see `docs/adapter-model-params.md` |
| `max_depth` | `3` | Max nesting depth for recursive calls |
| `skip_if_exists` | `True` | Skip if identical input hash already processed |
| `timeout_seconds` | `300` | Request timeout |
### `LLMResponse`
@ -92,55 +121,55 @@ response = adapter.execute_prompt(prompt, config)
print(response.content) # generated text
print(response.model) # model actually used
print(response.usage) # {"prompt_tokens": …, "completion_tokens": …, "total_tokens": …}
print(response.finish_reason) # "stop", "length", etc.
```
## Server diagnostics
Serve mode can include a debug envelope without changing normal responses:
```bash
LLM_CONNECT_DEBUG=1 python -m llm_connect.server --provider openrouter
curl 'http://127.0.0.1:8080/execute?debug=1' -d '{"prompt":"hi"}'
```
Set `LLM_CONNECT_AUDIT_DIR=/path/to/audit` to write per-call replay records,
then parse one without another provider call:
```bash
python -m llm_connect.replay /path/to/audit/record.json --json
```
## Server runtime profiles
Serve mode enables named runtime profiles by default. A client can send
`config.model_name="custodian-triage-balanced"` and the server resolves it to
the configured provider/model before calling the adapter.
Useful runtime environment variables:
```bash
LLM_CONNECT_HOST=0.0.0.0
LLM_CONNECT_PORT=8080
LLM_CONNECT_PROVIDER=openrouter
LLM_CONNECT_MODEL=google/gemini-2.5-flash
LLM_CONNECT_CUSTODIAN_TRIAGE_PROVIDER=openrouter
LLM_CONNECT_CUSTODIAN_TRIAGE_MODEL=google/gemini-2.5-flash
```
For local smoke tests without provider credentials:
```bash
export LLM_CONNECT_MOCK_RESPONSE="$(python -c 'import json; print(json.dumps(json.load(open("fixtures/activity_core/daily-triage-valid-content.json"))))')"
python -m llm_connect.server --provider mock
python scripts/smoke_activity_core_endpoint.py --url http://127.0.0.1:8080
```
Disable profile dispatch with `--disable-profiles`. Set
`LLM_CONNECT_STRICT_PROFILES=1` or pass `--strict-profiles` to reject direct
model names that are not configured profiles.
## Writing your own adapter
print(response.finish_reason) # "stop", "length", etc.
```
## Server diagnostics
Serve mode can include a debug envelope without changing normal responses:
```bash
LLM_CONNECT_DEBUG=1 python -m llm_connect.server --provider openrouter
curl 'http://127.0.0.1:8080/execute?debug=1' -d '{"prompt":"hi"}'
```
Set `LLM_CONNECT_AUDIT_DIR=/path/to/audit` to write per-call replay records,
then parse one without another provider call:
```bash
python -m llm_connect.replay /path/to/audit/record.json --json
```
## Server runtime profiles
Serve mode enables named runtime profiles by default. A client can send
`config.model_name="custodian-triage-balanced"` and the server resolves it to
the configured provider/model before calling the adapter.
Useful runtime environment variables:
```bash
LLM_CONNECT_HOST=0.0.0.0
LLM_CONNECT_PORT=8080
LLM_CONNECT_PROVIDER=openrouter
LLM_CONNECT_MODEL=google/gemini-2.5-flash
LLM_CONNECT_CUSTODIAN_TRIAGE_PROVIDER=openrouter
LLM_CONNECT_CUSTODIAN_TRIAGE_MODEL=google/gemini-2.5-flash
```
For local smoke tests without provider credentials:
```bash
export LLM_CONNECT_MOCK_RESPONSE="$(python -c 'import json; print(json.dumps(json.load(open("fixtures/activity_core/daily-triage-valid-content.json"))))')"
python -m llm_connect.server --provider mock
python scripts/smoke_activity_core_endpoint.py --url http://127.0.0.1:8080
```
Disable profile dispatch with `--disable-profiles`. Set
`LLM_CONNECT_STRICT_PROFILES=1` or pass `--strict-profiles` to reject direct
model names that are not configured profiles.
## Writing your own adapter
```python
from llm_connect import LLMAdapter, RunConfig, LLMResponse

19
contracts/functional/costs.md
View file

@ -1,7 +1,8 @@
# Cost Estimates
`llm_connect.costs` converts token estimates or observed token counts into
USD estimates using `ModelRateRegistry`.
USD estimates using `ModelRateRegistry`, and optionally into EUR via
`llm_connect.fx`.
## Contract
@ -9,6 +10,7 @@ USD estimates using `ModelRateRegistry`.
from llm_connect import estimate_cost
estimate = estimate_cost("openai/gpt-4o-mini", 28_000, 7_500)
estimate = estimate_cost("moonshotai/kimi-k3", 1_000, 500, fx=0.92)
```
For known models the result is:
@ -17,9 +19,16 @@ For known models the result is:
- `prompt_cost_usd`: prompt-token component.
- `completion_cost_usd`: completion-token component.
- `cost_source`: `rate_table:<model_id>`.
- `cost_eur` / `prompt_cost_eur` / `completion_cost_eur`: EUR display amounts
when FX is available (default: bundled snapshot, override with `fx=` or
env `LLM_CONNECT_EUR_PER_USD` as euros-per-USD).
- `fx_source`: how EUR was derived (`explicit`, `env:…`, `snapshot:…`).
Unknown models return `CostEstimate(cost_usd=None, cost_source="unknown")`.
Missing rates are never silently treated as zero cost.
Unknown models return `CostEstimate(cost_usd=None, cost_source="unknown")`
with EUR fields also `None`. Missing rates or FX are never silently treated
as zero cost.
The module also exposes `CostModel(registry=...)` for callers that prefer to
carry a registry object and call `model.estimate_cost(...)`.
Pass `apply_fx=False` to skip EUR conversion.
The module also exposes `CostModel(registry=..., fx=...)` for callers that
prefer to carry a registry object and call `model.estimate_cost(...)`.

9
contracts/functional/rates.md
View file

@ -8,7 +8,8 @@ post-hoc estimates.
- `ModelRate` records `model_id`, prompt and completion rates in USD per
1,000 tokens, `currency`, `source_url`, and `captured_at`.
- `ModelRateRegistry.default()` returns the bundled OpenRouter snapshot
captured on `2026-05-17`.
(base rates captured `2026-05-17`; `moonshotai/kimi-k3` captured
`2026-08-03` at `$0.003` / `$0.015` per 1k prompt/completion).
- `ModelRateRegistry.from_yaml(path)` accepts the package/consumer override
shape:
@ -21,10 +22,14 @@ rates:
openai/gpt-4o-mini:
prompt_per_1k: 0.00015
completion_per_1k: 0.00060
moonshotai/kimi-k3:
prompt_per_1k: 0.003
completion_per_1k: 0.015
```
- `merged_with(override)` returns a new registry where matching override
entries replace default entries by `model_id`.
Rates are a static snapshot. Consumers decide whether `captured_at` is fresh
enough for their workflow.
enough for their workflow. List prices remain USD; EUR display conversion is
handled by `llm_connect.fx` / `estimate_cost`.

33
contracts/functional/usage-ledger.md Normal file
View file

@ -0,0 +1,33 @@
# Usage Ledger and Spend Windows
`llm_connect.usage` stores append-only spend events for operator cost reporting.
It is separate from `QualityLedger` (adaptive routing quality signals).
## UsageEvent
Each event records provider, model, token counts, estimated `cost_usd` /
`cost_eur`, cost/FX sources, call `source` (`cli` | `server` | `library`),
and `recorded_at` (UTC).
## Paths
1. Explicit `--ledger` / constructor path
2. Env `LLM_CONNECT_USAGE_LEDGER`
3. `$XDG_DATA_HOME/llm-connect/usage.jsonl`
4. `~/.local/share/llm-connect/usage.jsonl`
Library and server auto-recording only runs when `LLM_CONNECT_USAGE_LEDGER` is
set (opt-in). CLI `run` records by default to the resolved path unless
`--no-ledger`.
## Week windows
`week_window("current"|"last", tz=...)` returns Monday-based half-open ranges
in local time (default `Europe/Berlin`, override with `LLM_CONNECT_TZ` or
`--tz`):
- **current**: Monday 00:00 → now
- **last**: previous Monday 00:00 → this Monday 00:00 (full Mon–Sun week)
`UsageLedger.sum_range(start, end)` aggregates tokens and costs for
`[start, end)`.

21
llm_connect/__init__.py
View file

@ -8,7 +8,7 @@ Quick start::
from llm_connect import create_adapter
adapter = create_adapter("openrouter", model="anthropic/claude-sonnet-4")
adapter = create_adapter("openrouter", model="moonshotai/kimi-k3")
response = adapter.execute_prompt(prompt, run_config)
"""
@ -16,6 +16,7 @@ from llm_connect.adapter import ErrorLLMAdapter, LLMAdapter, MockLLMAdapter
from llm_connect.claude_code import ClaudeCodeAdapter
from llm_connect.config import LLMConfig, load_config
from llm_connect.costs import CostEstimate, CostModel, estimate_cost
from llm_connect.fx import FxRate, resolve_fx_rate
from llm_connect.embedding_adapter import EmbeddingAdapter
from llm_connect.embedding_cache import EmbeddingCache
from llm_connect.embedding_factory import create_embedding_adapter
@ -71,6 +72,15 @@ from llm_connect.similarity import (
find_similar_pairs,
similarity_matrix,
)
from llm_connect.usage import (
UsageEvent,
UsageLedger,
UsageSummary,
default_usage_ledger_path,
event_from_response,
maybe_record_usage,
week_window,
)
__all__ = [
"RunConfig",
@ -120,6 +130,15 @@ __all__ = [
"CostEstimate",
"CostModel",
"estimate_cost",
"FxRate",
"resolve_fx_rate",
"UsageEvent",
"UsageLedger",
"UsageSummary",
"default_usage_ledger_path",
"event_from_response",
"maybe_record_usage",
"week_window",
"TokenEstimate",
"Observation",
"ProblemClass",

20
llm_connect/adapter.py
View file

@ -93,6 +93,26 @@ class LLMAdapter(ABC):
if config.budget_tracker is not None:
tokens = response.usage.get("total_tokens", 0)
config.budget_tracker.consume(tokens)
self._maybe_record_usage(response)
def _maybe_record_usage(self, response: LLMResponse) -> None:
"""Opt-in usage ledger write when ``LLM_CONNECT_USAGE_LEDGER`` is set."""
try:
from llm_connect.usage import maybe_record_usage
provider = (
(response.metadata or {}).get("provider")
or getattr(self, "provider_id", None)
or type(self).__name__
)
maybe_record_usage(
response,
provider=str(provider),
source="library",
)
except Exception:
# Spend accounting must never break inference.
return
class MockLLMAdapter(LLMAdapter):

219
llm_connect/cli.py
View file

@ -1,16 +1,30 @@
"""Command-line helpers for llm-connect registries."""
"""Command-line helpers for llm-connect registries, run, and spend."""
from __future__ import annotations
import argparse
import json
import sys
from collections.abc import Iterable, Mapping
from datetime import datetime
from pathlib import Path
from typing import Any
from llm_connect.costs import estimate_cost
from llm_connect.factory import create_adapter
from llm_connect.fx import resolve_fx_rate
from llm_connect.models import RunConfig
from llm_connect.problem_classes import ProblemClass, ProblemClassRegistry
from llm_connect.quality import QualityLedger
from llm_connect.rates import ModelRateRegistry
from llm_connect.usage import (
UsageLedger,
default_usage_ledger_path,
event_from_response,
format_cost_footer,
suppress_auto_usage_record,
week_window,
)
def main(argv: list[str] | None = None) -> int:
@ -43,6 +57,55 @@ def _build_parser() -> argparse.ArgumentParser:
class_fit.add_argument("--min-observations", type=int, default=3)
class_fit.add_argument("--json", action="store_true", help="Emit JSON")
class_fit.set_defaults(func=_classes_fit)
run = commands.add_parser("run", help="Execute a prompt and report tokens + cost")
run.add_argument("prompt", help="Prompt text to send")
run.add_argument("--provider", default="openrouter", help="Provider key (default openrouter)")
run.add_argument("--model", default=None, help="Model id (provider default if omitted)")
run.add_argument("--temperature", type=float, default=0.7)
run.add_argument("--max-tokens", type=int, default=2000)
run.add_argument("--timeout", type=int, default=300, help="Timeout seconds")
run.add_argument("--ledger", type=Path, default=None, help="Usage ledger JSONL path")
run.add_argument("--no-ledger", action="store_true", help="Do not append to the usage ledger")
run.add_argument("--json", action="store_true", help="Emit JSON response + cost")
run.add_argument(
"--eur-per-usd",
type=float,
default=None,
help="Override FX rate (euros per one USD)",
)
run.set_defaults(func=_run_prompt)
cost = commands.add_parser("cost", help="Cost estimation helpers")
cost_commands = cost.add_subparsers(dest="cost_command", required=True)
cost_estimate = cost_commands.add_parser(
"estimate",
help="Estimate cost for token counts without calling a provider",
)
cost_estimate.add_argument("--model", required=True, help="Model id")
cost_estimate.add_argument("--prompt-tokens", type=int, required=True)
cost_estimate.add_argument("--completion-tokens", type=int, default=0)
cost_estimate.add_argument("--rates", type=Path, help="YAML registry overlay")
cost_estimate.add_argument("--eur-per-usd", type=float, default=None)
cost_estimate.add_argument("--json", action="store_true", help="Emit JSON")
cost_estimate.set_defaults(func=_cost_estimate)
spend = commands.add_parser("spend", help="Inspect recorded spend")
spend_commands = spend.add_subparsers(dest="spend_command", required=True)
spend_week = spend_commands.add_parser(
"week",
help="Weekly spend: current Mon→now, or --last for previous Mon–Sun",
)
spend_week.add_argument(
"--last",
action="store_true",
help="Previous calendar week (Mon 00:00 → Sun end / this Mon 00:00)",
)
spend_week.add_argument("--ledger", type=Path, default=None, help="Usage ledger JSONL path")
spend_week.add_argument("--tz", default=None, help="Timezone (default Europe/Berlin)")
spend_week.add_argument("--json", action="store_true", help="Emit JSON")
spend_week.set_defaults(func=_spend_week)
return parser
@ -123,6 +186,160 @@ def _classes_fit(args: argparse.Namespace) -> int:
return 0
def _run_prompt(args: argparse.Namespace) -> int:
fx = resolve_fx_rate(args.eur_per_usd)
adapter_kwargs: dict[str, Any] = {}
if args.model is not None:
adapter_kwargs["model"] = args.model
adapter = create_adapter(args.provider, **adapter_kwargs)
config = RunConfig(
model_name=args.model or "gpt-4",
temperature=args.temperature,
max_tokens=args.max_tokens,
timeout_seconds=args.timeout,
)
with suppress_auto_usage_record():
response = adapter.execute_prompt(args.prompt, config)
usage = response.usage or {}
prompt_tokens = int(usage.get("prompt_tokens") or 0)
completion_tokens = int(usage.get("completion_tokens") or 0)
total_tokens = int(usage.get("total_tokens") or (prompt_tokens + completion_tokens))
model_id = response.model or args.model or "unknown"
estimate = estimate_cost(
model_id,
prompt_tokens,
completion_tokens,
fx=fx,
)
if not args.no_ledger:
ledger_path = args.ledger or default_usage_ledger_path()
event = event_from_response(
response,
provider=args.provider,
source="cli",
fx=fx,
)
UsageLedger(ledger_path).append(event)
footer = format_cost_footer(
estimate,
prompt_tokens=prompt_tokens,
completion_tokens=completion_tokens,
total_tokens=total_tokens,
model_id=model_id,
)
if args.json:
print(
json.dumps(
{
"content": response.content,
"model": model_id,
"usage": {
"prompt_tokens": prompt_tokens,
"completion_tokens": completion_tokens,
"total_tokens": total_tokens,
},
"cost_usd": estimate.cost_usd,
"cost_eur": estimate.cost_eur,
"cost_source": estimate.cost_source,
"fx_source": estimate.fx_source,
"finish_reason": response.finish_reason,
},
indent=2,
sort_keys=True,
)
)
return 0
print(response.content)
print(footer, file=sys.stderr)
return 0
def _cost_estimate(args: argparse.Namespace) -> int:
if args.prompt_tokens < 0 or args.completion_tokens < 0:
raise SystemExit("token counts must be non-negative")
registry = ModelRateRegistry.default()
if args.rates:
registry = registry.merged_with(ModelRateRegistry.from_yaml(args.rates))
estimate = estimate_cost(
args.model,
args.prompt_tokens,
args.completion_tokens,
registry=registry,
fx=args.eur_per_usd,
)
total = args.prompt_tokens + args.completion_tokens
if args.json:
print(
json.dumps(
{
"model": args.model,
"prompt_tokens": args.prompt_tokens,
"completion_tokens": args.completion_tokens,
"total_tokens": total,
"cost_usd": estimate.cost_usd,
"cost_eur": estimate.cost_eur,
"cost_source": estimate.cost_source,
"fx_source": estimate.fx_source,
},
indent=2,
sort_keys=True,
)
)
return 0
print(
format_cost_footer(
estimate,
prompt_tokens=args.prompt_tokens,
completion_tokens=args.completion_tokens,
total_tokens=total,
model_id=args.model,
)
)
return 0
def _spend_week(args: argparse.Namespace) -> int:
which = "last" if args.last else "current"
start, end = week_window(which, tz=args.tz)
ledger_path = args.ledger or default_usage_ledger_path()
summary = UsageLedger(ledger_path).sum_range(start, end)
if args.json:
payload = summary.to_dict()
payload["which"] = which
payload["tz"] = args.tz or "Europe/Berlin"
payload["ledger"] = str(ledger_path)
print(json.dumps(payload, indent=2, sort_keys=True))
return 0
label = "last week (Mon–Sun)" if which == "last" else "current week (Mon–now)"
print(f"spend {label}")
print(f"window: {_fmt_local(start)} → {_fmt_local(end)}")
print(f"ledger: {ledger_path}")
print(f"events: {summary.event_count}")
print(
f"tokens: in={summary.prompt_tokens} out={summary.completion_tokens} "
f"total={summary.total_tokens}"
)
usd = "unknown" if summary.cost_usd is None else f"{summary.cost_usd:.6f}"
eur = "unknown" if summary.cost_eur is None else f"{summary.cost_eur:.6f}"
print(f"cost: eur={eur} usd={usd}")
if summary.unknown_cost_events:
print(f"unknown_cost_events: {summary.unknown_cost_events}")
return 0
def _fmt_local(dt: datetime) -> str:
return dt.isoformat()
def _classes_payload(classes: Iterable[ProblemClass]) -> dict[str, dict[str, Any]]:
return {
problem_class.name: {

4
llm_connect/config.py
View file

@ -24,7 +24,7 @@ class LLMConfig:
"""
provider: str = "openrouter"
model: str = "anthropic/claude-sonnet-4"
model: str = "moonshotai/kimi-k3"
api_key: Optional[str] = None
api_base: str = "https://openrouter.ai/api/v1"
claude_cli_path: str = "claude"
@ -101,7 +101,7 @@ def load_config(
defaults: Dict[str, Any] = {
"provider": provider,
"model": model or "anthropic/claude-sonnet-4",
"model": model or "moonshotai/kimi-k3",
"api_key": resolved_key,
}
defaults.update(overrides)

47
llm_connect/costs.py
View file

@ -5,17 +5,27 @@ from __future__ import annotations
from dataclasses import dataclass
from typing import Any
from llm_connect.fx import FxRate, resolve_fx_rate
from llm_connect.rates import ModelRateRegistry
@dataclass(frozen=True)
class CostEstimate:
"""Cost estimate split by prompt and completion token spend."""
"""Cost estimate split by prompt and completion token spend.
USD fields come from the rate table. EUR fields are derived via :class:`FxRate`
when conversion is available; missing FX yields ``cost_eur=None`` with an
explicit ``fx_source`` rather than silently treating spend as zero.
"""
cost_usd: float | None
cost_source: str
prompt_cost_usd: float | None = None
completion_cost_usd: float | None = None
cost_eur: float | None = None
prompt_cost_eur: float | None = None
completion_cost_eur: float | None = None
fx_source: str | None = None
def estimate_cost(
@ -24,11 +34,16 @@ def estimate_cost(
completion_tokens: int = 0,
*,
registry: ModelRateRegistry | None = None,
fx: FxRate | float | None = None,
apply_fx: bool = True,
) -> CostEstimate:
"""Estimate USD cost for token counts using *registry*.
"""Estimate USD (and optionally EUR) cost for token counts using *registry*.
Unknown models return ``CostEstimate(None, "unknown")`` so callers can
record uncertainty explicitly instead of treating missing prices as zero.
When *apply_fx* is true (default), EUR fields are filled using *fx* or the
resolved default FX rate. Pass ``apply_fx=False`` to skip conversion.
"""
prompt_count = _non_negative_int("prompt_tokens", prompt_tokens)
completion_count = _non_negative_int("completion_tokens", completion_tokens)
@ -39,11 +54,31 @@ def estimate_cost(
prompt_cost = (prompt_count / 1000.0) * rate.prompt_per_1k
completion_cost = (completion_count / 1000.0) * rate.completion_per_1k
cost_usd = prompt_cost + completion_cost
cost_eur = None
prompt_cost_eur = None
completion_cost_eur = None
fx_source: str | None = None
if apply_fx:
resolved = resolve_fx_rate(fx)
if resolved is None:
fx_source = "unknown"
else:
fx_source = resolved.source
cost_eur = resolved.usd_to_eur(cost_usd)
prompt_cost_eur = resolved.usd_to_eur(prompt_cost)
completion_cost_eur = resolved.usd_to_eur(completion_cost)
return CostEstimate(
cost_usd=prompt_cost + completion_cost,
cost_usd=cost_usd,
cost_source=f"rate_table:{rate.model_id}",
prompt_cost_usd=prompt_cost,
completion_cost_usd=completion_cost,
cost_eur=cost_eur,
prompt_cost_eur=prompt_cost_eur,
completion_cost_eur=completion_cost_eur,
fx_source=fx_source,
)
@ -52,6 +87,8 @@ class CostModel:
"""Small wrapper for callers that prefer an object over a free function."""
registry: ModelRateRegistry | None = None
fx: FxRate | float | None = None
apply_fx: bool = True
def estimate_cost(
self,
@ -59,12 +96,14 @@ class CostModel:
prompt_tokens: int,
completion_tokens: int = 0,
) -> CostEstimate:
"""Estimate cost using this model's registry."""
"""Estimate cost using this model's registry and FX settings."""
return estimate_cost(
model_id,
prompt_tokens,
completion_tokens,
registry=self.registry,
fx=self.fx,
apply_fx=self.apply_fx,
)

93
llm_connect/fx.py Normal file
View file

@ -0,0 +1,93 @@
"""USD to EUR conversion helpers for display and spend reporting.
Rate tables remain USD-denominated (OpenRouter list prices). This module only
converts already-estimated USD amounts into euros.
"""
from __future__ import annotations
import os
from dataclasses import dataclass
from typing import Any
# Snapshot: euros per one US dollar. Operator can override via env.
DEFAULT_EUR_PER_USD = 0.92
DEFAULT_FX_CAPTURED_AT = "2026-08-03"
DEFAULT_FX_SOURCE = f"snapshot:{DEFAULT_FX_CAPTURED_AT}"
ENV_EUR_PER_USD = "LLM_CONNECT_EUR_PER_USD"
@dataclass(frozen=True)
class FxRate:
"""EUR conversion factor for one USD."""
eur_per_usd: float
source: str
captured_at: str = ""
def __post_init__(self) -> None:
rate = _positive_float("eur_per_usd", self.eur_per_usd)
source = str(self.source or "").strip()
if not source:
raise ValueError("source must be a non-empty string")
object.__setattr__(self, "eur_per_usd", rate)
object.__setattr__(self, "source", source)
object.__setattr__(self, "captured_at", str(self.captured_at or ""))
def usd_to_eur(self, amount_usd: float) -> float:
"""Convert a USD amount to EUR."""
return float(amount_usd) * self.eur_per_usd
def resolve_fx_rate(
explicit: FxRate | float | None = None,
*,
env: dict[str, str] | None = None,
) -> FxRate | None:
"""Resolve an FX rate from explicit value, environment, or bundled snapshot.
Resolution order:
1. *explicit* :class:`FxRate` or positive float (euros per USD)
2. Environment variable ``LLM_CONNECT_EUR_PER_USD``
3. Bundled snapshot rate
Returns ``None`` only when an explicit/env value is present but invalid
callers should pass nothing to get the snapshot. Empty env string falls
through to the snapshot.
"""
if explicit is not None:
if isinstance(explicit, FxRate):
return explicit
return FxRate(
eur_per_usd=float(explicit),
source="explicit",
captured_at="",
)
environ = env if env is not None else os.environ
raw = environ.get(ENV_EUR_PER_USD)
if raw is not None and str(raw).strip() != "":
return FxRate(
eur_per_usd=_positive_float(ENV_EUR_PER_USD, raw),
source=f"env:{ENV_EUR_PER_USD}",
captured_at="",
)
return FxRate(
eur_per_usd=DEFAULT_EUR_PER_USD,
source=DEFAULT_FX_SOURCE,
captured_at=DEFAULT_FX_CAPTURED_AT,
)
def _positive_float(name: str, value: Any) -> float:
if isinstance(value, bool):
raise ValueError(f"{name} must be a positive number")
try:
number = float(value)
except (TypeError, ValueError) as exc:
raise ValueError(f"{name} must be a positive number") from exc
if number <= 0:
raise ValueError(f"{name} must be a positive number")
return number

2
llm_connect/openrouter.py
View file

@ -12,7 +12,7 @@ from llm_connect.config import LLMConfig, find_project_root, resolve_api_key
from llm_connect.exceptions import LLMAPIError, LLMRateLimitError
from llm_connect.models import LLMResponse, RunConfig
_DEFAULT_MODEL = "anthropic/claude-sonnet-4"
_DEFAULT_MODEL = "moonshotai/kimi-k3"
class OpenRouterAdapter(LLMAdapter):

20
llm_connect/rates.py
View file

@ -91,7 +91,8 @@ class ModelRateRegistry:
return ModelRateRegistry(merged)
_DEFAULT_RATES: dict[str, tuple[float, float]] = {
# prompt_per_1k, completion_per_1k, optional captured_at override
_DEFAULT_RATES: dict[str, tuple[float, float] | tuple[float, float, str]] = {
"openai/gpt-4o-mini": (0.00015, 0.00060),
"openai/gpt-4o": (0.0025, 0.01),
"openai/gpt-4-turbo": (0.01, 0.03),
@ -101,21 +102,28 @@ _DEFAULT_RATES: dict[str, tuple[float, float]] = {
"google/gemini-1.5-flash": (0.000075, 0.0003),
"google/gemini-1.5-pro": (0.00125, 0.005),
"meta-llama/llama-3.1-70b-instruct": (0.00059, 0.00079),
# OpenRouter list price 2026-08-03: $0.000003/prompt token, $0.000015/completion
"moonshotai/kimi-k3": (0.003, 0.015, "2026-08-03"),
}
def _default_rate_payload() -> dict[str, ModelRate]:
return {
model_id: ModelRate(
rates: dict[str, ModelRate] = {}
for model_id, values in _DEFAULT_RATES.items():
if len(values) == 3:
prompt_rate, completion_rate, captured_at = values # type: ignore[misc]
else:
prompt_rate, completion_rate = values # type: ignore[misc]
captured_at = DEFAULT_RATE_CAPTURED_AT
rates[model_id] = ModelRate(
model_id=model_id,
prompt_per_1k=prompt_rate,
completion_per_1k=completion_rate,
currency=DEFAULT_RATE_CURRENCY,
source_url=DEFAULT_RATE_SOURCE_URL,
captured_at=DEFAULT_RATE_CAPTURED_AT,
captured_at=str(captured_at),
)
for model_id, (prompt_rate, completion_rate) in _DEFAULT_RATES.items()
}
return rates
def _coerce_rate(model_id: str, rate: ModelRate | Mapping[str, Any]) -> ModelRate:

14
llm_connect/server.py
View file

@ -45,6 +45,7 @@ from llm_connect.exceptions import (
)
from llm_connect.models import LLMResponse, RunConfig
from llm_connect.profiles import ProfiledLLMAdapter, default_runtime_profiles
from llm_connect.usage import maybe_record_usage, suppress_auto_usage_record
class _Handler(BaseHTTPRequestHandler):
@ -101,8 +102,19 @@ class _Handler(BaseHTTPRequestHandler):
"Adapter rejected RunConfig",
context={"model_name": config.model_name},
)
response = adapter.execute_prompt(prompt, config)
with suppress_auto_usage_record():
response = adapter.execute_prompt(prompt, config)
latency = time.time() - start
provider = (
(response.metadata or {}).get("provider")
or getattr(adapter, "provider_id", None)
or type(adapter).__name__
)
maybe_record_usage(
response,
provider=str(provider),
source="server",
)
body = response.to_dict()
debug = diagnostics.to_dict() if diagnostics is not None else None
if debug_enabled and debug is not None:

473
llm_connect/usage.py Normal file
View file

@ -0,0 +1,473 @@
"""Append-only usage ledger for operator spend reporting.
Separate from :class:`~llm_connect.quality.QualityLedger`, which stores adaptive
routing quality signals. This ledger is for tokens + estimated cost accounting.
"""
from __future__ import annotations
import contextvars
import json
import os
import threading
from contextlib import contextmanager
from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any, Iterator, TextIO
from zoneinfo import ZoneInfo
from llm_connect.costs import CostEstimate, estimate_cost
from llm_connect.fx import FxRate
from llm_connect.models import LLMResponse
from llm_connect.rates import ModelRateRegistry
ENV_USAGE_LEDGER = "LLM_CONNECT_USAGE_LEDGER"
ENV_TZ = "LLM_CONNECT_TZ"
DEFAULT_TZ = "Europe/Berlin"
_PATH_LOCKS: dict[Path, threading.Lock] = {}
_PATH_LOCKS_GUARD = threading.Lock()
_AUTO_RECORD_DISABLED: contextvars.ContextVar[bool] = contextvars.ContextVar(
"llm_connect_auto_record_disabled",
default=False,
)
@contextmanager
def suppress_auto_usage_record() -> Iterator[None]:
"""Disable env-based auto recording (e.g. while CLI records explicitly)."""
token = _AUTO_RECORD_DISABLED.set(True)
try:
yield
finally:
_AUTO_RECORD_DISABLED.reset(token)
def _utc_now() -> datetime:
return datetime.now(timezone.utc)
def _normalise_datetime(value: datetime | str) -> datetime:
if isinstance(value, datetime):
dt = value
elif isinstance(value, str):
dt = datetime.fromisoformat(value.replace("Z", "+00:00"))
else:
raise TypeError(f"Expected datetime or ISO string, got {type(value).__name__}")
if dt.tzinfo is None:
return dt.replace(tzinfo=timezone.utc)
return dt.astimezone(timezone.utc)
def _serialise_datetime(value: datetime) -> str:
return _normalise_datetime(value).isoformat().replace("+00:00", "Z")
def _validate_non_negative_int(name: str, value: int) -> None:
if not isinstance(value, int) or isinstance(value, bool) or value < 0:
raise ValueError(f"{name} must be a non-negative integer")
def _path_lock(path: Path) -> threading.Lock:
resolved = path.resolve()
with _PATH_LOCKS_GUARD:
lock = _PATH_LOCKS.get(resolved)
if lock is None:
lock = threading.Lock()
_PATH_LOCKS[resolved] = lock
return lock
def _lock_file(handle: TextIO) -> None:
if os.name == "nt":
import msvcrt
msvcrt.locking(handle.fileno(), msvcrt.LK_LOCK, 1)
else:
import fcntl
fcntl.flock(handle.fileno(), fcntl.LOCK_EX)
def _unlock_file(handle: TextIO) -> None:
if os.name == "nt":
import msvcrt
msvcrt.locking(handle.fileno(), msvcrt.LK_UNLCK, 1)
else:
import fcntl
fcntl.flock(handle.fileno(), fcntl.LOCK_UN)
@contextmanager
def _locked_file(path: Path, mode: str) -> Iterator[TextIO]:
path.parent.mkdir(parents=True, exist_ok=True)
local_lock = _path_lock(path)
with local_lock:
with path.open(mode, encoding="utf-8") as handle:
_lock_file(handle)
try:
yield handle
finally:
_unlock_file(handle)
def default_usage_ledger_path(*, env: dict[str, str] | None = None) -> Path:
"""Resolve the default usage ledger path.
Order: ``LLM_CONNECT_USAGE_LEDGER`` → ``$XDG_DATA_HOME/llm-connect/usage.jsonl``
→ ``~/.local/share/llm-connect/usage.jsonl``.
"""
environ = env if env is not None else os.environ
explicit = environ.get(ENV_USAGE_LEDGER)
if explicit and str(explicit).strip():
return Path(str(explicit).strip()).expanduser()
xdg = environ.get("XDG_DATA_HOME")
if xdg and str(xdg).strip():
return Path(str(xdg).strip()).expanduser() / "llm-connect" / "usage.jsonl"
return Path.home() / ".local" / "share" / "llm-connect" / "usage.jsonl"
@dataclass(frozen=True)
class UsageEvent:
"""One recorded LLM call for spend accounting."""
provider: str
model_id: str
prompt_tokens: int
completion_tokens: int
total_tokens: int
cost_usd: float | None
cost_eur: float | None
cost_source: str
source: str
fx_source: str | None = None
recorded_at: datetime = field(default_factory=_utc_now)
tags: dict[str, Any] = field(default_factory=dict)
def __post_init__(self) -> None:
for name in ("provider", "model_id", "source", "cost_source"):
if not str(getattr(self, name)).strip():
raise ValueError(f"{name} must be a non-empty string")
for name in ("prompt_tokens", "completion_tokens", "total_tokens"):
_validate_non_negative_int(name, getattr(self, name))
if self.cost_usd is not None and (
not isinstance(self.cost_usd, (int, float)) or float(self.cost_usd) < 0
):
raise ValueError("cost_usd must be a non-negative number or None")
if self.cost_eur is not None and (
not isinstance(self.cost_eur, (int, float)) or float(self.cost_eur) < 0
):
raise ValueError("cost_eur must be a non-negative number or None")
object.__setattr__(self, "provider", str(self.provider).strip())
object.__setattr__(self, "model_id", str(self.model_id).strip())
object.__setattr__(self, "source", str(self.source).strip())
object.__setattr__(self, "cost_source", str(self.cost_source).strip())
object.__setattr__(
self,
"cost_usd",
None if self.cost_usd is None else float(self.cost_usd),
)
object.__setattr__(
self,
"cost_eur",
None if self.cost_eur is None else float(self.cost_eur),
)
object.__setattr__(self, "recorded_at", _normalise_datetime(self.recorded_at))
object.__setattr__(self, "tags", dict(self.tags or {}))
if self.fx_source is not None:
object.__setattr__(self, "fx_source", str(self.fx_source))
def to_dict(self) -> dict[str, Any]:
"""Convert to a JSON-serialisable dictionary."""
return {
"provider": self.provider,
"model_id": self.model_id,
"prompt_tokens": self.prompt_tokens,
"completion_tokens": self.completion_tokens,
"total_tokens": self.total_tokens,
"cost_usd": self.cost_usd,
"cost_eur": self.cost_eur,
"cost_source": self.cost_source,
"fx_source": self.fx_source,
"source": self.source,
"recorded_at": _serialise_datetime(self.recorded_at),
"tags": dict(self.tags),
}
@classmethod
def from_dict(cls, data: dict[str, Any]) -> "UsageEvent":
"""Create an event from a JSON-decoded dictionary."""
return cls(
provider=data["provider"],
model_id=data["model_id"],
prompt_tokens=int(data["prompt_tokens"]),
completion_tokens=int(data["completion_tokens"]),
total_tokens=int(data["total_tokens"]),
cost_usd=data.get("cost_usd"),
cost_eur=data.get("cost_eur"),
cost_source=data.get("cost_source") or "unknown",
fx_source=data.get("fx_source"),
source=data.get("source") or "library",
recorded_at=data.get("recorded_at", _utc_now()),
tags=data.get("tags") or {},
)
@dataclass(frozen=True)
class UsageSummary:
"""Aggregated tokens and cost over a time window."""
start: datetime
end: datetime
event_count: int
prompt_tokens: int
completion_tokens: int
total_tokens: int
cost_usd: float | None
cost_eur: float | None
unknown_cost_events: int = 0
def to_dict(self) -> dict[str, Any]:
return {
"start": _serialise_datetime(self.start),
"end": _serialise_datetime(self.end),
"event_count": self.event_count,
"prompt_tokens": self.prompt_tokens,
"completion_tokens": self.completion_tokens,
"total_tokens": self.total_tokens,
"cost_usd": self.cost_usd,
"cost_eur": self.cost_eur,
"unknown_cost_events": self.unknown_cost_events,
}
class UsageLedger:
"""Append-only JSONL store for :class:`UsageEvent` records."""
def __init__(self, path: str | Path):
self._path = Path(path)
@property
def path(self) -> Path:
"""Ledger file path."""
return self._path
def append(self, event: UsageEvent) -> None:
"""Append one event as a locked JSONL record."""
line = json.dumps(event.to_dict(), sort_keys=True, separators=(",", ":"))
with _locked_file(self._path, "a") as handle:
handle.write(line + "\n")
handle.flush()
os.fsync(handle.fileno())
def read_all(self) -> list[UsageEvent]:
"""Return all parseable events, skipping malformed lines."""
events, _ = self._read_with_malformed_count()
return events
def iter_events(self) -> Iterator[UsageEvent]:
"""Yield parseable events in file order."""
yield from self.read_all()
def sum_range(
self,
start: datetime,
end: datetime,
) -> UsageSummary:
"""Aggregate events in the half-open interval ``[start, end)``."""
start_utc = _normalise_datetime(start)
end_utc = _normalise_datetime(end)
if end_utc < start_utc:
raise ValueError("end must be >= start")
prompt_tokens = 0
completion_tokens = 0
total_tokens = 0
cost_usd_total = 0.0
cost_eur_total = 0.0
has_usd = False
has_eur = False
unknown_cost = 0
count = 0
for event in self.read_all():
if event.recorded_at < start_utc or event.recorded_at >= end_utc:
continue
count += 1
prompt_tokens += event.prompt_tokens
completion_tokens += event.completion_tokens
total_tokens += event.total_tokens
if event.cost_usd is None:
unknown_cost += 1
else:
has_usd = True
cost_usd_total += event.cost_usd
if event.cost_eur is not None:
has_eur = True
cost_eur_total += event.cost_eur
return UsageSummary(
start=start_utc,
end=end_utc,
event_count=count,
prompt_tokens=prompt_tokens,
completion_tokens=completion_tokens,
total_tokens=total_tokens,
cost_usd=cost_usd_total if has_usd else (0.0 if count == 0 else None),
cost_eur=cost_eur_total if has_eur else (0.0 if count == 0 else None),
unknown_cost_events=unknown_cost,
)
def _read_with_malformed_count(self) -> tuple[list[UsageEvent], int]:
if not self._path.is_file():
return [], 0
events: list[UsageEvent] = []
malformed = 0
with _locked_file(self._path, "r") as handle:
for line in handle:
if not line.strip():
continue
try:
events.append(UsageEvent.from_dict(json.loads(line)))
except (json.JSONDecodeError, KeyError, TypeError, ValueError):
malformed += 1
return events, malformed
def week_window(
which: str = "current",
*,
now: datetime | None = None,
tz: str | None = None,
) -> tuple[datetime, datetime]:
"""Return ``(start, end)`` for a Monday-based spend window.
* ``current`` — Monday 00:00 local → *now*
* ``last`` — previous Monday 00:00 → this Monday 00:00 (full Mon–Sun)
Endpoints are timezone-aware in the requested local zone (default
Europe/Berlin). Comparisons in the ledger normalise to UTC.
"""
which_normalised = str(which).strip().lower()
if which_normalised not in {"current", "last"}:
raise ValueError("which must be 'current' or 'last'")
zone_name = (tz or os.environ.get(ENV_TZ) or DEFAULT_TZ).strip() or DEFAULT_TZ
zone = ZoneInfo(zone_name)
reference = now or datetime.now(zone)
if reference.tzinfo is None:
reference = reference.replace(tzinfo=zone)
else:
reference = reference.astimezone(zone)
start_of_this_week = reference.replace(
hour=0, minute=0, second=0, microsecond=0
) - timedelta(days=reference.weekday())
if which_normalised == "current":
return start_of_this_week, reference
end = start_of_this_week
start = end - timedelta(days=7)
return start, end
def event_from_response(
response: LLMResponse,
*,
provider: str,
source: str,
registry: ModelRateRegistry | None = None,
fx: FxRate | float | None = None,
tags: dict[str, Any] | None = None,
recorded_at: datetime | None = None,
) -> UsageEvent:
"""Build a :class:`UsageEvent` from an :class:`LLMResponse` and cost model."""
usage = response.usage or {}
prompt_tokens = int(usage.get("prompt_tokens") or 0)
completion_tokens = int(usage.get("completion_tokens") or 0)
total_tokens = int(usage.get("total_tokens") or (prompt_tokens + completion_tokens))
model_id = response.model or "unknown"
estimate = estimate_cost(
model_id,
prompt_tokens,
completion_tokens,
registry=registry,
fx=fx,
)
return UsageEvent(
provider=provider,
model_id=model_id,
prompt_tokens=prompt_tokens,
completion_tokens=completion_tokens,
total_tokens=total_tokens,
cost_usd=estimate.cost_usd,
cost_eur=estimate.cost_eur,
cost_source=estimate.cost_source,
fx_source=estimate.fx_source,
source=source,
recorded_at=recorded_at or _utc_now(),
tags=tags or {},
)
def maybe_record_usage(
response: LLMResponse,
*,
provider: str,
source: str,
tags: dict[str, Any] | None = None,
env: dict[str, str] | None = None,
registry: ModelRateRegistry | None = None,
fx: FxRate | float | None = None,
) -> UsageEvent | None:
"""Record usage when ``LLM_CONNECT_USAGE_LEDGER`` is set; otherwise no-op.
Returns the recorded event, or ``None`` when recording is disabled.
"""
if _AUTO_RECORD_DISABLED.get():
return None
environ = env if env is not None else os.environ
path_value = environ.get(ENV_USAGE_LEDGER)
if not path_value or not str(path_value).strip():
return None
event = event_from_response(
response,
provider=provider,
source=source,
registry=registry,
fx=fx,
tags=tags,
)
UsageLedger(Path(str(path_value).strip()).expanduser()).append(event)
return event
def format_cost_footer(
estimate: CostEstimate,
*,
prompt_tokens: int,
completion_tokens: int,
total_tokens: int,
model_id: str,
) -> str:
"""Human-readable one-line cost summary for CLI footers."""
tokens_part = (
f"tokens: in={prompt_tokens} out={completion_tokens} total={total_tokens}"
)
if estimate.cost_usd is None:
usd_part = "usd=unknown"
else:
usd_part = f"usd={estimate.cost_usd:.6f}"
if estimate.cost_eur is None:
eur_part = "eur=unknown"
else:
eur_part = f"eur={estimate.cost_eur:.6f}"
return f"cost[{model_id}]: {tokens_part} · {eur_part} · {usd_part} · source={estimate.cost_source}"

116
tests/test_cli.py
View file

@ -1,8 +1,10 @@
import json
from datetime import datetime, timezone
from zoneinfo import ZoneInfo
from llm_connect.cli import main
from llm_connect.quality import QualityLedger, QualityObservation
from llm_connect.usage import UsageEvent, UsageLedger
def test_rates_show_json_outputs_default_registry(capsys):
@ -11,6 +13,7 @@ def test_rates_show_json_outputs_default_registry(capsys):
payload = json.loads(capsys.readouterr().out)
assert payload["openai/gpt-4o-mini"]["prompt_per_1k"] == 0.00015
assert payload["moonshotai/kimi-k3"]["prompt_per_1k"] == 0.003
def test_classes_show_lists_builtins(capsys):
@ -52,3 +55,116 @@ def test_classes_fit_reads_quality_ledger(tmp_path, capsys):
payload = json.loads(capsys.readouterr().out)
assert payload["entity-extraction"]["params"]["tokens_per_entity"] == 70
def test_run_with_mock_reports_cost_and_writes_ledger(tmp_path, capsys):
ledger = tmp_path / "usage.jsonl"
assert (
main(
[
"run",
"hello world",
"--provider",
"mock",
"--model",
"moonshotai/kimi-k3",
"--ledger",
str(ledger),
"--eur-per-usd",
"1.0",
"--json",
]
)
== 0
)
payload = json.loads(capsys.readouterr().out)
assert payload["content"]
assert payload["usage"]["total_tokens"] > 0
assert payload["cost_usd"] is not None
assert payload["cost_eur"] is not None
events = UsageLedger(ledger).read_all()
assert len(events) == 1
assert events[0].source == "cli"
assert events[0].model_id == "moonshotai/kimi-k3"
def test_cost_estimate_cli(capsys):
assert (
main(
[
"cost",
"estimate",
"--model",
"moonshotai/kimi-k3",
"--prompt-tokens",
"1000",
"--completion-tokens",
"1000",
"--eur-per-usd",
"1",
"--json",
]
)
== 0
)
payload = json.loads(capsys.readouterr().out)
assert payload["cost_usd"] == 0.018
assert payload["cost_eur"] == 0.018
def test_spend_week_current_and_last(tmp_path, capsys, monkeypatch):
ledger_path = tmp_path / "usage.jsonl"
ledger = UsageLedger(ledger_path)
berlin = ZoneInfo("Europe/Berlin")
current_start = datetime(2026, 8, 3, 0, 0, tzinfo=berlin)
current_end = datetime(2026, 8, 5, 12, 0, tzinfo=berlin)
last_start = datetime(2026, 7, 27, 0, 0, tzinfo=berlin)
last_end = current_start
def _fake_week_window(which="current", *, now=None, tz=None):
if which == "last":
return last_start, last_end
return current_start, current_end
monkeypatch.setattr("llm_connect.cli.week_window", _fake_week_window)
ledger.append(
UsageEvent(
provider="mock",
model_id="moonshotai/kimi-k3",
prompt_tokens=100,
completion_tokens=20,
total_tokens=120,
cost_usd=0.01,
cost_eur=0.009,
cost_source="rate_table:moonshotai/kimi-k3",
source="cli",
recorded_at=datetime(2026, 8, 4, 9, 0, tzinfo=berlin),
)
)
ledger.append(
UsageEvent(
provider="mock",
model_id="moonshotai/kimi-k3",
prompt_tokens=50,
completion_tokens=10,
total_tokens=60,
cost_usd=0.005,
cost_eur=0.0045,
cost_source="rate_table:moonshotai/kimi-k3",
source="cli",
recorded_at=datetime(2026, 7, 29, 9, 0, tzinfo=berlin), # previous week
)
)
assert main(["spend", "week", "--ledger", str(ledger_path), "--json"]) == 0
current = json.loads(capsys.readouterr().out)
assert current["event_count"] == 1
assert current["total_tokens"] == 120
assert current["cost_eur"] == 0.009
assert main(["spend", "week", "--last", "--ledger", str(ledger_path), "--json"]) == 0
last = json.loads(capsys.readouterr().out)
assert last["event_count"] == 1
assert last["total_tokens"] == 60

29
tests/test_costs.py
View file

@ -10,12 +10,16 @@ def test_known_model_cost_matches_lefevre_smoke_budget():
assert estimate.cost_source == "rate_table:openai/gpt-4o-mini"
assert estimate.cost_usd == pytest.approx(0.0087)
assert estimate.cost_usd == pytest.approx(0.009, rel=0.2)
assert estimate.cost_eur is not None
assert estimate.fx_source is not None
def test_unknown_model_returns_unknown_without_zeroing_cost():
estimate = estimate_cost("unknown/model", 100, 50)
assert estimate == CostEstimate(cost_usd=None, cost_source="unknown")
assert estimate.cost_usd is None
assert estimate.cost_eur is None
assert estimate.cost_source == "unknown"
def test_registry_override_controls_estimate():
@ -29,21 +33,40 @@ def test_registry_override_controls_estimate():
}
)
estimate = estimate_cost("vendor/model", 1_000, 500, registry=registry)
estimate = estimate_cost("vendor/model", 1_000, 500, registry=registry, fx=1.0)
assert estimate.cost_usd == pytest.approx(2.0)
assert estimate.prompt_cost_usd == pytest.approx(1.0)
assert estimate.completion_cost_usd == pytest.approx(1.0)
assert estimate.cost_eur == pytest.approx(2.0)
def test_zero_tokens_are_valid_and_cost_zero_for_known_model():
estimate = CostModel().estimate_cost("openai/gpt-4o-mini", 0, 0)
estimate = CostModel(fx=1.0).estimate_cost("openai/gpt-4o-mini", 0, 0)
assert estimate.cost_usd == 0
assert estimate.prompt_cost_usd == 0
assert estimate.completion_cost_usd == 0
assert estimate.cost_eur == 0
def test_negative_tokens_are_rejected():
with pytest.raises(ValueError, match="prompt_tokens"):
estimate_cost("openai/gpt-4o-mini", -1, 0)
def test_kimi_k3_rate_and_eur_conversion():
estimate = estimate_cost("moonshotai/kimi-k3", 1_000, 1_000, fx=0.92)
# $0.003 + $0.015 = $0.018; ×0.92 = €0.01656
assert estimate.cost_usd == pytest.approx(0.018)
assert estimate.cost_eur == pytest.approx(0.01656)
assert estimate.cost_source == "rate_table:moonshotai/kimi-k3"
def test_apply_fx_false_skips_eur():
estimate = estimate_cost("moonshotai/kimi-k3", 100, 0, apply_fx=False)
assert estimate.cost_usd is not None
assert estimate.cost_eur is None
assert estimate.fx_source is None

40
tests/test_fx.py Normal file
View file

@ -0,0 +1,40 @@
import pytest
from llm_connect.fx import DEFAULT_EUR_PER_USD, FxRate, resolve_fx_rate
def test_resolve_fx_prefers_explicit_rate():
rate = resolve_fx_rate(0.85)
assert rate is not None
assert rate.eur_per_usd == pytest.approx(0.85)
assert rate.source == "explicit"
assert rate.usd_to_eur(10.0) == pytest.approx(8.5)
def test_resolve_fx_reads_env(monkeypatch):
monkeypatch.setenv("LLM_CONNECT_EUR_PER_USD", "0.9")
rate = resolve_fx_rate()
assert rate is not None
assert rate.eur_per_usd == pytest.approx(0.9)
assert rate.source == "env:LLM_CONNECT_EUR_PER_USD"
def test_resolve_fx_falls_back_to_snapshot():
rate = resolve_fx_rate(env={})
assert rate is not None
assert rate.eur_per_usd == pytest.approx(DEFAULT_EUR_PER_USD)
assert rate.source.startswith("snapshot:")
def test_invalid_env_rate_raises():
with pytest.raises(ValueError, match="LLM_CONNECT_EUR_PER_USD"):
resolve_fx_rate(env={"LLM_CONNECT_EUR_PER_USD": "0"})
def test_fx_rate_rejects_non_positive():
with pytest.raises(ValueError, match="eur_per_usd"):
FxRate(eur_per_usd=-1, source="x")

18
tests/test_package_exports.py
View file

@ -61,3 +61,21 @@ def test_wp_0006_profile_primitives_are_exported_from_package_root():
for name in expected_names:
assert hasattr(llm_connect, name)
assert name in llm_connect.__all__
def test_wp_0007_spend_primitives_are_exported_from_package_root():
expected_names = [
"FxRate",
"resolve_fx_rate",
"UsageEvent",
"UsageLedger",
"UsageSummary",
"default_usage_ledger_path",
"event_from_response",
"maybe_record_usage",
"week_window",
]
for name in expected_names:
assert hasattr(llm_connect, name)
assert name in llm_connect.__all__

5
tests/test_rates.py
View file

@ -7,9 +7,12 @@ def test_default_registry_contains_openrouter_seed_models():
registry = ModelRateRegistry.default()
rates = registry.all()
assert len(rates) >= 9
assert len(rates) >= 10
assert rates["openai/gpt-4o-mini"].captured_at == "2026-05-17"
assert rates["openai/gpt-4o-mini"].source_url == "https://openrouter.ai/models"
assert rates["moonshotai/kimi-k3"].prompt_per_1k == 0.003
assert rates["moonshotai/kimi-k3"].completion_per_1k == 0.015
assert rates["moonshotai/kimi-k3"].captured_at == "2026-08-03"
def test_from_yaml_loads_package_shape(tmp_path):

127
tests/test_usage.py Normal file
View file

@ -0,0 +1,127 @@
from datetime import datetime
from zoneinfo import ZoneInfo
import pytest
from llm_connect.models import LLMResponse
from llm_connect.usage import (
UsageEvent,
UsageLedger,
default_usage_ledger_path,
event_from_response,
maybe_record_usage,
suppress_auto_usage_record,
week_window,
)
def test_default_ledger_path_env_and_xdg():
assert default_usage_ledger_path(env={"LLM_CONNECT_USAGE_LEDGER": "/tmp/u.jsonl"}) == (
__import__("pathlib").Path("/tmp/u.jsonl")
)
path = default_usage_ledger_path(env={"XDG_DATA_HOME": "/tmp/xdg"})
assert path.as_posix().endswith("llm-connect/usage.jsonl")
assert "xdg" in path.as_posix()
def test_usage_ledger_append_and_sum_range(tmp_path):
ledger = UsageLedger(tmp_path / "usage.jsonl")
berlin = ZoneInfo("Europe/Berlin")
monday = datetime(2026, 8, 3, 10, 0, tzinfo=berlin) # Monday
sunday = datetime(2026, 8, 2, 12, 0, tzinfo=berlin) # previous Sunday
ledger.append(
UsageEvent(
provider="openrouter",
model_id="moonshotai/kimi-k3",
prompt_tokens=100,
completion_tokens=50,
total_tokens=150,
cost_usd=0.001,
cost_eur=0.00092,
cost_source="rate_table:moonshotai/kimi-k3",
fx_source="snapshot:2026-08-03",
source="cli",
recorded_at=monday,
)
)
ledger.append(
UsageEvent(
provider="openrouter",
model_id="moonshotai/kimi-k3",
prompt_tokens=10,
completion_tokens=5,
total_tokens=15,
cost_usd=0.0001,
cost_eur=0.000092,
cost_source="rate_table:moonshotai/kimi-k3",
source="cli",
recorded_at=sunday,
)
)
start, end = week_window(
"current",
now=datetime(2026, 8, 5, 18, 0, tzinfo=berlin),
tz="Europe/Berlin",
)
summary = ledger.sum_range(start, end)
assert summary.event_count == 1
assert summary.prompt_tokens == 100
assert summary.total_tokens == 150
assert summary.cost_eur == pytest.approx(0.00092)
last_start, last_end = week_window(
"last",
now=datetime(2026, 8, 5, 18, 0, tzinfo=berlin),
tz="Europe/Berlin",
)
last_summary = ledger.sum_range(last_start, last_end)
assert last_summary.event_count == 1
assert last_summary.total_tokens == 15
def test_week_window_current_and_last():
berlin = ZoneInfo("Europe/Berlin")
now = datetime(2026, 8, 5, 15, 30, tzinfo=berlin) # Wednesday
start, end = week_window("current", now=now, tz="Europe/Berlin")
assert start.weekday() == 0
assert start.hour == 0
assert end == now
last_start, last_end = week_window("last", now=now, tz="Europe/Berlin")
assert (last_end - last_start).days == 7
assert last_end == start
def test_event_from_response_estimates_kimi_cost():
response = LLMResponse(
content="hi",
model="moonshotai/kimi-k3",
usage={"prompt_tokens": 1000, "completion_tokens": 1000, "total_tokens": 2000},
)
event = event_from_response(response, provider="openrouter", source="cli", fx=1.0)
assert event.cost_usd == pytest.approx(0.018)
assert event.cost_eur == pytest.approx(0.018)
assert event.cost_source == "rate_table:moonshotai/kimi-k3"
def test_maybe_record_usage_opt_in(tmp_path, monkeypatch):
ledger = tmp_path / "usage.jsonl"
monkeypatch.setenv("LLM_CONNECT_USAGE_LEDGER", str(ledger))
response = LLMResponse(
content="x",
model="moonshotai/kimi-k3",
usage={"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15},
metadata={"provider": "openrouter"},
)
event = maybe_record_usage(response, provider="openrouter", source="library")
assert event is not None
assert ledger.is_file()
assert UsageLedger(ledger).read_all()[0].source == "library"
with suppress_auto_usage_record():
assert maybe_record_usage(response, provider="openrouter", source="library") is None

43
workplans/LLM-WP-0007-kimi-k3-default-and-spend-reporting.md
View file

@ -4,7 +4,7 @@ type: workplan
title: "Kimi K3 default via OpenRouter + EUR spend reporting"
domain: agents
repo: llm-connect
status: ready
status: finished
owner: codex
topic_slug: kimi-k3-default-spend-reporting
planning_priority: high
@ -20,7 +20,7 @@ state_hub_workstream_id: "a93d226a-f5c4-4656-b584-a783840b10fe"
# LLM-WP-0007 — Kimi K3 default via OpenRouter + EUR spend reporting
**status:** ready
**status:** finished
**owner:** codex
## Purpose
@ -173,7 +173,7 @@ print clear `unknown` when not.
```task
id: LLM-WP-0007-T01
status: todo
status: done
priority: high
state_hub_task_id: "af12101d-bcd5-4d6f-9507-c81de79a0685"
```
@ -186,7 +186,7 @@ or README providers table. Gate for T02.
```task
id: LLM-WP-0007-T02
status: todo
status: done
priority: high
state_hub_task_id: "ca836db8-9041-42e5-87ce-7c3ace1298a3"
```
@ -199,7 +199,7 @@ leave default unchanged and record the blocker in the task notes.
```task
id: LLM-WP-0007-T03
status: todo
status: done
priority: high
state_hub_task_id: "f379d09b-ea03-44f5-82fa-171da4681cef"
```
@ -211,7 +211,7 @@ model still yields `cost_usd=None` / `cost_eur=None`.
```task
id: LLM-WP-0007-T04
status: todo
status: done
priority: high
state_hub_task_id: "cadf1b0c-fdea-4446-8d01-964b2706540d"
```
@ -224,7 +224,7 @@ effort flock), and range aggregation.
```task
id: LLM-WP-0007-T05
status: todo
status: done
priority: high
state_hub_task_id: "8956fd74-cad4-435c-ae66-6b19088691e4"
```
@ -236,7 +236,7 @@ estimation from response usage + rate table + FX. Tests with mock adapter.
```task
id: LLM-WP-0007-T06
status: todo
status: done
priority: high
state_hub_task_id: "ac5e4322-cd00-4034-b162-a0e1c481180e"
```
@ -248,7 +248,7 @@ fixed timezone.
```task
id: LLM-WP-0007-T07
status: todo
status: done
priority: medium
state_hub_task_id: "a05761b9-d91c-4b6d-9e90-296aadadaf16"
```
@ -260,7 +260,7 @@ opt-in so library consumers are not surprised by home-dir writes.
```task
id: LLM-WP-0007-T08
status: todo
status: done
priority: medium
state_hub_task_id: "94ce6aee-2835-46c0-a3e6-3d0345e7715f"
```
@ -272,14 +272,14 @@ as needed. Smoke script or Makefile target for optional live Kimi check.
## Acceptance
- [ ] `moonshotai/kimi-k3` is in the default rate registry with documented USD rates
- [ ] After successful smoke, OpenRouter/library default model is `moonshotai/kimi-k3`
- [ ] `llm-connect run "..."` prints token counts and € cost (or explicit unknown)
- [ ] Successful CLI runs append to the usage ledger
- [ ] `llm-connect spend week` and `spend week --last` show tokens + € for the
- [x] `moonshotai/kimi-k3` is in the default rate registry with documented USD rates
- [x] After successful smoke, OpenRouter/library default model is `moonshotai/kimi-k3`
- [x] `llm-connect run "..."` prints token counts and € cost (or explicit unknown)
- [x] Successful CLI runs append to the usage ledger
- [x] `llm-connect spend week` and `spend week --last` show tokens + € for the
correct Mon-based windows
- [ ] Default CI tests pass offline (mock adapter, fixed FX, no live OpenRouter)
- [ ] Missing rate or FX never reported as €0.00 without an `unknown` marker
- [x] Default CI tests pass offline (mock adapter, fixed FX, no live OpenRouter)
- [x] Missing rate or FX never reported as €0.00 without an `unknown` marker
## Risks / notes
@ -298,3 +298,12 @@ as needed. Smoke script or Makefile target for optional live Kimi check.
T01 → T02 (gated) in parallel with T03 → T04 → T05 → T06 → T07 → T08.
T03/T04 can start before T01; T02 must wait for T01 smoke outcome.
## Implementation notes (2026-08-03)
- Live OpenRouter smoke: `moonshotai/kimi-k3` returned content "ok" with usage tokens (Together provider). Note: short `max_tokens` can yield empty `content` while spending tokens on reasoning.
- Default OpenRouter / LLMConfig model flipped to `moonshotai/kimi-k3`.
- Added `fx.py`, EUR fields on `CostEstimate`, `usage.py` ledger, CLI `run` / `cost estimate` / `spend week`.
- Opt-in library/server recording via `LLM_CONNECT_USAGE_LEDGER`; CLI suppresses double-record.
- Full unit suite: 224 passed offline.