feat: derive TPM reservation and pin the usage_source domain

Task 1 of the est_tokens decoupling: capability only, no call site
touched, so library behaviour is unchanged word for word.

SourceConfig.effective_est_tokens() returns the explicit est_tokens when
set, otherwise tpm // 60 floored at 1, otherwise 0 when the TPM gate is
off. The divisor is scale free: any quota size yields the same in-flight
ceiling of roughly sixty calls, which is what makes the default
explainable where a fixed constant was not.

USAGE_SOURCES lands with the two assertions the design asks for, not as
a dead constant. test_usage_source_domain.py drives every production
point -- _resolve_usage, _resolve_embedding_usage, _merge and the three
TelemetryEmitter.emit_* helpers -- and asserts the output stays inside
the domain; it is a separate file because the assertion spans
transports, embedding and telemetry, and the innermost kernel test
should not depend on implementations. The second assertion pins the
opposite ruling: constructing LLMResponse with an out-of-domain value
must not raise, since a bare ValueError at a runtime construction point
falls outside the four error categories and would escape chat().

tpm > 0 with est_tokens = 0 is still rejected until Task 4, so the
derivation tests build the future-legal shape through a helper that
bypasses the constraint; the helper collapses back to _make_source once
the constraint is gone.
This commit is contained in:
2026-07-30 10:05:11 -04:00
parent 61231f7f6e
commit e5dbcf5d33
3 changed files with 290 additions and 0 deletions
+14
View File
@@ -9,6 +9,12 @@ from typing import Any
_MISSING_DONE_DOMAIN = frozenset({"retry", "salvage"})
USAGE_SOURCES = frozenset({"measured", "estimated", "unavailable"})
"""usage_source 值域;仅约束库内生产侧取值,不在 frozen dataclass 上做运行时校验。"""
_EST_TOKENS_QUOTA_DIVISOR = 60
"""未显式配置时的预扣量除数: 假定一次调用约占一秒钟的 TPM 配额份额。"""
@dataclass(frozen=True)
class LLMResponse:
@@ -107,6 +113,14 @@ class SourceConfig:
self._validate_gates()
self._validate_watchdog()
def effective_est_tokens(self) -> int:
"""TPM 入场预扣量: 显式配置优先,否则按 tpm 派生(设计 §2.2)。"""
if self.est_tokens > 0:
return self.est_tokens
if self.tpm > 0:
return max(1, self.tpm // _EST_TOKENS_QUOTA_DIVISOR)
return 0
def _validate_identity(self) -> None:
for attr in ("name", "provider", "base_url", "api_key", "model"):
if not getattr(self, attr).strip():