diff --git a/src/polygateway/__init__.py b/src/polygateway/__init__.py
index 6254135..4043c5a 100644
--- a/src/polygateway/__init__.py
+++ b/src/polygateway/__init__.py
@@ -39,6 +39,7 @@ from polygateway.thinking import (
 )
 from polygateway.types import (
     EFFORT_ORDER,
+    CallStats,
     Effort,
     EmbeddingResponse,
     LLMResponse,
@@ -57,6 +58,7 @@ __all__ = [
     "EFFORT_ORDER",
     "Effort",
     "AllSourcesExhausted",
+    "CallStats",
     "CircuitOpenError",
     "EmbeddingClient",
     "EmbeddingResponse",
diff --git a/src/polygateway/client.py b/src/polygateway/client.py
index c5f347c..9ac1a64 100644
--- a/src/polygateway/client.py
+++ b/src/polygateway/client.py
@@ -9,6 +9,7 @@
 from __future__ import annotations
 
 import asyncio
+import dataclasses
 import hashlib
 import json
 import random
@@ -46,6 +47,7 @@ from polygateway.types import (
     Effort,
     LLMResponse,
     TelemetryStatus,
+    _CallContext,
     coerce_effort,
     validate_caller_dimensions,
     validate_request_overlay,
@@ -289,6 +291,10 @@ class GatewayClient:
         self._structured_available = structured_strategy is not None
         self._terminal = terminal  # 内部引用: 装配自省/测试用
         self._handler = compose(middlewares, terminal)
+        # 逻辑调用统计需要同一只注入钟(1.3.5);现之前只传给中间件未自存
+        self._now = now
+        # 终态行由公开边界统一写出(T3),故边界也需持有 emitter
+        self._emitter = emitter
         self._transport = transport
         self._telemetry = telemetry
         self._cache = cache
@@ -370,6 +376,8 @@ class GatewayClient:
             else coerce_effort(reasoning_effort, origin="chat(reasoning_effort=...)")
         )
         validate_thinking_raw(sampling, effort=effort, wire=None, origin="chat overlay")
+        # 三项校验均已通过 → 进入统计边界(设计 §3: 输入校验异常在边界之外,保持原行为)
+        context = _CallContext(now=self._now)
         request = ChatRequest(
             messages=messages,
             session_id=session_id,
@@ -383,8 +391,11 @@ class GatewayClient:
             reasoning_effort=effort,
             tenant_id=dimension_tenant_id,
             meta=dimensions,
+            call_context=context,
         )
-        return await self._handler(request)
+        response = await self._handler(request)
+        # 快照在返回前冻结: 故它含缓存命中路径与已完成的内联遥测耗时
+        return dataclasses.replace(response, call_stats=context.snapshot())
 
     async def aclose(self) -> None:
         """幂等释放**自建**资源: transport、遥测、缓存、限流/熔断后端。
diff --git a/src/polygateway/embedding.py b/src/polygateway/embedding.py
index ac042ca..44ac701 100644
--- a/src/polygateway/embedding.py
+++ b/src/polygateway/embedding.py
@@ -16,6 +16,7 @@
 from __future__ import annotations
 
 import asyncio
+import dataclasses
 import math
 import random
 import time
@@ -47,6 +48,7 @@ from polygateway.types import (
     EmbeddingResponse,
     LLMResponse,
     TelemetryStatus,
+    _CallContext,
     strip_unsupported_extra_body,
     validate_caller_dimensions,
 )
@@ -184,7 +186,11 @@ class EmbeddingClient:
         dimension_tenant_id, dimensions = validate_caller_dimensions(
             tenant_id, meta, origin="embed(tenant_id=..., meta=...)"
         )
+        # 校验均已通过 → 进入统计边界(设计 §3.5: `texts` 类型与调用方维度校验之后)
+        context = _CallContext(now=self._now)
         if not texts:
+            # 合法零尝试: 返回真实统计(attempts=0),且**不写任何遥测行**
+            # ——与 cache_hit 不同,不要按"遥测必录"推断它有台账行(设计 §3 M2)
             return EmbeddingResponse(
                 vectors=[],
                 dim=0,
@@ -195,6 +201,7 @@ class EmbeddingClient:
                 latency_ms=0,
                 call_id=str(uuid.uuid4()),
                 source_name="",
+                call_stats=context.snapshot(),
             )
         if not self._sources:
             raise AllSourcesExhausted(scope=self._scope, reason="no_sources", retry_after_s=0.0)
@@ -207,9 +214,11 @@ class EmbeddingClient:
                     parent_call_id,
                     dimension_tenant_id,
                     dimensions,
+                    context,
                 )
             )
-        return self._merge(outcomes)
+        # 全批共享同一上下文,故分批是实现细节而非 N 次独立逻辑调用
+        return dataclasses.replace(self._merge(outcomes), call_stats=context.snapshot())
 
     # —— 治理循环(与 RetryMW 同构;设计 §7.1 已声明的有限重复)——
 
@@ -220,6 +229,7 @@ class EmbeddingClient:
         parent_call_id: str | None,
         tenant_id: str | None,
         meta: dict[str, Any],
+        context: _CallContext,
     ) -> _BatchOutcome:
         fails = 0
         reasons: dict[str, str] = {}
@@ -232,7 +242,7 @@ class EmbeddingClient:
                 continue
             async with clock.attempting():
                 outcome = await self._attempt(
-                    batch, *picked, reasons, session_id, parent_call_id, tenant_id, meta
+                    batch, *picked, reasons, session_id, parent_call_id, tenant_id, meta, context
                 )
             if isinstance(outcome, _BatchOutcome):
                 return outcome
@@ -258,10 +268,13 @@ class EmbeddingClient:
         parent_call_id: str | None,
         tenant_id: str | None,
         meta: dict[str, Any],
+        context: _CallContext,
     ) -> _BatchOutcome | _FailedBatch:
         call_id = str(uuid.uuid4())
         started = self._now()
         actual = 0
+        # 登记在 transport 调用**之前**(同 RetryMW): 失败与取消的尝试也真的发出去了
+        context.register_attempt()
         try:
             result = await self._transport.embed(texts=batch, source=source, call_id=call_id)
             if self._expected_dim is not None and result.dim != self._expected_dim:
diff --git a/src/polygateway/middleware/cache.py b/src/polygateway/middleware/cache.py
index abdc2e7..af60be4 100644
--- a/src/polygateway/middleware/cache.py
+++ b/src/polygateway/middleware/cache.py
@@ -208,6 +208,10 @@ class CacheMW:
                 max_inter_token_ms=None,
                 call_id=str(uuid.uuid4()),
                 structured_data=structured_data,
+                # 显式覆盖: 历史条目里的 `call_stats` 是个 dict,而 `_RESPONSE_FIELDS`
+                # 过滤**会放行它**——不覆盖就会有 dict 冒充 `CallStats` 漏给调用方。
+                # 本次调用的真实统计由公开边界在返回前追加(设计 §3)
+                call_stats=None,
             )
             return LLMResponse(**fields)
         except Exception as exc:
@@ -230,6 +234,9 @@ class CacheMW:
     def _serialize(self, response: LLMResponse) -> str:
         data = dataclasses.asdict(response)
         data.pop("structured_data", None)  # pydantic 实例不可 JSON 往返(设计 §2.1)
+        # 统计描述**本次**调用,存进去再放出来等于向下一个调用方谎称
+        # 它重试了 N 次;`asdict` 会把 `CallStats` 摊成 dict,故必须显式剔除
+        data.pop("call_stats", None)
         return json.dumps(data, ensure_ascii=False)
 
     async def _safe_get(self, key: str) -> str | None:
diff --git a/src/polygateway/middleware/retry.py b/src/polygateway/middleware/retry.py
index 09a0236..0a3a1ea 100644
--- a/src/polygateway/middleware/retry.py
+++ b/src/polygateway/middleware/retry.py
@@ -278,6 +278,11 @@ class RetryMW:
         call_id = str(uuid.uuid4())
         started = self._now()
         actual = 0
+        # 登记在 transport 调用**之前**(1.3.5 设计 §4): 失败与取消的尝试同样
+        # "真的打出去了",挪到成功之后会让诊断最需要看见的那几次从计数里消失。
+        # 上下文为 None = 库内现场构造的请求,跳过而不是报错
+        if request.call_context is not None:
+            request.call_context.register_attempt()
         try:
             result = await self._transport.complete(
                 messages=request.messages,
diff --git a/src/polygateway/ocr.py b/src/polygateway/ocr.py
index 7dc0d35..683649f 100644
--- a/src/polygateway/ocr.py
+++ b/src/polygateway/ocr.py
@@ -40,12 +40,14 @@ from polygateway.middleware.retry import StallClock, _failure_reason, backoff_de
 from polygateway.middleware.telemetry import TelemetryEmitter
 from polygateway.ports import OutcomeAwareSelector
 from polygateway.types import (
+    CallStats,
     ChatRequest,
     LLMResponse,
     OcrLayoutResult,
     OcrTextResult,
     TelemetryStatus,
     Usage,
+    _CallContext,
     strip_unsupported_extra_body,
     validate_caller_dimensions,
 )
@@ -176,7 +178,7 @@ class OcrClient:
         dimension_tenant_id, dimensions = validate_caller_dimensions(
             tenant_id, meta, origin="recognize_text(tenant_id=..., meta=...)"
         )
-        outcome = await self._call(
+        outcome, call_stats = await self._call(
             "text", image, session_id, parent_call_id, dimension_tenant_id, dimensions
         )
         result = outcome.result
@@ -187,6 +189,7 @@ class OcrClient:
             latency_ms=outcome.latency_ms,
             call_id=outcome.call_id,
             raw=result.raw,
+            call_stats=call_stats,
         )
 
     async def parse_layout(
@@ -206,7 +209,7 @@ class OcrClient:
         dimension_tenant_id, dimensions = validate_caller_dimensions(
             tenant_id, meta, origin="parse_layout(tenant_id=..., meta=...)"
         )
-        outcome = await self._call(
+        outcome, call_stats = await self._call(
             "layout", image, session_id, parent_call_id, dimension_tenant_id, dimensions
         )
         result = outcome.result
@@ -218,6 +221,7 @@ class OcrClient:
             latency_ms=outcome.latency_ms,
             call_id=outcome.call_id,
             raw=result.raw,
+            call_stats=call_stats,
         )
 
     async def check_health(self) -> dict[str, bool]:
@@ -242,11 +246,14 @@ class OcrClient:
         parent_call_id: str | None,
         tenant_id: str | None,
         meta: dict[str, Any],
-    ) -> _AttemptOutcome:
+    ) -> tuple[_AttemptOutcome, CallStats]:
         if not isinstance(image, bytes):
             raise TypeError("image 必须是 bytes(路径读取/批量拼帧留业务侧,D9)")
         if not image:
             raise ValueError("image 不能为空")
+        # M1 例外: `image` 校验在 `_call` 内而非公开方法,故上下文在该校验
+        # **通过之后**创建——这样设计 §3 的"校验在统计边界外"对 OCR 才成立
+        context = _CallContext(now=self._now)
         if not self._sources:
             raise AllSourcesExhausted(scope=self._scope, reason="no_sources", retry_after_s=0.0)
         fails = 0
@@ -260,10 +267,18 @@ class OcrClient:
                 continue
             async with clock.attempting():
                 outcome = await self._attempt(
-                    kind, image, *picked, reasons, session_id, parent_call_id, tenant_id, meta
+                    kind,
+                    image,
+                    *picked,
+                    reasons,
+                    session_id,
+                    parent_call_id,
+                    tenant_id,
+                    meta,
+                    context,
                 )
             if isinstance(outcome, _AttemptOutcome):
-                return outcome
+                return outcome, context.snapshot()
             fails += 1
             if fails >= self._retry.max_attempts:
                 raise AllSourcesExhausted(
@@ -287,11 +302,14 @@ class OcrClient:
         parent_call_id: str | None,
         tenant_id: str | None,
         meta: dict[str, Any],
+        context: _CallContext,
     ) -> _AttemptOutcome | _FailedAttempt:
         call_id = str(uuid.uuid4())
         started = self._now()
         # 四个 emit 分支(成功/终态拒绝/取消/可重试失败)都必须带调用方维度:
         # 失败行与取消行同样需要租户归属,漏掉任一分支就会写出无归属的行
+        # layout 的 POST + ZIP GET 在同一次 `_invoke` 内,故这里只登记 **1** 次
+        context.register_attempt()
         try:
             result = await self._invoke(kind, image, source, call_id)
             await self._record_quietly(self._breaker.record_success(entry))
diff --git a/src/polygateway/types.py b/src/polygateway/types.py
index 2857d8e..8800036 100644
--- a/src/polygateway/types.py
+++ b/src/polygateway/types.py
@@ -8,11 +8,12 @@ import dataclasses
 import json
 import math
 import re
-from collections.abc import Mapping
+import uuid
+from collections.abc import Callable, Mapping
 from dataclasses import dataclass, field
 from enum import StrEnum
 from types import MappingProxyType
-from typing import Any
+from typing import Any, Literal
 
 from loguru import logger
 
@@ -274,6 +275,91 @@ class ThinkingObservation(StrEnum):
     UNKNOWN = "unknown"
 
 
+CallOperation = Literal["chat", "embed", "recognize_text", "parse_layout"]
+"""遥测 `operation` 列的值域: **公开方法**四值,由调用点给定。
+
+与 `PolyGatewayError.operation`(HTTP 子操作,如 `download_result`)是**两个语义**,
+不做自动转换;链路上任何位置都不得读 `exc.operation` 来填本列(设计 §5 I1/I2)。"""
+
+CALL_OPERATIONS: tuple[CallOperation, ...] = ("chat", "embed", "recognize_text", "parse_layout")
+
+EventKind = Literal["attempt", "cache_hit", "terminal_failure"]
+"""一行遥测描述的事件形态;旧行 NULL,不回填。
+
+终态行与 attempt 行**不是重复事实**(前者描述逻辑终态,后者描述单次尝试),
+故禁止按 `error IS NOT NULL` 跨两类直接计失败调用次数(设计 §6/§8)。"""
+
+EVENT_KINDS: tuple[EventKind, ...] = ("attempt", "cache_hit", "terminal_failure")
+
+
+@dataclass(frozen=True)
+class CallStats:
+    """一次**公开调用**(而非单次尝试)的统计快照(设计 §3)。
+
+    四种响应各平铺三字段会立刻漂移,故收敛成单一对象并由包根导出。
+    第三方合成响应的 `None` 表示**未知**,不得伪造 0。
+    """
+
+    logical_call_id: str
+    """每次公开调用一个 UUID;重试、结构化重问、embedding 分批共享同一个。
+
+    不占用既有 `parent_call_id`(后者是调用方的业务关联,语义不变)。"""
+
+    attempts: int
+    """准入后实际调用 transport 端口的次数;含免预算 429 与端口本地拒绝。
+
+    **不是 HTTP 请求条数**: OCR layout 的 POST + ZIP GET 在同一次 transport
+    调用内,计 1 次。缓存命中与空输入是合法的零尝试。"""
+
+    total_latency_ms: int
+    """从输入校验通过到返回/异常传播前的单调时钟快照。
+
+    含缓存 IO、退避等待、准入等待、重问、分批与内联记账。
+    "总耗时减最后一次尝试耗时"**不等于**纯等待(含其他本地工作)。"""
+
+
+class _CallContext:
+    """私有可变逻辑调用上下文: 只持计数、单调时钟与终态去重位,不做 I/O。
+
+    **每调用一个实例**的单任务对象: chat 重试、结构化重问、embedding 分批
+    都在同一任务内串行推进,故计数无需锁。**严禁提升为 client 实例属性**
+    ——那会让同一 client 的并发调用互相串掉计数与逻辑 ID(库铁律"纯 asyncio 中立"、
+    VT `evolve_llm = llm` 教训的同一形态)。
+    """
+
+    __slots__ = ("_attempts", "_now", "_started", "_terminal_claimed", "logical_call_id")
+
+    def __init__(self, *, now: Callable[[], float]) -> None:
+        self.logical_call_id = str(uuid.uuid4())
+        self._now = now
+        self._started = now()
+        self._attempts = 0
+        self._terminal_claimed = False
+
+    def register_attempt(self) -> None:
+        """transport 调用**前**登记一次尝试(含免预算 429 与端口本地拒绝)。
+
+        登记点在调用前而非成功后: 否则失败与取消的尝试会从计数里消失,
+        而那正是诊断时最需要看见的那几次。
+        """
+        self._attempts += 1
+
+    def snapshot(self) -> CallStats:
+        """同步冻结当前快照;**绝不 await**,可多次调用。"""
+        return CallStats(
+            logical_call_id=self.logical_call_id,
+            attempts=self._attempts,
+            total_latency_ms=int((self._now() - self._started) * 1000),
+        )
+
+    def claim_terminal(self) -> bool:
+        """首次 `True`、其后 `False`: 保证每逻辑调用至多写一条终态行。"""
+        if self._terminal_claimed:
+            return False
+        self._terminal_claimed = True
+        return True
+
+
 @dataclass(frozen=True)
 class LLMResponse:
     """一次治理调用的统一响应(与三项目超集兼容,ARCH §5.1)。"""
@@ -336,6 +422,9 @@ class LLMResponse:
     `None` 不是"没推理": 库不表态时也不推定模型自己的默认档——"没看见"不许说成
     "发生了"(同 `thinking_observation` 的 `UNKNOWN` 一脉)。"""
 
+    call_stats: CallStats | None = None
+    """本次**逻辑调用**的统计快照(1.3.5);`None` = 未知,不得读成 0。"""
+
 
 @dataclass(frozen=True)
 class ChatRequest:
@@ -382,6 +471,16 @@ class ChatRequest:
     而档位要经能力表校验、要进缓存 key、要落遥测——混进直通层等于放弃这三样,
     正是 issue #20 里下游手写 `extra_body` 绕过全部治理的那条路。"""
 
+    # —— 库内部逻辑调用上下文(1.3.5;追加在末尾,不扰动既有字段的位置构造)——
+    call_context: _CallContext | None = field(default=None, compare=False, repr=False)
+    """库内部逻辑调用上下文;`None` = 库内现场构造的请求,遥测 `logical_call_id` 落 NULL。
+
+    `compare=False, repr=False` 不是洁癖: 进 `compare` 会让两个内容相同的请求因
+    "不是同一次调用"而不相等,进 `repr` 则把库内部件泄进调用方的日志。
+
+    洋葱各层经 `dataclasses.replace` 派生请求时保留**同一引用**(不是拷贝),
+    重试/重问/分批才能共享同一个逻辑 ID 与计数。"""
+
 
 @dataclass(frozen=True)
 class Usage:
@@ -722,6 +821,8 @@ class OcrTextResult:
     latency_ms: int
     call_id: str
     raw: dict[str, Any]
+    call_stats: CallStats | None = None
+    """本次逻辑调用的统计快照(1.3.5);`None` = 未知。"""
 
 
 @dataclass(frozen=True)
@@ -739,6 +840,8 @@ class OcrLayoutResult:
     latency_ms: int
     call_id: str
     raw: dict[str, Any]
+    call_stats: CallStats | None = None
+    """本次逻辑调用的统计快照(1.3.5);`None` = 未知。"""
 
 
 @dataclass(frozen=True)
@@ -783,3 +886,5 @@ class EmbeddingResponse:
     call_id: str
     source_name: str
     cost: float | None = None
+    call_stats: CallStats | None = None
+    """本次逻辑调用(含全部分批)的统计快照(1.3.5);`None` = 未知。"""
diff --git a/tests/unit/test_cache.py b/tests/unit/test_cache.py
index 3f5b4b6..6e86044 100644
--- a/tests/unit/test_cache.py
+++ b/tests/unit/test_cache.py
@@ -818,3 +818,75 @@ class TestExplicitCacheMigration:
         finally:
             for transport in transports:
                 await transport.aclose()
+
+
+class TestCallStatsNotPoisoned:
+    """缓存不得回放历史统计(1.3.5 设计 §3)。
+
+    统计描述**本次**调用;把上次那条存进去再放出来,等于对调用方谎称这次
+    重试了 N 次、耗了 M 毫秒。
+    """
+
+    async def test_serialized_payload_carries_no_call_stats_key(self):
+        from polygateway.types import CallStats
+
+        backend = InMemoryCache()
+        mw = _mw(backend)
+        stats = CallStats(logical_call_id="lc-1", attempts=3, total_latency_ms=900)
+        terminal = _Terminal(_resp(call_stats=stats))
+        await mw(ChatRequest(messages=_MSGS), terminal)
+        key = build_cache_key("m", _MSGS, "proj", None)
+        stored = json.loads(await backend.get(key))
+        assert "call_stats" not in stored  # asdict 会把它摊成 dict,必须显式剔除
+
+    async def test_historic_dict_never_impersonates_call_stats(self):
+        """旧条目里的 `call_stats` dict 会被 `_RESPONSE_FIELDS` 放行,必须显式覆盖。
+
+        不覆盖就会有一个 dict 冒充 `CallStats` 从公共 API 漏给调用方,
+        `resp.call_stats.attempts` 当场 `AttributeError`。
+        """
+        backend = InMemoryCache()
+        mw = _mw(backend)
+        key = build_cache_key("m", _MSGS, "proj", None)
+        poisoned = {
+            "content": "legacy",
+            "thinking": "",
+            "model": "m",
+            "provider": "p",
+            "prompt_tokens": 1,
+            "completion_tokens": 2,
+            "latency_ms": 30,
+            "ttft_ms": 5.0,
+            "max_inter_token_ms": 2.0,
+            "cache_hit": False,
+            "call_id": "orig",
+            "source_name": "s1",
+            "usage_source": "measured",
+            "call_stats": {
+                "logical_call_id": "stale-lc",
+                "attempts": 7,
+                "total_latency_ms": 9999,
+            },
+        }
+        await backend.set(key, json.dumps(poisoned), ttl_s=100)
+        terminal = _Terminal(_resp())
+        hit = await mw(ChatRequest(messages=_MSGS), terminal)
+        assert hit.content == "legacy" and terminal.calls == 0  # 真的走了缓存
+        assert hit.call_stats is None  # dict 不得冒充 CallStats
+
+    async def test_cache_key_is_unchanged_by_the_new_field(self):
+        """新增内部字段不得扰动 key 公式,否则存量缓存全量冷启动(黄金值)。"""
+        from polygateway.types import _CallContext
+
+        class _Clock:
+            def __call__(self):
+                return 1000.0
+
+        ctx = _CallContext(now=_Clock())
+        bare = build_cache_key("m", _MSGS, "proj", None)
+        assert bare == build_cache_key("m", _MSGS, "proj", None)
+        # 带上下文的请求与不带的请求必须落在同一个 key 上
+        with_ctx = ChatRequest(messages=_MSGS, call_context=ctx)
+        without = ChatRequest(messages=_MSGS)
+        assert with_ctx.cache_namespace == without.cache_namespace
+        assert digest_messages(with_ctx.messages) == digest_messages(without.messages)
diff --git a/tests/unit/test_client.py b/tests/unit/test_client.py
index 2f4d8e7..bc77f2c 100644
--- a/tests/unit/test_client.py
+++ b/tests/unit/test_client.py
@@ -1362,3 +1362,140 @@ async def test_synthetic_runtime_protocol_and_legacy_call_signatures():
         assert client._transport._clients == {}
     finally:
         await client.aclose()
+
+
+class _StatsClock:
+    """确定性单调钟;测试主动推进以断言"哪些区段计入了总耗时"。"""
+
+    def __init__(self, start=1000.0):
+        self.t = start
+
+    def __call__(self):
+        return self.t
+
+    def advance(self, seconds):
+        self.t += seconds
+
+
+class _TickingCache:
+    """假缓存后端: 每次 IO 推进注入钟。
+
+    不推进时钟的替身会让"缓存 IO 计入总耗时"的断言退化成恒等于 0 的空转绿
+    (计划 §T1 替身构造要求)。
+    """
+
+    def __init__(self, clock, tick=0.25):
+        self._clock = clock
+        self._tick = tick
+        self._data = {}
+
+    async def get(self, key):
+        self._clock.advance(self._tick)
+        return self._data.get(key)
+
+    async def set(self, key, value, ttl_s):
+        self._clock.advance(self._tick)
+        self._data[key] = value
+
+
+class _TickingRecorder:
+    """假 recorder: 写入时推进注入钟,用于断言内联遥测收尾计入总耗时。"""
+
+    def __init__(self, clock, tick=0.5):
+        self._clock = clock
+        self._tick = tick
+        self.rows = []
+
+    async def record_llm_call(self, **fields):
+        self._clock.advance(self._tick)
+        self.rows.append(fields)
+
+
+class TestLogicalCallStats:
+    """一次公开 chat 调用的统计(1.3.5 设计 §3)。"""
+
+    _MSG = [{"role": "user", "content": "hi"}]
+
+    async def test_success_reports_one_attempt(self):
+        async with _client() as client:
+            resp = await client.chat(self._MSG)
+        assert resp.call_stats is not None
+        assert resp.call_stats.attempts == 1
+        assert resp.call_stats.logical_call_id
+
+    async def test_concurrent_calls_do_not_share_counters_or_ids(self):
+        """同一 client 并发两路必须各自计数与各自 ID(库铁律「纯 asyncio 中立」)。
+
+        上下文若被提升成 client 实例属性,这条就会红——那正是 VT
+        `evolve_llm = llm` 教训的同一形态。
+        """
+        async with _client() as client:
+            a, b = await asyncio.gather(client.chat(self._MSG), client.chat(self._MSG))
+        assert a.call_stats.logical_call_id != b.call_stats.logical_call_id
+        assert a.call_stats.attempts == b.call_stats.attempts == 1
+
+    async def test_cache_hit_is_zero_attempts_with_a_fresh_logical_id(self):
+        """命中不产生网关调用 → 0 尝试;且是**新**逻辑调用,不回放历史统计。"""
+        clock = _StatsClock()
+        cache = _TickingCache(clock)
+        async with _client(
+            cache=cache, cache_namespace="proj", cache_ttl_s=600, now=clock
+        ) as client:
+            first = await client.chat(self._MSG)
+            second = await client.chat(self._MSG)
+        assert first.cache_hit is False and first.call_stats.attempts == 1
+        assert second.cache_hit is True
+        assert second.call_stats.attempts == 0
+        assert second.call_stats.logical_call_id != first.call_stats.logical_call_id
+
+    async def test_cache_io_counts_into_total_latency(self):
+        """缓存读写是本次调用真实花掉的时间,必须进总耗时(设计 §3)。"""
+        clock = _StatsClock()
+        cache = _TickingCache(clock, tick=0.25)
+        async with _client(
+            cache=cache, cache_namespace="proj", cache_ttl_s=600, now=clock
+        ) as client:
+            hit = (await client.chat(self._MSG), await client.chat(self._MSG))[1]
+        # 命中路径只有一次 get(0.25s),无网关调用
+        assert hit.call_stats.attempts == 0
+        assert hit.call_stats.total_latency_ms == 250
+
+    async def test_inline_telemetry_teardown_counts_into_total_latency(self):
+        """成功响应的快照含返回前已完成的内联遥测耗时(设计 §6)。"""
+        clock = _StatsClock()
+        recorder = _TickingRecorder(clock, tick=0.5)
+        async with _client(telemetry=recorder, now=clock) as client:
+            resp = await client.chat(self._MSG)
+        assert recorder.rows  # 确实写了行,否则本断言空转
+        assert resp.call_stats.total_latency_ms == 500
+
+    async def test_milliseconds_not_seconds(self):
+        """毫秒/秒不混用: 1.5s 必须是 1500 而不是 1 或 1.5。"""
+        clock = _StatsClock()
+        recorder = _TickingRecorder(clock, tick=1.5)
+        async with _client(telemetry=recorder, now=clock) as client:
+            resp = await client.chat(self._MSG)
+        assert resp.call_stats.total_latency_ms == 1500
+
+    async def test_stats_work_without_any_telemetry(self):
+        """统计生效与否**不由 telemetry 是否启用决定**(设计 §3.5)。"""
+        async with _client(telemetry=None) as client:
+            resp = await client.chat(self._MSG)
+        assert resp.call_stats is not None and resp.call_stats.attempts == 1
+
+    async def test_failure_exception_carries_no_stats_attribute(self):
+        """本版**不向异常对象附加统计**(设计 §3.1): 第三方可能复用同一异常实例。"""
+
+        def reject(request):
+            return httpx.Response(400, json={"error": {"message": "bad"}})
+
+        async with _client(handler=reject) as client:
+            with pytest.raises(RequestRejectedError) as exc:
+                await client.chat(self._MSG)
+        assert hasattr(exc.value, "call_stats") is False
+
+    async def test_input_validation_stays_outside_the_stats_boundary(self):
+        """校验异常保持原行为,发生在统计边界之外(设计 §3)。"""
+        async with _client() as client:
+            with pytest.raises(ValueError, match="meta"):
+                await client.chat(self._MSG, meta={"BAD-KEY": 1})
diff --git a/tests/unit/test_embedding.py b/tests/unit/test_embedding.py
index 94b2c6c..3384a75 100644
--- a/tests/unit/test_embedding.py
+++ b/tests/unit/test_embedding.py
@@ -600,3 +600,43 @@ class TestReasonlessTelemetryContract:
             assert seen == [{"model": "embed-1", "input": ["text"]}]
         finally:
             await transport.aclose()
+
+
+class TestEmbedLogicalCallStats:
+    """分批共享同一逻辑调用(1.3.5 设计 §3/§3.5)。"""
+
+    async def test_three_batches_count_three_attempts(self):
+        """分批是库的实现细节,但每批都真打了一次网关,故计 3 次尝试。"""
+        client, _ = _embed_client([_src()], ["ok", "ok", "ok"], batch_size=2)
+        resp = await client.embed(["a", "bb", "ccc", "dddd", "eeeee"])
+        assert resp.call_stats is not None
+        assert resp.call_stats.attempts == 3
+
+    async def test_separate_calls_get_distinct_logical_ids(self):
+        """一次公开调用一个 ID: 两次 `embed` 不得共用同一个。
+
+        共用就意味着上下文被提升成了 client 实例属性(库铁律禁止的形态)。
+        """
+        client, _ = _embed_client([_src()], ["ok"] * 5, batch_size=2)
+        first = await client.embed(["a", "bb", "ccc", "dddd", "eeeee"])  # 3 批
+        second = await client.embed(["x", "y"])  # 1 批
+        assert first.call_stats.attempts == 3 and second.call_stats.attempts == 1
+        assert first.call_stats.logical_call_id != second.call_stats.logical_call_id
+
+    async def test_retry_within_a_batch_is_counted(self):
+        client, _ = _embed_client([_src(), _src(name="e2")], [TransientError("t1"), "ok"])
+        resp = await client.embed(["a"])
+        assert resp.call_stats.attempts == 2
+
+    async def test_empty_input_is_zero_attempts_and_writes_no_telemetry_row(self):
+        """合法零尝试: 返回真实统计,且**不写任何遥测行**(设计 §3 M2)。
+
+        与 cache_hit 不同——不要按"遥测必录"推断空输入也有台账行。
+        """
+        rec = _MemoryRecorder()
+        client, _ = _embed_client([_src()], [], telemetry=rec)
+        resp = await client.embed([])
+        assert resp.call_stats is not None
+        assert resp.call_stats.attempts == 0
+        assert resp.call_stats.logical_call_id  # 真实 ID,不是空串
+        assert rec.rows == []  # 零遥测行
diff --git a/tests/unit/test_ocr_client.py b/tests/unit/test_ocr_client.py
index 00e435b..acac9c2 100644
--- a/tests/unit/test_ocr_client.py
+++ b/tests/unit/test_ocr_client.py
@@ -615,3 +615,34 @@ class TestReasonlessTelemetryContract:
             await getattr(client, method)(b"image")
         assert len(recorder.rows) == len(script)
         assert all(r["error"] and r["reasoning_effort"] is None for r in recorder.rows)
+
+
+class TestOcrLogicalCallStats:
+    """OCR 两个公开方法各自拥有一次逻辑调用(1.3.5 设计 §3/§3.5)。"""
+
+    async def test_text_success_counts_one_attempt(self):
+        client, _, _ = _client([_src()], ["text"])
+        r = await client.recognize_text(b"jpg")
+        assert r.call_stats is not None and r.call_stats.attempts == 1
+
+    async def test_layout_two_http_calls_count_as_one_attempt(self):
+        """POST + ZIP GET 在同一次 transport 调用内,计 **1** 次尝试而非 2。
+
+        `attempts` 的语义是"调用 transport 端口的次数",不是 HTTP 请求条数。
+        """
+        client, _, _ = _client([_src()], ["layout"])
+        r = await client.parse_layout(b"jpg")
+        assert r.call_stats is not None and r.call_stats.attempts == 1
+
+    async def test_retry_counts_every_attempt(self):
+        client, _, _ = _client([_src(), _src(name="m2")], [TransientError("t1"), "text"])
+        r = await client.recognize_text(b"jpg")
+        assert r.call_stats.attempts == 2
+
+    async def test_input_validation_stays_outside_the_stats_boundary(self):
+        """`image` 类型/空校验先于上下文创建(M1 例外),保持原异常行为。"""
+        client, _, _ = _client([_src()], [])
+        with pytest.raises(TypeError):
+            await client.recognize_text("not-bytes")
+        with pytest.raises(ValueError):
+            await client.recognize_text(b"")
diff --git a/tests/unit/test_retry.py b/tests/unit/test_retry.py
index 96d566d..5a91b7a 100644
--- a/tests/unit/test_retry.py
+++ b/tests/unit/test_retry.py
@@ -5,6 +5,7 @@
 """
 
 import asyncio
+import dataclasses
 
 import pytest
 
@@ -776,3 +777,60 @@ class TestRateLimitPushback:
             await mw(_REQ)
         assert ei.value.reason == "retry_exhausted"
         assert len(transport.calls) == 3
+
+
+class TestLogicalAttemptCounting:
+    """尝试登记在 transport 调用**之前**(1.3.5 设计 §4)。
+
+    登记点若挪到成功之后,失败与取消的尝试就会从计数里消失——而那正是
+    诊断时最需要看见的几次。
+    """
+
+    def _ctx(self, clock):
+        from polygateway.types import _CallContext
+
+        return _CallContext(now=clock)
+
+    async def test_single_success_counts_one(self):
+        mw, _, _, _, _, clock = _harness([_src("a")], [_ok()])
+        ctx = self._ctx(clock)
+        await mw(dataclasses.replace(_REQ, call_context=ctx))
+        assert ctx.snapshot().attempts == 1
+
+    async def test_failed_retries_are_counted(self):
+        """两次可重试失败 + 一次成功 = 3 次尝试,不是 1 次。"""
+        mw, _, _, transport, _, clock = _harness(
+            [_src("a")], [TransientError("t1"), TransientError("t2"), _ok()]
+        )
+        ctx = self._ctx(clock)
+        await mw(dataclasses.replace(_REQ, call_context=ctx))
+        assert ctx.snapshot().attempts == 3 == len(transport.calls)
+
+    async def test_budget_free_429_still_counts_as_an_attempt(self):
+        """429 免的是重试预算,不是"没发生过"——它确实打到了网关。"""
+        mw, _, _, transport, _, clock = _harness(
+            [_src("a")],
+            [
+                TransientError("t1", status_code=429, retry_after_s=1.0),
+                TransientError("t2", status_code=429, retry_after_s=1.0),
+                _ok(),
+            ],
+        )
+        ctx = self._ctx(clock)
+        await mw(dataclasses.replace(_REQ, call_context=ctx))
+        assert ctx.snapshot().attempts == 3 == len(transport.calls)
+
+    async def test_retry_exhausted_counts_every_attempt(self):
+        mw, _, _, transport, _, clock = _harness(
+            [_src("a")], [TransientError(str(i)) for i in range(5)], max_attempts=3
+        )
+        ctx = self._ctx(clock)
+        with pytest.raises(AllSourcesExhausted):
+            await mw(dataclasses.replace(_REQ, call_context=ctx))
+        assert ctx.snapshot().attempts == 3 == len(transport.calls)
+
+    async def test_absent_context_does_not_break_the_call(self):
+        """库内现场构造的 `ChatRequest` 没有上下文,不得因此报错(设计 §3.5)。"""
+        mw, _, _, _, _, _ = _harness([_src("a")], [_ok()])
+        resp = await mw(_REQ)
+        assert resp.content == "ok" and _REQ.call_context is None
diff --git a/tests/unit/test_types.py b/tests/unit/test_types.py
index 6efafe8..4459383 100644
--- a/tests/unit/test_types.py
+++ b/tests/unit/test_types.py
@@ -610,3 +610,144 @@ class TestSourceConfigEffortNormalization:
         """非字符串同样只能是 `ValueError`: 公共入口不许把类型错误漏成 `AttributeError`。"""
         with pytest.raises(ValueError, match="推理档位"):
             _make_source(reasoning_effort=3)
+
+
+class TestCallStatsAndContext:
+    """逻辑调用统计内核(1.3.5 设计 §3/§4)。"""
+
+    def test_call_stats_is_frozen_snapshot(self):
+        from polygateway.types import CallStats
+
+        stats = CallStats(logical_call_id="lc-1", attempts=2, total_latency_ms=15)
+        with pytest.raises(dataclasses.FrozenInstanceError):
+            stats.attempts = 3
+
+    def test_context_counts_attempts_and_freezes_elapsed(self):
+        """快照是同步冻结的时间切片: 登记两次尝试后耗时按注入钟折算成毫秒。"""
+        from polygateway.types import _CallContext
+
+        clock = _FakeMonotonic()
+        ctx = _CallContext(now=clock)
+        clock.advance(1.5)
+        ctx.register_attempt()
+        ctx.register_attempt()
+        stats = ctx.snapshot()
+        assert stats.attempts == 2
+        assert stats.total_latency_ms == 1500  # 秒→毫秒,不混用单位
+
+    def test_snapshot_is_repeatable_and_tracks_later_time(self):
+        from polygateway.types import _CallContext
+
+        clock = _FakeMonotonic()
+        ctx = _CallContext(now=clock)
+        first = ctx.snapshot()
+        clock.advance(2.0)
+        second = ctx.snapshot()
+        assert first.total_latency_ms == 0 and second.total_latency_ms == 2000
+        assert first.logical_call_id == second.logical_call_id
+
+    def test_each_context_gets_its_own_logical_id(self):
+        from polygateway.types import _CallContext
+
+        clock = _FakeMonotonic()
+        assert _CallContext(now=clock).logical_call_id != _CallContext(now=clock).logical_call_id
+
+    def test_claim_terminal_is_true_once(self):
+        """终态去重位: 保证每逻辑调用至多写一条终态行(设计 §6 不变量 I3)。"""
+        from polygateway.types import _CallContext
+
+        ctx = _CallContext(now=_FakeMonotonic())
+        assert ctx.claim_terminal() is True
+        assert ctx.claim_terminal() is False
+
+    def test_chat_request_context_does_not_affect_equality_or_repr(self):
+        """上下文是库内部件: 进 `compare`/`repr` 会污染既有请求语义与日志。"""
+        from polygateway.types import _CallContext
+
+        ctx = _CallContext(now=_FakeMonotonic())
+        bare = ChatRequest(messages=[{"role": "user", "content": "hi"}])
+        with_ctx = dataclasses.replace(bare, call_context=ctx)
+        assert with_ctx.call_context is ctx
+        assert with_ctx == bare
+        assert "call_context" not in repr(with_ctx)
+
+    def test_replace_preserves_the_same_context_reference(self):
+        """洋葱各层经 `replace` 派生请求,上下文必须是同一实例而非拷贝。"""
+        from polygateway.types import _CallContext
+
+        ctx = _CallContext(now=_FakeMonotonic())
+        req = ChatRequest(messages=[{"role": "user", "content": "hi"}], call_context=ctx)
+        derived = dataclasses.replace(req, stream=False)
+        assert derived.call_context is ctx
+
+    def test_four_responses_default_call_stats_to_none(self):
+        """第三方合成响应的 `None` 表示未知,不得伪造 0(设计 §3)。"""
+        from polygateway.types import (
+            EmbeddingResponse,
+            OcrLayoutResult,
+            OcrTextResult,
+        )
+
+        llm = LLMResponse(
+            content="c",
+            thinking="",
+            model="m",
+            provider="p",
+            prompt_tokens=1,
+            completion_tokens=1,
+            latency_ms=1,
+            ttft_ms=None,
+            max_inter_token_ms=None,
+            cache_hit=False,
+            call_id="c1",
+        )
+        emb = EmbeddingResponse(
+            vectors=[],
+            dim=0,
+            model="m",
+            provider="p",
+            prompt_tokens=0,
+            usage_source="measured",
+            latency_ms=0,
+            call_id="c2",
+            source_name="s",
+        )
+        text = OcrTextResult(
+            text="", source_name="s", usage=Usage(0, 0), latency_ms=0, call_id="c3", raw={}
+        )
+        layout = OcrLayoutResult(
+            elements=[],
+            page_sizes=[],
+            source_name="s",
+            usage=Usage(0, 0),
+            latency_ms=0,
+            call_id="c4",
+            raw={},
+        )
+        assert (llm.call_stats, emb.call_stats, text.call_stats, layout.call_stats) == (
+            None,
+            None,
+            None,
+            None,
+        )
+
+    def test_call_stats_is_exported_from_package_root(self):
+        """四份平铺字段会漂移,故统计以单一对象出现在公共 API(设计 §3)。"""
+        import polygateway
+        from polygateway.types import CallStats
+
+        assert polygateway.CallStats is CallStats
+        assert "CallStats" in polygateway.__all__
+
+
+class _FakeMonotonic:
+    """确定性单调钟;不复用 contracts 的 FakeClock 以免 unit 反向依赖契约包。"""
+
+    def __init__(self, start: float = 1000.0) -> None:
+        self.t = start
+
+    def __call__(self) -> float:
+        return self.t
+
+    def advance(self, seconds: float) -> None:
+        self.t += seconds