diff --git a/src/polygateway/types.py b/src/polygateway/types.py index 7a21924..f67d485 100644 --- a/src/polygateway/types.py +++ b/src/polygateway/types.py @@ -6,6 +6,8 @@ fake,字段顺序即公共承诺;新增字段只增不删且必带默认值。 import dataclasses import json +import math +import re from collections.abc import Mapping from dataclasses import dataclass, field from types import MappingProxyType @@ -31,6 +33,18 @@ USAGE_SOURCES = frozenset({"measured", "estimated", "unavailable"}) _EST_TOKENS_QUOTA_DIVISOR = 60 """未显式配置时的预扣量除数: 假定一次调用约占一秒钟的 TPM 配额份额。""" +_TENANT_ID_MAX_LEN = 128 +_META_MAX_KEYS = 16 +_META_VALUE_MAX_LEN = 256 +_META_RESERVED_PREFIX = "pg_" +_META_KEY_RE = re.compile(r"[a-z0-9_.]{1,64}") +"""调用方维度的形态上限(issue #11 §4.2)。 + +数值取自同类系统的量级(Loki labels 15 / Salesforce 自定义索引 25 / +Sentry tag 200 字符),非本项目实测;键字符集照搬 OTel semconv。 +`pg_` 前缀留给库将来的内建维度——同类做法见 LangSmith 的 `ls_`、 +Traceloop 的 `traceloop.`;本版库自身不写入任何该前缀的键。""" + def validate_request_overlay(overlay: Mapping[str, Any], *, origin: str) -> dict[str, Any]: """校验采样参数覆盖层并返回浅拷贝;origin 用于把错误指回配置/调用点。 @@ -59,6 +73,85 @@ def validate_request_overlay(overlay: Mapping[str, Any], *, origin: str) -> dict return dict(overlay) +def validate_caller_dimensions( + tenant_id: str | None, + meta: Mapping[str, Any] | None, + *, + origin: str, +) -> tuple[str | None, dict[str, Any]]: + """校验调用方自定义维度并返回浅拷贝;origin 用于把错误指回调用点(issue #11 §4.2)。 + + 一切超限**报错而非静默丢弃**(P5): 同类系统里 Langfuse 对超长 value 直接 + 丢掉,那会让调用方以为记上了而实际没有。报错点必须在进洋葱之前——洋葱内 + 的失败都被遥测层降级成 warning,校验放那里等于没有校验。 + """ + _validate_tenant_id(tenant_id, origin) + if meta is None: + return tenant_id, {} + # 键形态须先于值校验: 非 str 键若拖到值校验之后,会以更晦涩的形态报出来 + _validate_meta_keys(meta, origin) + _validate_meta_values(meta, origin) + return tenant_id, dict(meta) + + +def _validate_tenant_id(tenant_id: str | None, origin: str) -> None: + """租户标识形态;首尾空白**拒绝而非 strip**(见函数体注释)。""" + if tenant_id is None: + return + if not isinstance(tenant_id, str): + raise ValueError(f"{origin} 的 tenant_id 必须是 str: {tenant_id!r}") + # 悄悄 strip 会让 " t1" 变成 "t1": 二者在 RLS policy 的等值比较下是两个 + # 不同租户,替调用方改写值等于把它的行藏进另一个租户,且不报错 + if tenant_id != tenant_id.strip(): + raise ValueError( + f"{origin} 的 tenant_id 不得含首尾空白: {tenant_id!r}" + "(RLS 等值比较下它与去空白版本是两个租户)" + ) + if not tenant_id: + raise ValueError(f"{origin} 的 tenant_id 不得为空串(空串是未归属行的哨兵值)") + if len(tenant_id) > _TENANT_ID_MAX_LEN: + raise ValueError(f"{origin} 的 tenant_id 超长(上限 {_TENANT_ID_MAX_LEN}): {len(tenant_id)}") + + +def _validate_meta_keys(meta: Mapping[str, Any], origin: str) -> None: + """键形态与数量;键集合被假定为低基数且稳定,故收紧到 OTel semconv 字符集。""" + for key in meta: + if not isinstance(key, str): + raise ValueError(f"{origin} 的 meta 键必须是 str: {key!r}") + if key.startswith(_META_RESERVED_PREFIX): + raise ValueError( + f"{origin} 的 meta 键 {key!r} 使用了保留前缀 {_META_RESERVED_PREFIX!r}" + "(留给库将来的内建维度,避免与调用方的键撞名)" + ) + if not _META_KEY_RE.fullmatch(key): + raise ValueError( + f"{origin} 的 meta 键 {key!r} 不合法: 只允许小写字母/数字/下划线/点,长度 1-64" + ) + if len(meta) > _META_MAX_KEYS: + raise ValueError(f"{origin} 的 meta 键数超限(上限 {_META_MAX_KEYS}): {len(meta)}") + + +def _validate_meta_values(meta: Mapping[str, Any], origin: str) -> None: + """值只收扁平标量;非有限 float 必须挡在这里。 + + `json.dumps` 会把 `nan`/`inf` 写成 `NaN`/`Infinity` 字面量——不是合法 JSON, + PG 的 JSONB 拒收。放行则写入失败会被遥测的降级 try 吞成 warning,即把调用方 + 的输入错误转成静默丢遥测(Codex 审查推翻了初稿"序列化不可达"的论断)。 + """ + for key, value in meta.items(): + if not isinstance(value, (str, int, float, bool)): + raise ValueError( + f"{origin} 的 meta 值必须是 str/int/float/bool: {key}={value!r}" + "(嵌套结构请调用方自行序列化)" + ) + if isinstance(value, float) and not math.isfinite(value): + raise ValueError(f"{origin} 的 meta 值不得是 nan/inf: {key}={value!r}(非合法 JSON)") + if isinstance(value, str) and len(value) > _META_VALUE_MAX_LEN: + raise ValueError( + f"{origin} 的 meta 值超长(上限 {_META_VALUE_MAX_LEN}): {key} 长 {len(value)}" + ) + + def merge_sampling(extra_body: Mapping[str, Any], sampling: Mapping[str, Any]) -> dict[str, Any]: """合并配置级与调用级采样参数;调用级优先(issue #4 设计决策 A)。""" return {**extra_body, **sampling} @@ -129,6 +222,21 @@ class ChatRequest: 不同深度取值不同;缓存 key 与三个遥测入口需要一个跨层恒定的读取点,否则 同一列在不同行口径分叉。""" + # —— 调用方自定义维度(issue #11;追加在末尾,不扰动既有字段的位置构造)—— + tenant_id: str | None = None + """调用方的租户标识,进遥测的 `tenant_id` 真实列(issue #11)。 + + 独立成字段而非混进 `meta`,因为它是唯一享有真实列待遇的维度——可挂 RLS、 + 可进复合索引。混在 `meta` 里则调用方拼错(`tenantId`)不会报错,只会静默 + 降级成一个普通维度,正是本 issue 抱怨的失败形态。""" + + meta: Mapping[str, Any] = field(default_factory=dict) + """调用方自定义维度的只读快照,库不解释其含义,库内中间件**永不修改**。 + + **不进缓存 key**: 租户隔离已由 `cache_namespace` 负责并已进 key(ARCH §7.5), + 再进一次既重复又会让存量缓存全量冷启动;且 `meta` 承载的是审计维度而非 + 语义维度,同 messages 同 namespace 下换个 batch_id 不应导致 miss。""" + @dataclass(frozen=True) class Usage: diff --git a/tests/unit/test_types.py b/tests/unit/test_types.py index de288c1..5d4325f 100644 --- a/tests/unit/test_types.py +++ b/tests/unit/test_types.py @@ -398,3 +398,132 @@ class TestSourceConfigExtraBody: """ with pytest.raises(TypeError): hash(_make_source()) + + +class TestCallerDimensionsValidation: + """调用方自定义维度的入口校验(issue #11 设计 §4.2)。 + + 这些红线全部要求**报错**而非静默丢弃: Langfuse 对超长 value 的做法是 + 直接丢掉,本库不抄——P5 严禁默认值掩盖错误。且报错点必须在进洋葱之前, + 洋葱内的一切失败都会被遥测层降级成 warning,校验放那里等于没有校验。 + """ + + @pytest.mark.parametrize( + "bad", + [ + "", # 空串是哨兵值的地盘(老行/未归属) + " ", # 纯空白 strip 后为空 + " t1", # 首尾空白: 与 "t1" 在 RLS 等值比较下是两个租户 + "t1 ", + "x" * 129, # 上限 128 + 123, # 非 str + ], + ) + def test_bad_tenant_id_rejected(self, bad): + """租户标识形态错误必须当场报错,而非带着走到落库。""" + from polygateway.types import validate_caller_dimensions + + with pytest.raises(ValueError) as exc: + validate_caller_dimensions(bad, None, origin="chat(tenant_id=...)") + assert "chat(tenant_id=...)" in str(exc.value) # 信息须能定位来源 + + @pytest.mark.parametrize( + "key", + [ + "Batch", # 大写不合字符集 + "batch-id", # 连字符不合字符集 + "b" * 65, # 键长上限 64 + "pg_internal", # 保留前缀 + "", # 空键 + ], + ) + def test_bad_meta_key_rejected(self, key): + """键集合被假定为低基数且稳定,形态必须收紧(OTel semconv 字符集)。""" + from polygateway.types import validate_caller_dimensions + + with pytest.raises(ValueError, match="meta"): + validate_caller_dimensions(None, {key: "v"}, origin="test") + + def test_non_str_meta_key_rejected(self): + """非 str 键无法进 JSON 对象,须先于值校验报出键的问题。""" + from polygateway.types import validate_caller_dimensions + + with pytest.raises(ValueError, match="str"): + validate_caller_dimensions(None, {1: "a"}, origin="test") + + def test_too_many_meta_keys_rejected(self): + """上限 16: 容器是审计维度,不是给调用方塞整个请求体的地方。""" + from polygateway.types import validate_caller_dimensions + + with pytest.raises(ValueError, match="16"): + validate_caller_dimensions(None, {f"k{i}": "v" for i in range(17)}, origin="test") + + @pytest.mark.parametrize("bad", [["a"], {"a": 1}, None, object()]) + def test_non_scalar_meta_value_rejected(self, bad): + """只收扁平标量(OTel AnyValue 的可移植子集);嵌套让调用方自己序列化。""" + from polygateway.types import validate_caller_dimensions + + with pytest.raises(ValueError, match="值"): + validate_caller_dimensions(None, {"k": bad}, origin="test") + + @pytest.mark.parametrize("bad", [float("nan"), float("inf"), float("-inf")]) + def test_non_finite_float_rejected(self, bad): + """json.dumps 会把它们写成 NaN/Infinity 字面量——非合法 JSON,PG JSONB 拒收。 + + 放行则调用方的输入错误会在写入层失败、被遥测降级吞成 warning, + 即"输入错误"静默变成"丢遥测"(Codex 审查推翻了初稿的"不可达"论断)。 + """ + from polygateway.types import validate_caller_dimensions + + with pytest.raises(ValueError): + validate_caller_dimensions(None, {"k": bad}, origin="test") + + def test_too_long_meta_value_rejected(self): + """字符串值上限 256(对齐 Sentry tag / Langfuse 的量级)。""" + from polygateway.types import validate_caller_dimensions + + with pytest.raises(ValueError): + validate_caller_dimensions(None, {"k": "v" * 257}, origin="test") + + def test_absent_dimensions_pass_through(self): + """两者都不传是绝大多数调用点的现状,必须零摩擦放行。""" + from polygateway.types import validate_caller_dimensions + + assert validate_caller_dimensions(None, None, origin="test") == (None, {}) + + @pytest.mark.parametrize("good", ["s", 1, 1.5, True, False, 0]) + def test_scalar_meta_values_accepted(self, good): + """bool 是 int 子类,两者都合法;0/False 不得被真值判断误杀。""" + from polygateway.types import validate_caller_dimensions + + _, meta = validate_caller_dimensions(None, {"k": good}, origin="test") + assert meta == {"k": good} + + def test_returns_independent_copy(self): + """调用方复用同一 dict 逐次改值是预期模式,不拷贝会有竞态(同 overlay 决策 E)。""" + from polygateway.types import validate_caller_dimensions + + caller_dict = {"batch": "b-42"} + _, meta = validate_caller_dimensions("t1", caller_dict, origin="test") + caller_dict["batch"] = "b-99" + assert meta == {"batch": "b-42"} + + +class TestChatRequestDimensions: + """ChatRequest 承载维度的字段契约(issue #11)。""" + + def test_defaults_are_absent_dimensions(self): + """新字段必须带默认值——三项目逐字段构造的 fake 才能零改动(ARCH §5.1 约定①)。""" + request = ChatRequest(messages=[{"role": "user", "content": "x"}]) + assert request.tenant_id is None + assert request.meta == {} + + def test_dimensions_are_carried(self): + """维度随请求在洋葱内流转,是缓存 key 之外三个遥测入口的共同读取点。""" + request = ChatRequest( + messages=[{"role": "user", "content": "x"}], + tenant_id="t1", + meta={"batch": "b-42"}, + ) + assert request.tenant_id == "t1" + assert request.meta == {"batch": "b-42"}