symfonic.core.observability.pricing

pricing

Model pricing registry for cost calculation.

calculate_cost

calculate_cost(
    model: str,
    input_tokens: int,
    output_tokens: int,
    cached_tokens: int = 0,
    cache_creation_tokens: int = 0,
    reasoning_tokens: int = 0,
    cache_ttl: str | None = None,
) -> float

Calculate USD cost for a single LLM call.

Returns 0.0 if the model is not in the registry. Cached tokens are billed at cache_read rate; the remainder of input tokens are billed at the standard input rate.

v7.15.0: reasoning_tokens is observability-only by default -- the pricing layer only adds a separate reasoning line when the model's pricing row explicitly opts in via a "reasoning" key. See :func:_compute_cost docstring for the no-double-charge invariant on Opus-4-x.

v7.20.0 (architect verdict 2026-06-05 §5): cache_ttl lets Anthropic 1-hour ephemeral caches bill at the row's cache_write_1h rate instead of the 5-minute cache_write default. See :func:_compute_cost for the fallback contract.

For callers that need to distinguish "known model, zero cost" from "unknown model, fallback to zero", use :func:calculate_cost_with_meta -- it returns the same float plus a pricing_unknown: bool discriminator. This function preserves the historical float-return contract for BudgetTracker, MetricsHandler, and other callers that sum / compare the value.

Source code in src/symfonic/core/observability/pricing.py

def calculate_cost(
    model: str,
    input_tokens: int,
    output_tokens: int,
    cached_tokens: int = 0,
    cache_creation_tokens: int = 0,
    reasoning_tokens: int = 0,
    cache_ttl: str | None = None,
) -> float:
    """Calculate USD cost for a single LLM call.

    Returns 0.0 if the model is not in the registry.
    Cached tokens are billed at cache_read rate; the remainder of input
    tokens are billed at the standard input rate.

    v7.15.0: ``reasoning_tokens`` is observability-only by default --
    the pricing layer only adds a separate reasoning line when the
    model's pricing row explicitly opts in via a ``"reasoning"`` key.
    See :func:`_compute_cost` docstring for the no-double-charge
    invariant on Opus-4-x.

    v7.20.0 (architect verdict 2026-06-05 §5): ``cache_ttl`` lets
    Anthropic 1-hour ephemeral caches bill at the row's
    ``cache_write_1h`` rate instead of the 5-minute ``cache_write``
    default.  See :func:`_compute_cost` for the fallback contract.

    For callers that need to distinguish "known model, zero cost"
    from "unknown model, fallback to zero", use
    :func:`calculate_cost_with_meta` -- it returns the same float plus
    a ``pricing_unknown: bool`` discriminator.  This function preserves
    the historical ``float``-return contract for ``BudgetTracker``,
    ``MetricsHandler``, and other callers that sum / compare the value.
    """
    if not model:
        return 0.0

    pricing = _resolve_pricing(model)
    if pricing is None:
        _warn_unknown_once(model)
        return 0.0

    return _compute_cost(
        pricing,
        input_tokens,
        output_tokens,
        cached_tokens,
        cache_creation_tokens,
        reasoning_tokens=reasoning_tokens,
        cache_ttl=cache_ttl,
    )

calculate_cost_with_meta

calculate_cost_with_meta(
    model: str,
    input_tokens: int,
    output_tokens: int,
    cached_tokens: int = 0,
    cache_creation_tokens: int = 0,
    reasoning_tokens: int = 0,
    cache_ttl: str | None = None,
) -> tuple[float, bool]

Same as :func:calculate_cost, but also returns a pricing_unknown flag.

Returns (cost_usd, pricing_unknown) where:

• pricing_unknown=True means the model name had no entry in the registry (empty model string OR no exact/prefix match). cost_usd is then 0.0 and a one-time WARNING was logged (same behaviour as the legacy calculate_cost path).
• pricing_unknown=False means the registry resolved a pricing row; cost_usd is the computed value (can still be 0.0 for zero-token calls or zero-rate entries).

Introduced in v7.4.5 (Jarvio Ask 6) so :class:TokenUsage and :class:LLMEndEvent can ship a programmatic "unknown pricing" channel distinct from the "known model, $0" case (e.g. zero-token continuation, cache-only turn at a $0 cache rate).

v7.15.0: reasoning_tokens opt-in kwarg. See :func:calculate_cost for the no-double-charge invariant.

Source code in src/symfonic/core/observability/pricing.py

