feat: record each call's tenant and caller-defined dimensions

Both telemetry backends gain tenant_id and meta at the end of the
column list, and TelemetryEmitter fills them from the request. The two
halves ship together because the emitter is the only caller of
record_llm_call: adding the columns without filling them leaves every
row short of two keys, and the backends read those keys outside their
try block, so the KeyError degrades to a warning and the whole table
stops filling.

The columns are appended, never inserted. An old table can only gain
columns through ALTER, which puts them last; a new table built from the
DDL would put them wherever the DDL says. Anywhere but the end and the
two paths produce different physical column orders, while the INSERT
uses positional placeholders.

The two backends spell the default differently for different reasons.
SQLite refuses a NOT NULL column without a non-NULL constant default
outright, so the default is what makes the backfill legal at all. On
Postgres a non-volatile constant default is what keeps the ALTER from
rewriting the table, and NOT NULL DEFAULT '' is what keeps old rows out
of the black hole a NULL tenant_id falls into under an RLS policy.

Normalisation happens in the emitter, not the recorder, matching how
canonical_sampling_json already settles the sampling column: None
becomes the empty string, an empty mapping becomes the literal '{}'.
Keys are sorted so one set of dimensions serialises identically on
every row, and allow_nan=False is a second gate behind the entry
validation -- json.dumps would otherwise write a bare NaN, which JSONB
rejects, and the failed insert would be swallowed as a warning.

All three emit entry points read the request. Cache hits read it too,
rather than the replayed response: the dimensions answer who made this
call, not who made the one whose result is being replayed.

src/polygateway/middleware/telemetry.py

@@ -26,13 +26,30 @@ from polygateway.middleware.cache import digest_messages
 from polygateway.types import canonical_sampling_json, merge_sampling
 
 if TYPE_CHECKING:
-    from collections.abc import Callable
+    from collections.abc import Callable, Mapping
+    from typing import Any
 
     from polygateway.ports import CallNext, TelemetryRecorder
     from polygateway.pricing import PricingTable
     from polygateway.types import ChatRequest, LLMResponse, SourceConfig
 
 
