Coverage for src / lexigram / contracts / ai / relay / ledger.py: 26%

1"""Relay ledger contracts: quota credit-in records and service protocol. 

2 

3The ledger is the framework's credit-in mechanism: it journals 

4top-up and check-in credits so operators (via ``relay.billing`` 

5permissioned admin surfaces) and applications have an audited, single 

6mutation path for adding quota. The framework records usage but no 

7wallet balance; applications compute or enforce balances from ledger 

8credits and usage records. Check-in awards are caller-supplied — 

9award amounts and cadences are application policy, not framework 

10constants. All mutations are idempotent or compare-and-set: a top-up 

11settles once (CAS on status), a check-in is PK-guaranteed once per 

12``(user_id, day)``. 

13""" 

14 

15from __future__ import annotations 

16 

17from dataclasses import dataclass 

18from datetime import date 

19from typing import TYPE_CHECKING, Literal, Protocol, runtime_checkable 

20 

21from lexigram.contracts.core.result import Result 

22 

23if TYPE_CHECKING: 

24 from lexigram.contracts.ai.governance import RelayUsageScope 

25 

26RelayTopUpStatus = Literal["pending", "completed", "failed"] 

27 

28 

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

30class RelayTopUpRecord: 

31 """One quota credit-in record. 

32 

33 Attributes: 

34 reference_id: Unique reference for the credit (primary key). 

35 user_id: User the credit applies to. 

36 amount: Credited amount as a Decimal string, never negative. 

37 status: ``pending`` (awaiting settlement), ``completed``, or 

38 ``failed``. 

39 created_at: ISO-8601 creation timestamp (UTC). 

40 """ 

41 

42 reference_id: str 

43 user_id: str 

44 amount: str 

45 status: RelayTopUpStatus 

46 created_at: str 

47 

48 def __post_init__(self) -> None: 

49 """Reject empty identities, negative amounts, and bad statuses.""" 

50 if not self.reference_id: 

51 raise ValueError("reference_id must be non-empty") 

52 if not self.user_id: 

53 raise ValueError("user_id must be non-empty") 

54 try: 

55 if float(self.amount) < 0: 

56 raise ValueError("amount must be non-negative") 

57 except ValueError as exc: 

58 raise ValueError("amount must be a non-negative number") from exc 

59 if self.status not in ("pending", "completed", "failed"): 

60 raise ValueError(f"unknown top-up status: {self.status}") 

61 

62 

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

64class RelayCheckinRecord: 

65 """One daily check-in award granted to a user. 

66 

67 Attributes: 

68 user_id: User the award applies to. 

69 day: Award day in ISO ``YYYY-MM-DD`` (UTC). 

70 award: Awarded amount as a Decimal string, never negative. 

71 created_at: ISO-8601 creation timestamp (UTC). 

72 """ 

73 

74 user_id: str 

75 day: str 

76 award: str 

77 created_at: str 

78 

79 def __post_init__(self) -> None: 

80 """Reject empty identities, bad dates, and negative awards.""" 

81 if not self.user_id: 

82 raise ValueError("user_id must be non-empty") 

83 try: 

84 date.fromisoformat(self.day) 

85 except ValueError as exc: 

86 raise ValueError("day must be ISO YYYY-MM-DD") from exc 

87 try: 

88 if float(self.award) < 0: 

89 raise ValueError("award must be non-negative") 

90 except ValueError as exc: 

91 raise ValueError("award must be a non-negative number") from exc 

92 

93 

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

95class RelayLedgerError: 

96 """A domain error returned from ledger operations. 

97 

98 Attributes: 

99 code: Machine-readable error code (``already_checked_in``, 

100 ``not_found``, ``stale_settlement``, ...). 

101 message: Public, redaction-safe error message. 

102 """ 

103 

104 code: str 

105 message: str 

106 

107 

108@runtime_checkable 

109class RelayLedgerServiceProtocol(Protocol): 

110 """Governance quota credit-in over the relay ledger. 

111 

112 All mutations journal a record and emit a structured event; 

113 reservations and settled usage (Plan C) are never touched by this 

114 protocol. 

115 """ 

116 

117 async def credit( 

118 self, scope: RelayUsageScope, amount: str, reason: str 

119 ) -> Result[None, RelayLedgerError]: 

120 """Journal an immediate completed credit for *scope*.""" 

121 ... 

122 

123 async def settle_topup( 

124 self, reference_id: str, expected_status: str 

125 ) -> Result[None, RelayLedgerError]: 

126 """Flip *reference_id* from *expected_status* to completed exactly once.""" 

127 ... 

128 

129 async def checkin( 

130 self, user_id: str, award: str 

131 ) -> Result[RelayCheckinRecord, RelayLedgerError]: 

132 """Award *award* to *user_id* once per UTC day.""" 

133 ... 

134 

135 async def list_topups( 

136 self, user_id: str | None, limit: int 

137 ) -> list[RelayTopUpRecord]: 

138 """List top-up records, newest first, optionally for one user.""" 

139 ... 

140 

141 

142__all__ = [ 

143 "RelayCheckinRecord", 

144 "RelayLedgerError", 

145 "RelayLedgerServiceProtocol", 

146 "RelayTopUpRecord", 

147 "RelayTopUpStatus", 

148]