def calculate_cost_with_meta(
    model: str,
    input_tokens: int,
    output_tokens: int,
    cached_tokens: int = 0,
    cache_creation_tokens: int = 0,
    reasoning_tokens: int = 0,
    cache_ttl: str | None = None,
) -> tuple[float, bool]:
    """Same as :func:`calculate_cost`, but also returns a ``pricing_unknown`` flag.

    Returns ``(cost_usd, pricing_unknown)`` where:

    * ``pricing_unknown=True`` means the model name had no entry in the
      registry (empty model string OR no exact/prefix match).  ``cost_usd``
      is then ``0.0`` and a one-time WARNING was logged (same behaviour as
      the legacy ``calculate_cost`` path).
    * ``pricing_unknown=False`` means the registry resolved a pricing row;
      ``cost_usd`` is the computed value (can still be ``0.0`` for
      zero-token calls or zero-rate entries).

    Introduced in v7.4.5 (Jarvio Ask 6) so :class:`TokenUsage` and
    :class:`LLMEndEvent` can ship a programmatic "unknown pricing" channel
    distinct from the "known model, $0" case (e.g. zero-token continuation,
    cache-only turn at a $0 cache rate).

    v7.15.0: ``reasoning_tokens`` opt-in kwarg.  See
    :func:`calculate_cost` for the no-double-charge invariant.
    """
    if not model:
        # Empty model is the "no model" no-op; do NOT warn and do NOT
        # flag as unknown -- callers (e.g. TokenUsage.from_dict() on an
        # empty payload) should not see false-positive unknown signals.
        return 0.0, False

    pricing = _resolve_pricing(model)
    if pricing is None:
        _warn_unknown_once(model)
        return 0.0, True

    cost = _compute_cost(
        pricing,
        input_tokens,
        output_tokens,
        cached_tokens,
        cache_creation_tokens,
        reasoning_tokens=reasoning_tokens,
        cache_ttl=cache_ttl,
    )
    return cost, False

clear_pricing_overrides

clear_pricing_overrides() -> None

Clear all adopter-supplied pricing overrides.

Test isolation helper -- production callers should re-register via register_pricing_overrides({}) instead so the intent is explicit.

Source code in src/symfonic/core/observability/pricing.py

def clear_pricing_overrides() -> None:
    """Clear all adopter-supplied pricing overrides.

    Test isolation helper -- production callers should re-register via
    ``register_pricing_overrides({})`` instead so the intent is explicit.
    """
    global _PRICING_OVERRIDES, _override_replace_warned
    _PRICING_OVERRIDES = {}
    _override_replace_warned = False

emit_overrides_audit_log

emit_overrides_audit_log(
    overrides: dict[str, dict[str, float]],
) -> None

Emit a single INFO line listing the loaded overrides at agent init.

v7.20.0 (architect verdict 2026-06-05, pricing-registry hardening §4.a): called by SymfonicAgent.__init__ after override registration so operators see what the deployment installed.

CRITICAL SECURITY INVARIANT: the line carries model IDs ONLY, NEVER rate values. Rate values can be commercially-sensitive (e.g. a negotiated $0.40/M Anthropic enterprise rate); logging them surfaces confidential data into observability pipelines where they get indexed by CloudWatch / Datadog / etc. and become accessible to anyone with log-read permission.

Empty mapping is a silent no-op -- adopters who set no overrides should see zero log noise.

Source code in src/symfonic/core/observability/pricing.py

def emit_overrides_audit_log(overrides: dict[str, dict[str, float]]) -> None:
    """Emit a single INFO line listing the loaded overrides at agent init.

    v7.20.0 (architect verdict 2026-06-05, pricing-registry hardening
    §4.a): called by ``SymfonicAgent.__init__`` after override
    registration so operators see what the deployment installed.

    CRITICAL SECURITY INVARIANT: the line carries model IDs ONLY,
    NEVER rate values. Rate values can be commercially-sensitive
    (e.g. a negotiated $0.40/M Anthropic enterprise rate); logging
    them surfaces confidential data into observability pipelines
    where they get indexed by CloudWatch / Datadog / etc. and become
    accessible to anyone with log-read permission.

    Empty mapping is a silent no-op -- adopters who set no overrides
    should see zero log noise.
    """
    if not overrides:
        return
    # Sort for deterministic output across calls (helps log-grep + diff
    # against prior boot's line) and to avoid leaking ordering-derived
    # signals about how the dict was constructed.
    model_ids = sorted(overrides.keys())
    count = len(model_ids)
    plural = "" if count == 1 else "s"
    logger.info(
        "Loaded %d pricing override%s for model%s: %s",
        count,
        plural,
        plural,
        model_ids,
    )

get_pricing_overrides

get_pricing_overrides() -> dict[str, dict[str, float]]

Return a deep-ish copy of the currently-installed override table.

Copy semantics: mutating the returned mapping (or its rows) does not affect the process-wide registry.