+def _canonical_meta_json(meta: Mapping[str, Any]) -> str:
+    """把调用方自定义维度定型为 JSON 文本(issue #11);空 dict 落字面量 `'{}'`。
+
+    `sort_keys=True` 让同一份维度在任意两行里字节一致,可直接等值比对与去重;
+    `ensure_ascii=False` 保留中文原文,避免落库成 `\\uXXXX` 串而无法肉眼审计。
+
+    `allow_nan=False` 是**第二道闸**(主防线是 `types.validate_caller_dimensions`
+    在公共入口的校验): `json.dumps` 默认把 `nan` 写成裸 `NaN` 字面量,那不是合法
+    JSON,PG 的 JSONB 会拒收;而写入失败会被 `_record` 的降级 try 吞成 warning,
+    等于把调用方的输入错误转化成静默丢遥测。宁可在这里显式抛。
+    """
+    if not meta:
+        return "{}"
+    return json.dumps(dict(meta), sort_keys=True, ensure_ascii=False, allow_nan=False)
+
+
 @dataclass(frozen=True)
 class _AttemptUsage:
     """一次尝试的用量视图;默认值即"失败尝试"档(无用量可言,记 0 并标 unavailable)。
@@ -72,7 +89,7 @@ class _AttemptUsage:
 
 
 class TelemetryEmitter:
-    """从请求与结果组装 21 字段并写入 recorder;一切写失败降级 warning。"""
+    """从请求与结果组装 24 字段并写入 recorder;一切写失败降级 warning。"""
 
     def __init__(self, recorder: TelemetryRecorder, *, pricing: PricingTable | None = None) -> None:
         self._recorder = recorder
@@ -111,6 +128,8 @@ class TelemetryEmitter:
             reasoning_tokens=usage.reasoning_tokens,
             # 唯一有"生效源"的入口,故是唯一能并上 extra_body 的(设计决策 D)
             sampling=canonical_sampling_json(merge_sampling(source.extra_body, request.sampling)),
+            tenant_id=request.tenant_id,
+            meta=request.meta,
         )
 
     async def emit_cache_hit(self, *, request: ChatRequest, response: LLMResponse) -> None:
@@ -139,6 +158,11 @@ class TelemetryEmitter:
             # 由最外层 TelemetryMW 调用,手上没有 source。缓存命中行无损:
             # sampling 已进缓存 key,能命中即意味调用级参数与历史那次逐字相同
             sampling=canonical_sampling_json(request.sampling),
+            # 与上面的 model/prompt_tokens 相反,维度读 request 而非 response:
+            # 维度回答的是"本次调用由谁发起",不是历史那次。读历史会把本次调用
+            # 记到上一个租户头上,两边的账同时错且无任何报错(issue #11 设计 §4.3)
+            tenant_id=request.tenant_id,
+            meta=request.meta,
         )
 
     async def emit_terminal_failure(
@@ -166,6 +190,9 @@ class TelemetryEmitter:
             reasoning_tokens=None,
             # 无具体源,与 model/provider/source_name 置空同一先例(设计决策 D)
             sampling=canonical_sampling_json(request.sampling),
+            # 源不可知,但租户归属是已知的——终态失败行恰是审计最需要的
+            tenant_id=request.tenant_id,
+            meta=request.meta,
         )
 
     async def _record(
@@ -190,6 +217,9 @@ class TelemetryEmitter:
         model_reported: str | None,
         sampling: str | None,
         reasoning_tokens: int | None,
+        # issue #11: 未归一化的调用方维度,归一化在本方法内收口(recorder 只落库)
+        tenant_id: str | None,
+        meta: Mapping[str, Any],
     ) -> None:
         try:
             # 成本换算(M2 §6): 成功行按单价换算;缓存命中 0.0(未产生新调用);
@@ -231,6 +261,10 @@ class TelemetryEmitter:
                 model_reported=model_reported,
                 sampling=sampling,
                 reasoning_tokens=reasoning_tokens,
+                # 空串是哨兵而非 NULL: NULL 的 tenant_id 在 PG 的 RLS policy 下
+                # 对所有人永久不可见,空串则可用一条 SQL 审计出未归属的行
+                tenant_id=tenant_id or "",
+                meta=_canonical_meta_json(meta),
             )
         except asyncio.CancelledError:
             raise

src/polygateway/ports.py

@@ -245,10 +245,15 @@ class StructuredOutputStrategy(Protocol):
 
 @runtime_checkable
 class TelemetryRecorder(Protocol):
-    """遥测后端;22 字段冻结(M1 设计 §4.4 + issue #3/#4),唯一调用点是 TelemetryEmitter。
+    """遥测后端;24 字段冻结(M1 设计 §4.4 + issue #3/#4/#11),唯一调用点是 TelemetryEmitter。
 
     新增参数不设默认值: 库外无第三方实现者(三项目迁移时删除了各自的同名
     Protocol),完整签名的成本为零,而少写一列会被 emitter 的降级吞成 warning。
+
+    `tenant_id` 与 `meta` 到达 recorder 时**已由 emitter 归一化**——`tenant_id`
+    的 `None` 已转空串,`meta` 已序列化为 JSON 字符串(空 dict 为 `'{}'`)。
+    recorder 只负责落库,不做任何语义判断,与 `sampling` 列由
+    `canonical_sampling_json()` 在 emitter 侧定型是同一先例。
     """
 
     async def record_llm_call(
@@ -276,4 +281,6 @@ class TelemetryRecorder(Protocol):
         model_reported: str | None,
         sampling: str | None,
         reasoning_tokens: int | None,
+        tenant_id: str,
+        meta: str,
     ) -> None: ...

src/polygateway/telemetry/postgres.py

@@ -48,7 +48,9 @@ CREATE TABLE IF NOT EXISTS llm_calls (
     cached_prompt_tokens INTEGER,
     model_reported       TEXT,
     sampling             TEXT,
-    reasoning_tokens     INTEGER
+    reasoning_tokens     INTEGER,
+    tenant_id            TEXT NOT NULL DEFAULT '',
+    meta                 JSONB NOT NULL DEFAULT '{}'::jsonb
 );
 """
 
@@ -58,6 +60,15 @@ _BACKFILL = (
     ("model_reported", "ALTER TABLE llm_calls ADD COLUMN model_reported TEXT"),
     ("sampling", "ALTER TABLE llm_calls ADD COLUMN sampling TEXT"),
     ("reasoning_tokens", "ALTER TABLE llm_calls ADD COLUMN reasoning_tokens INTEGER"),
+    # 两个默认值都是非易失常量,PG 11+ 只改 catalog 不重写全表,故大表补列亦是秒级
+    (
+        "tenant_id",
+        "ALTER TABLE llm_calls ADD COLUMN tenant_id TEXT NOT NULL DEFAULT ''",
+    ),
+    (
+        "meta",
+        "ALTER TABLE llm_calls ADD COLUMN meta JSONB NOT NULL DEFAULT '{}'::jsonb",
+    ),
 )
 
 # 探测表是否存在;不需要任何权限,且与 INSERT 走同一套 search_path 解析
@@ -92,6 +103,8 @@ _COLUMNS = (
     "model_reported",
     "sampling",
     "reasoning_tokens",
+    "tenant_id",
+    "meta",
 )
 
 _INSERT = (

src/polygateway/telemetry/sqlite.py

@@ -46,7 +46,9 @@ CREATE TABLE IF NOT EXISTS llm_calls (
     cached_prompt_tokens INTEGER,
     model_reported       TEXT,
     sampling             TEXT,
-    reasoning_tokens     INTEGER
+    reasoning_tokens     INTEGER,
+    tenant_id            TEXT    NOT NULL DEFAULT '',
+    meta                 TEXT    NOT NULL DEFAULT '{}'
 );
 """
 
@@ -57,6 +59,10 @@ _BACKFILL_COLUMNS = (
     ("model_reported", "TEXT"),
     ("sampling", "TEXT"),
     ("reasoning_tokens", "INTEGER"),
+    # NOT NULL 补列必须带非 NULL 常量默认值,否则 SQLite 直接拒绝该 ALTER
+    # ("Cannot add a NOT NULL column with default value NULL"),补列全盘失败。
+    ("tenant_id", "TEXT NOT NULL DEFAULT ''"),
+    ("meta", "TEXT NOT NULL DEFAULT '{}'"),
 )
 
 _COLUMNS = (
@@ -82,6 +88,8 @@ _COLUMNS = (
     "model_reported",
     "sampling",
     "reasoning_tokens",
+    "tenant_id",
+    "meta",
 )
 
 _INSERT = (
@@ -137,7 +145,7 @@ class SQLiteRecorder:
                     logger.warning("SQLite 遥测补列失败(写入将逐行降级): {}", exc)
 
     async def record_llm_call(self, **fields: object) -> None:
-        """写一行遥测;字段集合即 21 字段冻结签名(ports.TelemetryRecorder)。"""
+        """写一行遥测;字段集合即 24 字段冻结签名(ports.TelemetryRecorder)。"""
         if self._conn is None:
             return
         row = tuple(fields[col] for col in _COLUMNS)

tests/integration/test_postgres_telemetry.py

@@ -45,6 +45,8 @@ _EXPECTED_COLUMNS = [
     "model_reported",
     "sampling",
     "reasoning_tokens",
+    "tenant_id",
+    "meta",
 ]
 
 # run 级前缀: 同库并存的其他运行(迁移批跑/另一开发机)互不可见
@@ -110,6 +112,9 @@ async def _record_minimal(
         "model_reported": None,
         "sampling": None,
         "reasoning_tokens": None,
+        # 到达 recorder 时已由 emitter 归一化: None → '',空 dict → '{}'
+        "tenant_id": "",
+        "meta": "{}",
     }
     fields.update(overrides)
     await recorder.record_llm_call(**fields)

tests/unit/test_ports.py

@@ -118,6 +118,8 @@ class _DummyRecorder:
         model_reported,
         sampling,
         reasoning_tokens,
+        tenant_id,
+        meta,
     ) -> None: ...
 
 
@@ -205,6 +207,29 @@ class TestGateUpdate:
             )
 
 
+class TestTelemetryRecorderSignature:
+    """`record_llm_call` 的冻结签名以 `inspect.signature` 实测,不凭记忆断言。
+
+    该 Protocol 的纪律是新增参数**不设默认值**(ports.py docstring):库外无第三方
+    实现者,而带默认值的参数会让 emitter 漏传时静默落默认值——遥测里的租户归属
+    一旦静默错位,事后无从分辨是"没传"还是"就是空的"。
+    """
+
+    def test_caller_dimensions_are_declared(self):
+        import inspect
+
+        params = inspect.signature(TelemetryRecorder.record_llm_call).parameters
+        assert {"tenant_id", "meta"} <= set(params)
+
+    @pytest.mark.parametrize("name", ["tenant_id", "meta"])
+    def test_caller_dimensions_have_no_default(self, name):
+        import inspect
+
+        param = inspect.signature(TelemetryRecorder.record_llm_call).parameters[name]
+        assert param.default is inspect.Parameter.empty
+        assert param.kind is inspect.Parameter.KEYWORD_ONLY
+
+
 class TestOcrPorts:
     """M3 三个 OCR Protocol(设计 §3.2): runtime_checkable 结构判定。"""
 

tests/unit/test_telemetry.py

@@ -40,6 +40,8 @@ _EXPECTED_COLUMNS = [
     "model_reported",
     "sampling",
     "reasoning_tokens",
+    "tenant_id",
+    "meta",
 ]
 
 
@@ -104,11 +106,38 @@ async def _record_minimal(recorder, call_id="c1", **overrides):
         "model_reported": None,
         "sampling": None,
         "reasoning_tokens": None,
+        # 到达 recorder 时已由 emitter 归一化: None → '',空 dict → '{}'
+        "tenant_id": "",
+        "meta": "{}",
     }
     fields.update(overrides)
     await recorder.record_llm_call(**fields)
 
 
