qfa.domain.usage_models

Usage tracking + aggregation domain models.

Split out from qfa.domain.models to keep the broad request/response models separate from the usage-tracking cluster. The cluster has two halves:

• Persistence primitives — Operation, CallStatus, CallContext, LLMCallRecord. These describe a single LLM-call attempt and the context propagated to the tracking adapter.

• Aggregations — DistributionStats, UsageMetrics, OperationStats, TenantUsageStats, TenantStats, OperationUsageStats. These are the per-tenant / per-operation views returned by /v1/usage endpoints.

All models are immutable (frozen) Pydantic models per ADR-001.

Classes

CallContext(*, tenant_id, operation, call_id)	Per-call context propagated via ContextVar from orchestrator to tracker.
CallStatus(*values)	Outcome of a single LLM call attempt.
DistributionStats(*, avg, min, max, p5, p95, ...)	Statistical distribution summary over a numeric column.
LLMCallRecord(*, tenant_id, operation, ...)	A single recorded LLM call attempt for usage and cost tracking.
Operation(*values)	Orchestrator operations that produce LLM calls.
OperationStats(*, total_calls[, ...])	Per-operation usage stats nested inside TenantUsageStats.
OperationUsageStats(*, total_calls[, ...])	Per-operation (or grand-total) usage stats with nested per-tenant breakdown.
TenantStats(*, total_calls[, failed_calls, ...])	Per-tenant usage stats nested inside an operation block.
TenantUsageStats(*, total_calls[, ...])	Per-tenant (or grand-total) usage stats — per-invocation + per-LLM-call.
UsageMetrics(*, total_calls[, failed_calls, ...])	Aggregated stats over a set of records.

class qfa.domain.usage_models.Operation(*values)

Bases: StrEnum

Orchestrator operations that produce LLM calls.

Stored as plain strings in the database; new members can be added without a DB migration. UNKNOWN is a sentinel for backfilled rows from before per-operation tracking was introduced and must never be removed (removal would orphan historical rows).

ANALYZE = 'analyze'

SUMMARIZE = 'summarize'

SUMMARIZE_AGGREGATE = 'summarize_aggregate'

ASSIGN_CODES = 'assign_codes'

DETECT_SENSITIVE = 'detect_sensitive'

UNKNOWN = 'unknown'

class qfa.domain.usage_models.CallStatus(*values)

Bases: StrEnum

Outcome of a single LLM call attempt.

OK = 'ok'

ERROR = 'error'

class qfa.domain.usage_models.CallContext(*, tenant_id: str, operation: Operation, call_id: UUID)

Bases: BaseModel

Per-call context propagated via ContextVar from orchestrator to tracker.

Variables:

• tenant_id (str) – Tenant making the call.

• operation (Operation) – Public orchestrator operation that issued the call.

