symfonic.services.observability.cost

cost

Cost accounting — one calculation, delegated to the shipped registry.

This is emphatically not a second pricing implementation. The framework already ships one registry-backed calculator (:meth:symfonic.core.contracts.callbacks.TokenUsage.from_dict, which resolves model rows, applies cache and reasoning rates, and reports when a model is unknown). A parallel calculator that agrees today is precisely the failure this task exists to end, so the accountant converts vocabulary and delegates.

What it adds is a boundary: cost becomes an observation on a port instead of a field several observers each recompute from a usage dict.

CostAccountant

Prices one run's usage, and says whether the registry actually knew.

assess

assess(scope: RunScope, usage: UsageDelta, *, text: str | None = None, duration_ms: float = 0.0) -> CostObservation

Price usage for scope's model.

A zero-usage run is still priced rather than skipped: "this run cost nothing" and "nobody looked" are different facts, and only the first one is worth reporting.

Source code in src/symfonic/services/observability/cost.py

def assess(
    self,
    scope: RunScope,
    usage: UsageDelta,
    *,
    text: str | None = None,
    duration_ms: float = 0.0,
) -> CostObservation:
    """Price ``usage`` for ``scope``'s model.

    A zero-usage run is still priced rather than skipped: "this run cost
    nothing" and "nobody looked" are different facts, and only the first
    one is worth reporting.
    """
    typed = TokenUsage.from_dict(usage_dict(usage), scope.model)
    return CostObservation(
        scope=scope,
        model=scope.model,
        usage=usage,
        cost_usd=typed.cost_usd,
        pricing_unknown=typed.pricing_unknown,
        breakdown=MappingProxyType(
            {
                "input_tokens": typed.input_tokens,
                "output_tokens": typed.output_tokens,
                "cached_tokens": typed.cached_tokens,
                "cache_creation_tokens": typed.cache_creation_tokens,
                "reasoning_tokens": typed.reasoning_tokens,
            }
        ),
        text=text,
        duration_ms=duration_ms,
    )

price_or_none

price_or_none(accountant: Any, scope: RunScope, usage: UsageDelta, *, text: str | None, duration_ms: float) -> CostObservation | None

Price a run, or report None because the registry could not.

This guard lives beside the calculation rather than at the call site because the call site's own isolation cannot reach it: the bridge wraps every observer call, but the accountant runs while the argument to that call is still being built. A malformed pricing row therefore escapes an event sink that documents "never raises", and the run loses its closing observation entirely — the metrics collector keeps its per-run entry forever and an exporter's root span never closes. A pricing fault must cost the cost observation and nothing else.

Source code in src/symfonic/services/observability/cost.py

def price_or_none(
    accountant: Any,
    scope: RunScope,
    usage: UsageDelta,
    *,
    text: str | None,
    duration_ms: float,
) -> CostObservation | None:
    """Price a run, or report ``None`` because the registry could not.

    This guard lives beside the calculation rather than at the call site
    because the call site's own isolation cannot reach it: the bridge wraps
    every *observer* call, but the accountant runs while the argument to that
    call is still being built. A malformed pricing row therefore escapes an
    event sink that documents "never raises", and the run loses its closing
    observation entirely — the metrics collector keeps its per-run entry
    forever and an exporter's root span never closes. A pricing fault must
    cost the cost observation and nothing else.
    """
    try:
        return accountant.assess(scope, usage, text=text, duration_ms=duration_ms)
    except Exception:  # noqa: BLE001 - telemetry never breaks a run
        logger.warning(
            "cost accounting raised for run %s; suppressing", scope.run_id,
            exc_info=True,
        )
        return None

usage_dict

usage_dict(usage: UsageDelta) -> dict[str, Any]

Kernel usage in the dict shape every shipped callback already parses.

total_tokens is deliberately absent: the shipped TokenUsage derives totals itself, and handing it a third number invites the two to drift.

The cache and reasoning dimensions are carried through because the registry prices them: it subtracts cache reads and cache writes from the billable input and charges each at its own rate. Rendering only the two totals is not a simplification — it bills a cache-heavy run at roughly ten times what it cost.

cache_ttl rides along for the same reason one dimension over: it is the rate selector for the cache writes, and TokenUsage.from_dict reads it to pick the 1h write rate over the 5-minute default. Dropping it while keeping the write count reports a number that looks right and bills wrong.

Source code in src/symfonic/services/observability/cost.py

def usage_dict(usage: UsageDelta) -> dict[str, Any]:
    """Kernel usage in the dict shape every shipped callback already parses.

    ``total_tokens`` is deliberately absent: the shipped ``TokenUsage`` derives
    totals itself, and handing it a third number invites the two to drift.

    The cache and reasoning dimensions are carried through because the registry
    prices them: it subtracts cache reads and cache writes from the billable
    input and charges each at its own rate. Rendering only the two totals is
    not a simplification — it bills a cache-heavy run at roughly ten times what
    it cost.

    ``cache_ttl`` rides along for the same reason one dimension over: it is the
    *rate selector* for the cache writes, and ``TokenUsage.from_dict`` reads it
    to pick the 1h write rate over the 5-minute default. Dropping it while
    keeping the write count reports a number that looks right and bills wrong.
    """
    rendered: dict[str, Any] = {
        "input_tokens": usage.input_tokens,
        "output_tokens": usage.output_tokens,
    }
    for field_name, key in _DIMENSIONS:
        value = getattr(usage, field_name, 0)
        if value:
            rendered[key] = int(value)
    cache_ttl = getattr(usage, "cache_ttl", None)
    if cache_ttl:
        rendered["cache_ttl"] = cache_ttl
    return rendered