+class TestBackendColumnParity:
+    """两个后端的 `_COLUMNS` 必须逐字同名同序(issue #11)。
+
+    emitter 只组装一份 `fields`,两个后端各自按自己的 `_COLUMNS` 取值;两份清单
+    一旦分叉,同一次调用在 SQLite 上写得进、在 PG 上抛 KeyError 被降级吞掉,
+    差异只在换后端时才暴露。列**序**同样断言: INSERT 用位置占位符,顺序错位
+    会把值写进错误的列而不报错。
+    """
+
+    def test_two_backends_agree_on_columns(self):
+        from polygateway.telemetry.postgres import _COLUMNS as PG_COLUMNS
+        from polygateway.telemetry.sqlite import _COLUMNS as SQLITE_COLUMNS
+
+        assert SQLITE_COLUMNS == PG_COLUMNS
+
+    def test_caller_dimensions_are_appended_last(self):
+        """新列只能追加在末尾: 旧表经 ALTER 补列必落末尾,插在中间会让两条路径分叉。"""
+        from polygateway.telemetry.postgres import _COLUMNS as PG_COLUMNS
+        from polygateway.telemetry.sqlite import _COLUMNS as SQLITE_COLUMNS
+
+        assert SQLITE_COLUMNS[-2:] == ("tenant_id", "meta")
+        assert PG_COLUMNS[-2:] == ("tenant_id", "meta")
+
+
 class TestSQLiteRecorder:
     async def test_schema_has_frozen_columns(self, tmp_path):
         recorder = SQLiteRecorder(tmp_path / "t.db")
