primus.optimization.semantic_authority module

Durable host authority shared by all semantic diagnosis model attempts.

class primus.optimization.semantic_authority.SemanticAttemptAuthority(*, coordinator: SemanticBudgetCoordinator, target_id: str, call_site: str, max_attempts: int)

Bases: object

Per-target/call-site view over one shared run ledger.

__init__(*, coordinator: SemanticBudgetCoordinator, target_id: str, call_site: str, max_attempts: int) → None

direct_plan(*, attempt: int, max_input_tokens: int, max_output_tokens: int, request_payload: Mapping[str, Any]) → SemanticCallPlan

reserve(plan: Any) → Any: Implement Tactus ModelAttemptAuthority without importing it eagerly.

reserve_direct(plan: SemanticCallPlan) → SemanticReservationDecision

settle(outcome: Any) → None

settle_direct(reservation_id: str, usage: SemanticUsage, *, output_text: str) → dict[str, Any]

unknown_direct(reservation_id: str, *, reason: str) → None

exception primus.optimization.semantic_authority.SemanticAuthorityError

Bases: RuntimeError

A semantic attempt could not be safely authorized or reconciled.

exception primus.optimization.semantic_authority.SemanticAuthorityPublicationError

Bases: SemanticAuthorityError

Durable semantic authority state could not be loaded or committed.

class primus.optimization.semantic_authority.SemanticBudgetCoordinator(*, ledger: SemanticBudgetLedger, persist: Callable[[Mapping[str, Any]], Any])

Bases: object

Own one Report-persisted ledger and publish every mutation in order.

__init__(*, ledger: SemanticBudgetLedger, persist: Callable[[Mapping[str, Any]], Any]) → None

property ledger: SemanticBudgetLedger

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

reserve(plan: SemanticCallPlan) → SemanticReservationDecision

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

classmethod start_or_resume(*, report_service: Any, run_key: str, spec: SemanticBudgetSpec) → SemanticBudgetCoordinator

view(*, target_id: str, call_site: str, max_attempts: int) → SemanticAttemptAuthority

exception primus.optimization.semantic_authority.SemanticOutcomeUnknown

Bases: SemanticAuthorityError

Provider contact occurred but durable settlement is incomplete.

exception primus.optimization.semantic_authority.SemanticReplayBlocked

Bases: SemanticAuthorityError

Durable evidence proves that contacting the provider again is unsafe.

class primus.optimization.semantic_authority.SemanticReservationDecision(status: 'str', reservation_id: 'str', replay_payload: 'dict[str, Any] | None' = None)

Bases: object

__init__(status: str, reservation_id: str, replay_payload: dict[str, Any] | None = None) → None

replay_payload: dict[str, Any] | None = None

reservation_id: str

status: str

primus.optimization.semantic_authority.semantic_budget_spec(max_cost_usd: str) → SemanticBudgetSpec