Source code in src/symfonic/core/observability/pricing.py

def get_pricing_overrides() -> dict[str, dict[str, float]]:
    """Return a deep-ish copy of the currently-installed override table.

    Copy semantics: mutating the returned mapping (or its rows) does not
    affect the process-wide registry.
    """
    return {k: dict(v) for k, v in _PRICING_OVERRIDES.items()}

install_pricing_overrides

install_pricing_overrides(
    overrides: dict[str, dict[str, float]],
    *,
    skip_validation: bool = False,
) -> None

Agent-construction-safe install of pricing overrides (v8.7.1 M7).

Unlike :func:register_pricing_overrides (which treats an empty mapping as an explicit CLEAR), this:

• SKIPS registration entirely when overrides is empty, so a default-config agent cannot wipe a prior agent's process-wide overrides (the M7 clobber), and
• WARNS ONCE when a non-empty override set REPLACES a DIFFERENT non-empty set (two agents with divergent pricing in one process — only the last install wins, violating the one-agent-per-process contract).

Non-empty installs still flow through register_pricing_overrides (full validation) and emit_overrides_audit_log.

Source code in src/symfonic/core/observability/pricing.py

def install_pricing_overrides(
    overrides: dict[str, dict[str, float]],
    *,
    skip_validation: bool = False,
) -> None:
    """Agent-construction-safe install of pricing overrides (v8.7.1 M7).

    Unlike :func:`register_pricing_overrides` (which treats an empty mapping
    as an explicit CLEAR), this:

    * SKIPS registration entirely when ``overrides`` is empty, so a
      default-config agent cannot wipe a prior agent's process-wide
      overrides (the M7 clobber), and
    * WARNS ONCE when a non-empty override set REPLACES a DIFFERENT
      non-empty set (two agents with divergent pricing in one process —
      only the last install wins, violating the one-agent-per-process
      contract).

    Non-empty installs still flow through ``register_pricing_overrides``
    (full validation) and ``emit_overrides_audit_log``.
    """
    global _override_replace_warned
    if not overrides:
        # Empty -> preserve any previously-installed overrides.
        return
    existing = _PRICING_OVERRIDES
    if existing and existing != overrides and not _override_replace_warned:
        logger.warning(
            "install_pricing_overrides: replacing %d existing pricing "
            "override(s) with a different set of %d row(s). The pricing "
            "registry is process-wide; only the last install wins. This "
            "usually means two agents with divergent pricing were built in "
            "one process (one-agent-per-process contract).",
            len(existing),
            len(overrides),
        )
        _override_replace_warned = True
    register_pricing_overrides(overrides, skip_validation=skip_validation)
    emit_overrides_audit_log(overrides)

register_pricing_overrides

register_pricing_overrides(
    overrides: dict[str, dict[str, float]],
    *,
    skip_validation: bool = False,
) -> None

Install adopter-supplied pricing overrides.

v7.8.0 (Jarvio billing-pipeline ask). Each entry maps a model id (or a strict prefix) to a complete pricing row with keys input / output / cache_read / cache_write in USD per 1M tokens. Applied by SymfonicAgent.__init__ from FrameworkConfig.pricing_overrides.

Override lookup precedence is: exact match -> strict prefix match, BEFORE the built-in registry. This lets adopters either correct a stale built-in entry (exact match) or pre-empt a family-level lookup for an unreleased model (prefix match).

Partial-row overrides are rejected: a row that omits any of the required keys raises ValueError at registration time rather than silently defaulting to 0.0 at compute time. All values must be non-negative numbers.

v7.20.0 (architect verdict 2026-06-05, pricing-registry hardening §4): adds sanity bounds on input / output rates --

$1000/M (unit-confusion class) or < $0.0001/M (typo class) rejected at registration. Cache rates are exempted; they can be 0.0 for non-caching providers and can vary widely (Anthropic 1h multipliers, DashScope no-cache). Adopters with legitimate outlier rates (specialty embeddings, audio-token billing extras) pass skip_validation=True and own the consequences -- required-key and non-negative checks STILL fire even with skip_validation set.

Parameters:

Name	Type	Description	Default
overrides	dict[str, dict[str, float]]	Mapping of model-id-or-prefix to pricing row. Empty mapping is a no-op (clears any prior overrides).	required
skip_validation	bool	When True, bypasses the input/output sanity bounds. Required-key check, type check, and non-negative check are unaffected (these are footgun guards, not unit guards). Default False.	False

Raises:

Type	Description
ValueError	When any row omits a required key, carries a negative / non-numeric value, or (unless skip_validation=True) has an input/output rate outside the sanity-bound range.