@@ -327,6 +356,8 @@ class TestPostgresBackfillDiscipline:
         "model_reported",
         "sampling",
         "reasoning_tokens",
+        "tenant_id",
+        "meta",
     ]
 
     def _recorder(self, conn):
@@ -379,6 +410,8 @@ class TestPostgresTableProbe:
         "model_reported",
         "sampling",
         "reasoning_tokens",
+        "tenant_id",
+        "meta",
     ]
 
     def _recorder(self, conn):
@@ -613,6 +646,110 @@ class TestEmitterSamplingColumn:
         assert rec.rows[0]["sampling"] is None
 
 
+class TestEmitterCallerDimensions:
+    """issue #11: 三个 emit 入口统一从 `request` 读维度,`_record` 落库前归一化。
+
+    维度只有一个读取点(`request`),否则同一列在三种行里口径分叉——那正是
+    "遥测调用点收敛为单一 helper"这条铁律要防的形态。
+    """
+
+    _META = {"z_last": "z", "a_first": 1, "m_mid": True}
+    _REQ_A = ChatRequest(
+        messages=[{"role": "user", "content": "hi"}],
+        session_id="sess-1",
+        tenant_id="tenant-a",
+        meta=_META,
+    )
+
+    @pytest.mark.parametrize("emit", ["attempt", "cache_hit", "terminal_failure"])
+    async def test_every_entry_point_carries_the_dimensions(self, emit):
+        """三条路径写出的行都必须带维度: 漏掉任一条,该租户的账就永远对不上。"""
+        rec = _MemoryRecorder()
+        emitter = TelemetryEmitter(rec)
+        if emit == "attempt":
+            await emitter.emit_attempt(
+                request=self._REQ_A,
+                source=_source(),
+                call_id="c",
+                latency_ms=1,
+                response=_resp(),
+                error=None,
+            )
+        elif emit == "cache_hit":
+            await emitter.emit_cache_hit(request=self._REQ_A, response=_resp(cache_hit=True))
+        else:
+            await emitter.emit_terminal_failure(
+                request=self._REQ_A, call_id="c", latency_ms=1, error="dead"
+            )
+        row = rec.rows[0]
+        assert row["tenant_id"] == "tenant-a"
+        assert json.loads(row["meta"]) == self._META
+
+    async def test_cache_hit_records_the_current_caller_not_the_cached_one(self):
+        """缓存命中行的维度是"本次由谁发起",不是历史那次——最容易实现反的一处。
+
+        历史那次由租户 B 发起并把响应留在了缓存里;本次由租户 A 发起并命中。
+        若读了历史那次的归属,租户 A 的调用会记到 B 头上,而 A 的账面凭空少一行
+        ——两个租户的账同时错,且错得没有任何报错。
+        """
+        historical = ChatRequest(
+            messages=[{"role": "user", "content": "hi"}],
+            tenant_id="tenant-b",
+            meta={"batch": "old-batch"},
+        )
+        rec = _MemoryRecorder()
+        mw = TelemetryMW(TelemetryEmitter(rec))
+
+        async def terminal(request):
+            # 缓存层回放的是历史那次的响应对象(其 call_id 属于 historical 那次)
+            return _resp(cache_hit=True, latency_ms=0, call_id="cache-cid")
+
+        assert historical.tenant_id == "tenant-b"  # 历史归属确实不同,否则本用例是空转
+        await mw(self._REQ_A, terminal)
+
+        row = rec.rows[0]
+        assert row["cache_hit"] is True
+        assert row["tenant_id"] == "tenant-a"
+        assert "old-batch" not in row["meta"]
+
+    async def test_absent_dimensions_land_as_sentinels(self):
+        """未传维度落哨兵值: `tenant_id` 空串、`meta` 字面量 `'{}'`,都不是 NULL。
+
+        NULL 的 `tenant_id` 在 PG 的 RLS policy 下对所有人永久不可见(设计 §4.4),
+        空串则可用一条 SQL 审计出还有多少行未归属;`meta` 同理,`'{}'` 可被
+        JSON 函数直接查询,NULL 则要每条查询都额外判空。
+        """
+        rec = _MemoryRecorder()
+        await TelemetryEmitter(rec).emit_attempt(
+            request=_REQ,  # tenant_id=None, meta={}
+            source=_source(),
+            call_id="c",
+            latency_ms=1,
+            response=_resp(),
+            error=None,
+        )
+        row = rec.rows[0]
+        assert row["tenant_id"] == ""
+        assert row["meta"] == "{}"
+
+    async def test_meta_is_serialized_with_sorted_keys(self):
+        """键序固定,同一份维度在任意两行里字节一致,可直接做等值比对与去重。"""
+        rec = _MemoryRecorder()
+        await TelemetryEmitter(rec).emit_terminal_failure(
+            request=self._REQ_A, call_id="c", latency_ms=1, error="dead"
+        )
+        assert list(json.loads(rec.rows[0]["meta"])) == ["a_first", "m_mid", "z_last"]
+
+    async def test_non_ascii_meta_stays_readable(self):
+        """`ensure_ascii=False`: 中文维度按原文落库,而非 `\\uXXXX` 转义串。"""
+        rec = _MemoryRecorder()
+        req = ChatRequest(messages=[{"role": "user", "content": "hi"}], meta={"dept": "研发"})
+        await TelemetryEmitter(rec).emit_terminal_failure(
+            request=req, call_id="c", latency_ms=1, error="dead"
+        )
+        assert "研发" in rec.rows[0]["meta"]
+
+
 class TestCostWithCachedTier:
     """issue #3: 命中部分按缓存单价计费,避免 cost 系统性高估。"""