primus.optimization.semantic_budget module

Exact, replay-safe authorization ledger for semantic model calls.

This module is intentionally independent from the float-oriented runtime and optimizer budgets. It authorizes calls from a checked-in pricing snapshot and serializes all money as canonical decimal strings.

exception primus.optimization.semantic_budget.SemanticBudgetConflict

Bases: SemanticBudgetError

Replay supplied data that conflicts with a durable ledger entry.

exception primus.optimization.semantic_budget.SemanticBudgetError

Bases: RuntimeError

Base class for semantic budget authorization failures.

exception primus.optimization.semantic_budget.SemanticBudgetExceeded

Bases: SemanticBudgetError

A worst-case reservation would exceed the frozen run budget.

class primus.optimization.semantic_budget.SemanticBudgetLedger(*, run_key: str, spec: SemanticBudgetSpec, revision: int = 0, entries: Mapping[str, Mapping[str, Any]] | None = None)

Bases: object

Single-run authorization ledger with deterministic replay transitions.

__init__(*, run_key: str, spec: SemanticBudgetSpec, revision: int = 0, entries: Mapping[str, Mapping[str, Any]] | None = None) → None

cancel_pre_contact(reservation_id: str, *, reason: str) → dict[str, Any]

digest() → str

classmethod from_dict(value: Mapping[str, Any], *, expected_spec: SemanticBudgetSpec | None = None) → SemanticBudgetLedger

get(reservation_id: str) → dict[str, Any]

mark_outcome_unknown(reservation_id: str, *, reason: str) → dict[str, Any]

reserve(plan: SemanticCallPlan) → dict[str, Any]

settle(reservation_id: str, usage: SemanticUsage, *, replay_payload: Mapping[str, Any] | None = None) → dict[str, Any]

summary() → dict[str, Any]

to_dict() → dict[str, Any]

class primus.optimization.semantic_budget.SemanticBudgetSpec(max_cost_usd: 'str', pricing_version: 'str', schema_version: 'str' = 'semantic-budget-v1')

Bases: object

__init__(max_cost_usd: str, pricing_version: str, schema_version: str = 'semantic-budget-v1') → None

classmethod from_dict(value: Mapping[str, Any]) → SemanticBudgetSpec

property max_cost: Decimal

max_cost_usd: str

pricing_version: str

schema_version: str = 'semantic-budget-v1'

to_dict() → dict[str, str]

class primus.optimization.semantic_budget.SemanticCallPlan(run_key: 'str', target_id: 'str', call_site: 'str', attempt: 'int', max_attempts: 'int', provider: 'str', model: 'str', pricing_version: 'str', max_input_tokens: 'int', max_output_tokens: 'int', request_hash: 'str | None' = None, external_attempt_id: 'str | None' = None, schema_version: 'str' = 'semantic-call-plan-v1')

Bases: object

__init__(run_key: str, target_id: str, call_site: str, attempt: int, max_attempts: int, provider: str, model: str, pricing_version: str, max_input_tokens: int, max_output_tokens: int, request_hash: str | None = None, external_attempt_id: str | None = None, schema_version: str = 'semantic-call-plan-v1') → None

attempt: int

call_site: str

external_attempt_id: str | None = None

classmethod from_dict(value: Mapping[str, Any]) → SemanticCallPlan

identity_dict() → dict[str, Any]

max_attempts: int

max_input_tokens: int

max_output_tokens: int

model: str

pricing_version: str

provider: str

request_hash: str | None = None

property reservation_id: str

run_key: str

schema_version: str = 'semantic-call-plan-v1'

target_id: str

to_dict() → dict[str, Any]

class primus.optimization.semantic_budget.SemanticModelPrice(provider: 'str', model: 'str', input_usd_per_million: 'Decimal', cached_input_usd_per_million: 'Decimal', output_usd_per_million: 'Decimal')

Bases: object

__init__(provider: str, model: str, input_usd_per_million: Decimal, cached_input_usd_per_million: Decimal, output_usd_per_million: Decimal) → None

cached_input_usd_per_million: Decimal

input_usd_per_million: Decimal

model: str

output_usd_per_million: Decimal

provider: str

class primus.optimization.semantic_budget.SemanticPricingSnapshot(pricing_version: 'str', models: 'Mapping[str, SemanticModelPrice]')

Bases: object

__init__(pricing_version: str, models: Mapping[str, SemanticModelPrice]) → None

models: Mapping[str, SemanticModelPrice]

price_for(provider: str, model: str) → SemanticModelPrice

pricing_version: str

class primus.optimization.semantic_budget.SemanticUsage(input_tokens: 'int', output_tokens: 'int', cached_input_tokens: 'int' = 0, provider_request_id: 'str | None' = None, schema_version: 'str' = 'semantic-usage-v1')

Bases: object

__init__(input_tokens: int, output_tokens: int, cached_input_tokens: int = 0, provider_request_id: str | None = None, schema_version: str = 'semantic-usage-v1') → None

cached_input_tokens: int = 0

classmethod from_dict(value: Mapping[str, Any]) → SemanticUsage

input_tokens: int

output_tokens: int

provider_request_id: str | None = None

schema_version: str = 'semantic-usage-v1'

to_dict() → dict[str, Any]

exception primus.optimization.semantic_budget.UnknownSemanticPricing

Bases: SemanticBudgetError

The requested immutable provider/model price is not authorized.

primus.optimization.semantic_budget.canonical_json_bytes(value: Mapping[str, Any]) → bytes

primus.optimization.semantic_budget.decimal_string(value: Decimal) → str: Return stable non-exponent decimal JSON text without losing precision.

primus.optimization.semantic_budget.load_semantic_pricing(pricing_version: str) → SemanticPricingSnapshot

primus.optimization.semantic_budget.stakeholder_budget_evidence(ledger_value: Mapping[str, Any], *, evidence_reference: str) → dict[str, Any]

Return a privacy-safe, deterministic projection of a frozen ledger.

This is intentionally an aggregate rather than a transport copy of the ledger. Operators need exact Decimal reconciliation and a durable digest, but stakeholder artifacts must not expose target identifiers, request payloads, replay payloads, provider request IDs, or unknown-outcome text.