praxis/server/cost.py

"""Cost logging — per-session cost derivation (REQ-NFR-COST-01, D-012, TASK-04-04).

Counts LLM input/output tokens (gemma4 + deepseek-v4-flash), Deepgram audio
minutes, Cartesia/Piper characters; derives an estimated cost in cents using
cost_rates.yaml. No enforced ceiling (D-012 — pilot). The derived cost +
breakdown are stored in sessions.cost_estimated_cents / cost_breakdown_json.

v0.1 logged costs are NOT representative of at-scale per-learner cost (G-005):
Ollama tier-based pricing + Canada cloud + low volume = the most expensive
configuration. The $3/learner target requires self-hosted gemma4:e4b + Piper
(post-pilot). The logging infrastructure is the v0.1 contribution.
"""

from __future__ import annotations

from dataclasses import dataclass, field
from pathlib import Path
from typing import Any

import yaml

_DEFAULT_RATES_PATH = Path(__file__).resolve().parent.parent / "scenarios" / "cost_rates.yaml"


@dataclass
class CostBreakdown:
    """Per-session cost inputs + derived cents."""

    llm_input_tokens: int = 0
    llm_output_tokens: int = 0
    deepgram_audio_minutes: float = 0.0
    tts_characters: int = 0
    debrief_input_tokens: int = 0
    debrief_output_tokens: int = 0
    rates: dict[str, float] = field(default_factory=dict)
    derived_cents: int = 0

    def as_dict(self) -> dict[str, Any]:
        return {
            "llm_input_tokens": self.llm_input_tokens,
            "llm_output_tokens": self.llm_output_tokens,
            "deepgram_audio_minutes": round(self.deepgram_audio_minutes, 3),
            "tts_characters": self.tts_characters,
            "debrief_input_tokens": self.debrief_input_tokens,
            "debrief_output_tokens": self.debrief_output_tokens,
            "rates": self.rates,
            "derived_cents": self.derived_cents,
        }


def load_rates(path: Path | None = None) -> dict[str, float]:
    """Load cost rates from cost_rates.yaml (or defaults if absent)."""
    p = path or _DEFAULT_RATES_PATH
    if p.exists():
        with p.open("r", encoding="utf-8") as f:
            return yaml.safe_load(f) or {}
    # Defaults — vendor-list prices, per-unit (pilot estimates, G-005).
    return {
        "gemma4_cloud_per_1k_tokens_cents": 0.5,        # Ollama tier (pro plan amortized)
        "deepseek_v4_flash_per_1k_tokens_cents": 1.0,   # Ollama tier
        "deepgram_per_audio_minute_cents": 0.43,        # $0.0043/min
        "cartesia_per_1k_chars_cents": 3.0,             # per-char pricing
        "piper_per_1k_chars_cents": 0.0,                # self-hosted, $0
    }


def derive_cost(
    llm_input_tokens: int = 0,
    llm_output_tokens: int = 0,
    deepgram_audio_minutes: float = 0.0,
    tts_characters: int = 0,
    debrief_input_tokens: int = 0,
    debrief_output_tokens: int = 0,
    tts_provider: str = "cartesia",
    rates: dict[str, float] | None = None,
) -> CostBreakdown:
    """Derive the per-session cost in cents from the usage inputs + rates."""
    r = rates or load_rates()

    # LLM role-play (gemma4:cloud).
    rp_tokens = llm_input_tokens + llm_output_tokens
    rp_cents = (rp_tokens / 1000.0) * r.get("gemma4_cloud_per_1k_tokens_cents", 0.5)

    # Debrief (deepseek-v4-flash:cloud).
    db_tokens = debrief_input_tokens + debrief_output_tokens
    db_cents = (db_tokens / 1000.0) * r.get("deepseek_v4_flash_per_1k_tokens_cents", 1.0)

    # ASR (Deepgram).
    asr_cents = deepgram_audio_minutes * r.get("deepgram_per_audio_minute_cents", 0.43)

    # TTS (Cartesia or Piper).
    tts_rate_key = (
        "piper_per_1k_chars_cents" if tts_provider == "piper"
        else "cartesia_per_1k_chars_cents"
    )
    tts_cents = (tts_characters / 1000.0) * r.get(tts_rate_key, 3.0)

    total = int(round(rp_cents + db_cents + asr_cents + tts_cents))
    return CostBreakdown(
        llm_input_tokens=llm_input_tokens,
        llm_output_tokens=llm_output_tokens,
        deepgram_audio_minutes=deepgram_audio_minutes,
        tts_characters=tts_characters,
        debrief_input_tokens=debrief_input_tokens,
        debrief_output_tokens=debrief_output_tokens,
        rates=r,
        derived_cents=total,
    )


def derive_assist_turn_cost(
    llm_input_tokens: int = 0,
    llm_output_tokens: int = 0,
    tts_characters: int = 0,
    tts_provider: str = "piper",
    rates: dict[str, float] | None = None,
) -> CostBreakdown:
    """Derive the per-assist-turn cost in cents (TASK-11-01, REQ-IDEATE-07).

    An assist turn is a short coaching exchange — a single gemma4:cloud LLM
    call + Piper TTS (D-065 — Piper is the assist default). No debrief tokens
    (assist has no debrief — D-063) + no Deepgram audio minutes (the assist
    turn's ASR is accounted in the shift's Deepgram minutes, not per-turn —
    the per-turn cost is the LLM + TTS only).

    Uses the same load_rates() + the same CostBreakdown dataclass as
    derive_cost(). The per-turn cost is logged via
    AssistSession.add_assist_turn_cost() + aggregated at shift-end as
    assist_cost_cents in the session_outcome (for the C-3 budget check —
    TASK-11-02).

    The existing derive_cost() is unchanged (practice sessions keep their
    cost logging — backward compat).
    """
    r = rates or load_rates()

    # LLM (gemma4:cloud) — the assist coaching call.
    llm_tokens = llm_input_tokens + llm_output_tokens
    llm_cents = (llm_tokens / 1000.0) * r.get("gemma4_cloud_per_1k_tokens_cents", 0.5)

    # TTS (Piper default for assist — D-065; Cartesia fallback).
    tts_rate_key = (
        "piper_per_1k_chars_cents" if tts_provider == "piper"
        else "cartesia_per_1k_chars_cents"
    )
    tts_cents = (tts_characters / 1000.0) * r.get(tts_rate_key, 3.0)

    total = int(round(llm_cents + tts_cents))
    return CostBreakdown(
        llm_input_tokens=llm_input_tokens,
        llm_output_tokens=llm_output_tokens,
        deepgram_audio_minutes=0.0,  # assist ASR accounted at shift level
        tts_characters=tts_characters,
        debrief_input_tokens=0,       # assist has no debrief (D-063)
        debrief_output_tokens=0,
        rates=r,
        derived_cents=total,
    )


__all__ = ["CostBreakdown", "derive_cost", "derive_assist_turn_cost", "load_rates"]