Source code in src/symfonic/core/observability/pricing.py

def register_pricing_overrides(
    overrides: dict[str, dict[str, float]],
    *,
    skip_validation: bool = False,
) -> None:
    """Install adopter-supplied pricing overrides.

    v7.8.0 (Jarvio billing-pipeline ask).  Each entry maps a model id
    (or a strict prefix) to a complete pricing row with keys
    ``input`` / ``output`` / ``cache_read`` / ``cache_write`` in USD
    per 1M tokens.  Applied by ``SymfonicAgent.__init__`` from
    ``FrameworkConfig.pricing_overrides``.

    Override lookup precedence is: exact match -> strict prefix
    match, BEFORE the built-in registry.  This lets adopters either
    correct a stale built-in entry (exact match) or pre-empt a
    family-level lookup for an unreleased model (prefix match).

    Partial-row overrides are rejected: a row that omits any of the
    required keys raises ``ValueError`` at registration time rather
    than silently defaulting to 0.0 at compute time.  All values must
    be non-negative numbers.

    v7.20.0 (architect verdict 2026-06-05, pricing-registry hardening
    §4): adds sanity bounds on ``input`` / ``output`` rates --
    > $1000/M (unit-confusion class) or < $0.0001/M (typo class)
    rejected at registration.  Cache rates are exempted; they can be
    0.0 for non-caching providers and can vary widely (Anthropic 1h
    multipliers, DashScope no-cache).  Adopters with legitimate
    outlier rates (specialty embeddings, audio-token billing extras)
    pass ``skip_validation=True`` and own the consequences --
    required-key and non-negative checks STILL fire even with
    skip_validation set.

    Args:
        overrides: Mapping of model-id-or-prefix to pricing row.
            Empty mapping is a no-op (clears any prior overrides).
        skip_validation: When True, bypasses the input/output sanity
            bounds. Required-key check, type check, and non-negative
            check are unaffected (these are footgun guards, not unit
            guards). Default False.

    Raises:
        ValueError: When any row omits a required key, carries a
            negative / non-numeric value, or (unless
            ``skip_validation=True``) has an input/output rate outside
            the sanity-bound range.
    """
    global _PRICING_OVERRIDES
    validated: dict[str, dict[str, float]] = {}
    for model_id, row in overrides.items():
        if not isinstance(model_id, str) or not model_id.strip():
            raise ValueError(
                f"register_pricing_overrides: model id {model_id!r} "
                f"must be a non-empty string."
            )
        if not isinstance(row, dict):
            raise ValueError(
                f"register_pricing_overrides: pricing row for "
                f"{model_id!r} must be a dict; got {type(row).__name__}."
            )
        missing = _REQUIRED_PRICING_KEYS - row.keys()
        if missing:
            raise ValueError(
                f"register_pricing_overrides: pricing row for "
                f"{model_id!r} missing required keys {sorted(missing)}."
            )
        clean_row: dict[str, float] = {}
        for key in _REQUIRED_PRICING_KEYS:
            value = row[key]
            if not isinstance(value, (int, float)) or value < 0:
                raise ValueError(
                    f"register_pricing_overrides: pricing row for "
                    f"{model_id!r} key {key!r} must be a non-negative "
                    f"number; got {value!r}."
                )
            if (
                not skip_validation
                and key in _BOUNDS_CHECKED_KEYS
                and value > 0  # 0.0 input/output is unusual but not unit-confusion
            ):
                if value > _INPUT_OUTPUT_UPPER_BOUND_USD_PER_M:
                    raise ValueError(
                        f"register_pricing_overrides: pricing row for "
                        f"{model_id!r} key {key!r}={value!r} exceeds "
                        f"upper bound ${_INPUT_OUTPUT_UPPER_BOUND_USD_PER_M}/M "
                        f"tokens (rates this high are typically a unit-confusion "
                        f"bug -- per-token rate submitted as per-1M-token rate). "
                        f"For legitimate outlier rates pass "
                        f"register_pricing_overrides(..., skip_validation=True)."
                    )
                if value < _INPUT_OUTPUT_LOWER_BOUND_USD_PER_M:
                    raise ValueError(
                        f"register_pricing_overrides: pricing row for "
                        f"{model_id!r} key {key!r}={value!r} is below "
                        f"lower bound ${_INPUT_OUTPUT_LOWER_BOUND_USD_PER_M}/M "
                        f"tokens (rates this low are typically a typo / "
                        f"misread decimal). For legitimate sub-cent rates "
                        f"pass register_pricing_overrides(..., skip_validation=True)."
                    )
            clean_row[key] = float(value)
        validated[model_id] = clean_row
    _PRICING_OVERRIDES = validated