Coverage for src / lexigram / contracts / ai / relay / usage.py: 0%

1"""Usage and rankings read contracts for the relay gateway. 

2 

3The usage service aggregates persisted request logs into daily per-token 

4usage and per-model rankings. Value types are frozen so they can cross 

5package boundaries safely. 

6""" 

7 

8from __future__ import annotations 

9 

10from dataclasses import dataclass 

11from typing import TYPE_CHECKING, Protocol, runtime_checkable 

12 

13if TYPE_CHECKING: 

14 from lexigram.contracts.ai.relay.logs import RelayRequestLogEntry 

15 

16 

17@dataclass(frozen=True, slots=True) 

18class RelayDailyUsage: 

19 """One day of token/cost usage for a user.""" 

20 

21 day: str # ISO date (YYYY-MM-DD) as stored by the aggregate 

22 prompt_tokens: int 

23 completion_tokens: int 

24 cost: str # Decimal string 

25 

26 

27@dataclass(frozen=True, slots=True) 

28class RelayModelRank: 

29 """Aggregate usage for one model over a window.""" 

30 

31 model: str 

32 completion_tokens: int 

33 request_count: int 

34 cost: str # Decimal string 

35 

36 

37@runtime_checkable 

38class RelayUsageServiceProtocol(Protocol): 

39 """Read model over persisted relay request logs. 

40 

41 Both methods are best-effort reads over the durable store; query 

42 failures surface as warnings, not request-path failures. 

43 """ 

44 

45 async def daily_usage(self, user_id: str, days: int) -> list[RelayDailyUsage]: ... 

46 

47 async def model_rank(self, days: int, limit: int) -> list[RelayModelRank]: ... 

48 

49 async def list_requests( 

50 self, 

51 days: int, 

52 page: int, 

53 page_size: int, 

54 *, 

55 user_id: str | None = None, 

56 token_id: str | None = None, 

57 ) -> list[RelayRequestLogEntry]: 

58 """List recent request-log entries, newest first.""" 

59 ... 

60 

61 

62__all__ = ["RelayDailyUsage", "RelayModelRank", "RelayUsageServiceProtocol"]