• call_id (UUID) – Correlation ID for the API call. All LLM calls made inside one call_scope share this ID, enabling per-invocation cost aggregation across the fan-out of LLM calls.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config = {'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

tenant_id: str

operation: Operation

call_id: UUID

class qfa.domain.usage_models.LLMCallRecord(*, tenant_id: str, operation: Operation, call_id: UUID, timestamp: datetime, call_duration_ms: int, model: str, input_tokens: int = 0, output_tokens: int = 0, cost_usd: Decimal = Decimal('0'), status: CallStatus, error_class: str | None = None)

Bases: BaseModel

A single recorded LLM call attempt for usage and cost tracking.

Recorded once per LLM-call attempt — success or failure. cost_usd and token counts are populated only for successful attempts; failures record zeros plus error_class.

Variables:

• tenant_id (str) – Tenant that made the call.

• operation (Operation) – Public orchestrator operation that issued the call.

• call_id (UUID) – Correlation ID linking all LLM calls made within a single API invocation. Shared across the fan-out of LLM calls from one call_scope, enabling per-invocation aggregation in usage reports.

• timestamp (datetime) – UTC wall-clock when the call started.

• call_duration_ms (int) – Wall-clock duration of the call in milliseconds.

• model (str) – The LLM model used.

• input_tokens (int) – Number of input (prompt) tokens; 0 on failure.

• output_tokens (int) – Number of output (completion) tokens; 0 on failure.

• cost_usd (Decimal) – Estimated cost in USD; 0 on failure.

• status (CallStatus) – Outcome of the attempt.

• error_class (str | None) – type(exc).__name__ when status == CallStatus.ERROR; None otherwise. Enforced by model_validator.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config = {'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

tenant_id: str

operation: Operation

call_id: UUID

timestamp: datetime

call_duration_ms: int

model: str

input_tokens: int

output_tokens: int

cost_usd: Decimal

status: CallStatus

error_class: str | None

class qfa.domain.usage_models.DistributionStats(*, avg: float, min: float, max: float, p5: float, p95: float, total: int)

Bases: BaseModel

Statistical distribution summary over a numeric column.

Used uniformly for call_duration (milliseconds), input_tokens, and output_tokens. total is the sum of the underlying values in the window and is identical between the per-invocation and per-LLM-call views — both sum the same raw rows, just regrouped.

Variables:

• avg (float) – Mean value.

• min (float) – Minimum value.

• max (float) – Maximum value.

• p5 (float) – 5th percentile.

• p95 (float) – 95th percentile.

• total (int) – Sum of the values in the window (total milliseconds of LLM time for call_duration; total tokens for input_tokens / output_tokens).

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config = {'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

avg: float

min: float

max: float

p5: float

p95: float

total: int

class qfa.domain.usage_models.UsageMetrics(*, total_calls: int, failed_calls: int = 0, total_cost_usd: Decimal = Decimal('0'), call_duration: DistributionStats, input_tokens: DistributionStats, output_tokens: DistributionStats)

Bases: BaseModel

Aggregated stats over a set of records.

Whether the records are per-LLM-call rows or per-invocation roll-ups is fixed by the containing field, not by this class. UsageMetrics is used directly for the per-LLM-call llm_call_stats block on TenantUsageStats and OperationStats, and as the base class for the per-invocation totals on TenantUsageStats / OperationStats.

Per-field semantics are in the Field(description=...) below and surface in the OpenAPI schema at GET /docs.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config = {'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

total_calls: int

failed_calls: int

total_cost_usd: Decimal

call_duration: DistributionStats

input_tokens: DistributionStats

output_tokens: DistributionStats

class qfa.domain.usage_models.OperationStats(*, total_calls: int, failed_calls: int = 0, total_cost_usd: Decimal = Decimal('0'), call_duration: DistributionStats, input_tokens: DistributionStats, output_tokens: DistributionStats, operation: Operation, llm_call_stats: UsageMetrics)

Bases: UsageMetrics

Per-operation usage stats nested inside TenantUsageStats.

Inherits all metric fields from UsageMetrics (per-invocation semantics) and adds the operation discriminator plus a parallel llm_call_stats block giving the per-LLM-call view for the same operation.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config = {'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

operation: Operation

llm_call_stats: UsageMetrics

class qfa.domain.usage_models.TenantUsageStats(*, total_calls: int, failed_calls: int = 0, total_cost_usd: Decimal = Decimal('0'), call_duration: DistributionStats, input_tokens: DistributionStats, output_tokens: DistributionStats, tenant_id: str | None = None, llm_call_stats: UsageMetrics, operations: tuple[OperationStats, ...] = ())

Bases: UsageMetrics

Per-tenant (or grand-total) usage stats — per-invocation + per-LLM-call.

Inherits per-invocation metric fields from UsageMetrics and adds the per-LLM-call view, the per-operation breakdown, and the optional tenant_id (None is the grand-total sentinel used by /v1/usage/all/by-tenant).

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config = {'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

tenant_id: str | None

llm_call_stats: UsageMetrics

operations: tuple[OperationStats, ...]

class qfa.domain.usage_models.TenantStats(*, total_calls: int, failed_calls: int = 0, total_cost_usd: Decimal = Decimal('0'), call_duration: DistributionStats, input_tokens: DistributionStats, output_tokens: DistributionStats, tenant_id: str, llm_call_stats: UsageMetrics)

Bases: UsageMetrics

Per-tenant usage stats nested inside an operation block.

Mirrors OperationStats but for the inverse hierarchy used by /v1/usage/all/by-operation: each OperationUsageStats carries a list of these blocks, one per tenant that has activity for that operation in the window. Inherits per-invocation metric fields from UsageMetrics and adds the tenant_id discriminator plus the parallel llm_call_stats block for the per-LLM-call view of the same (operation, tenant) slice.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config = {'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

tenant_id: str

llm_call_stats: UsageMetrics

class qfa.domain.usage_models.OperationUsageStats(*, total_calls: int, failed_calls: int = 0, total_cost_usd: Decimal = Decimal('0'), call_duration: DistributionStats, input_tokens: DistributionStats, output_tokens: DistributionStats, operation: Operation | None = None, llm_call_stats: UsageMetrics, tenants: tuple[TenantStats, ...] = ())

Bases: UsageMetrics

Per-operation (or grand-total) usage stats with nested per-tenant breakdown.

Inverse hierarchy of TenantUsageStats: top-level aggregation is by orchestrator operation, with a list of per-tenant blocks underneath. Used by /v1/usage/all/by-operation. operation is None on the grand-total entry (cross-operation, cross-tenant).

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config = {'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

operation: Operation | None

llm_call_stats: UsageMetrics

tenants: tuple[TenantStats, ...]