diff --git a/src/polygateway/middleware/telemetry.py b/src/polygateway/middleware/telemetry.py
index 058637f..04251de 100644
--- a/src/polygateway/middleware/telemetry.py
+++ b/src/polygateway/middleware/telemetry.py
@@ -23,7 +23,7 @@ from polygateway.errors import (
     SourceNotConfiguredError,
 )
 from polygateway.middleware.cache import digest_messages
-from polygateway.types import canonical_sampling_json, merge_sampling
+from polygateway.types import ThinkingObservation, canonical_sampling_json, merge_sampling
 
 if TYPE_CHECKING:
     from collections.abc import Callable, Mapping
@@ -111,6 +111,9 @@ class _AttemptUsage:
     cached_prompt_tokens: int | None = None
     model_reported: str | None = None
     reasoning_tokens: int | None = None
+    # 内部字段用枚举类型;裸 str 归一化只发生在 `_record` 下沉 recorder 那一步。
+    # 失败尝试无响应可言,默认 UNKNOWN 本身就是事实("观测不到"),不撒谎
+    thinking_observation: ThinkingObservation = ThinkingObservation.UNKNOWN
 
     @classmethod
     def of(cls, response: LLMResponse | None) -> _AttemptUsage:
@@ -128,11 +131,12 @@ class _AttemptUsage:
             cached_prompt_tokens=response.cached_prompt_tokens,
             model_reported=response.model_reported,
             reasoning_tokens=response.reasoning_tokens,
+            thinking_observation=response.thinking_observation,
         )
 
 
 class TelemetryEmitter:
-    """从请求与结果组装 24 字段并写入 recorder;一切写失败降级 warning。"""
+    """从请求与结果组装 25 字段并写入 recorder;一切写失败降级 warning。"""
 
     def __init__(
         self,
@@ -185,6 +189,7 @@ class TelemetryEmitter:
             cached_prompt_tokens=usage.cached_prompt_tokens,
             model_reported=usage.model_reported,
             reasoning_tokens=usage.reasoning_tokens,
+            thinking_observation=usage.thinking_observation,
             # 唯一有"生效源"的入口,故是唯一能并上 extra_body 的(设计决策 D)
             sampling=canonical_sampling_json(merge_sampling(source.extra_body, request.sampling)),
             tenant_id=request.tenant_id,
@@ -214,6 +219,8 @@ class TelemetryEmitter:
             cached_prompt_tokens=response.cached_prompt_tokens,
             model_reported=response.model_reported,
             reasoning_tokens=response.reasoning_tokens,
+            # 与 model/prompt_tokens 同一口径: 原样回放历史那次的裁定结果
+            thinking_observation=response.thinking_observation,
             # 由最外层 TelemetryMW 调用,手上没有 source。缓存命中行无损:
             # sampling 已进缓存 key,能命中即意味调用级参数与历史那次逐字相同
             sampling=canonical_sampling_json(request.sampling),
@@ -247,6 +254,8 @@ class TelemetryEmitter:
             cached_prompt_tokens=None,
             model_reported=None,
             reasoning_tokens=None,
+            # 无响应可言,故裁不出结果;UNKNOWN 正是"观测不到"本身,不是伪装的"没推理"
+            thinking_observation=ThinkingObservation.UNKNOWN,
             # 无具体源,与 model/provider/source_name 置空同一先例(设计决策 D)
             sampling=canonical_sampling_json(request.sampling),
             # 源不可知,但租户归属是已知的——终态失败行恰是审计最需要的
@@ -276,6 +285,8 @@ class TelemetryEmitter:
         model_reported: str | None,
         sampling: str | None,
         reasoning_tokens: int | None,
+        # issue #16: 枚举形态进来,取 `.value` 后才下沉(归一化同样在本方法内收口)
+        thinking_observation: ThinkingObservation,
         # issue #11: 未归一化的调用方维度,归一化在本方法内收口(recorder 只落库)
         tenant_id: str | None,
         meta: Mapping[str, Any],
@@ -328,6 +339,10 @@ class TelemetryEmitter:
                 # 对所有人永久不可见,空串则可用一条 SQL 审计出未归属的行
                 tenant_id=tenant_id or "",
                 meta=_canonical_meta_json(meta),
+                # 取 `.value` 落裸 str: `StrEnum` 虽是 `str` 子类,asyncpg 的参数
+                # 编码对子类不保证接受,而遥测写失败只降级成一条 warning——不会当场
+                # 炸,只会让 Postgres 那一路悄悄少一列数据
+                thinking_observation=thinking_observation.value,
             )
         except asyncio.CancelledError:
             raise
diff --git a/src/polygateway/ports.py b/src/polygateway/ports.py
index 9429e4c..09a6c3f 100644
--- a/src/polygateway/ports.py
+++ b/src/polygateway/ports.py
@@ -260,13 +260,16 @@ class TelemetryStatusProvider(Protocol):
 
 @runtime_checkable
 class TelemetryRecorder(Protocol):
-    """遥测后端;24 字段冻结(M1 设计 §4.4 + issue #3/#4/#11),唯一调用点是 TelemetryEmitter。
+    """遥测后端;25 字段冻结(M1 设计 §4.4 + issue #3/#4/#11/#16),唯一调用点是 TelemetryEmitter。
 
     新增参数不设默认值: 库外无第三方实现者(三项目迁移时删除了各自的同名
     Protocol),完整签名的成本为零,而少写一列会被 emitter 的降级吞成 warning。
 
     `tenant_id` 与 `meta` 到达 recorder 时**已由 emitter 归一化**——`tenant_id`
     的 `None` 已转空串,`meta` 已序列化为 JSON 字符串(空 dict 为 `'{}'`)。
+    `thinking_observation` 同理: emitter 已把 `ThinkingObservation` 取成 `.value`
+    的裸 `str`(`StrEnum` 是 `str` 子类,而 asyncpg 的参数编码对子类不保证接受,
+    遥测写失败又只降级成 warning——PG 那一路会静默少一列数据)。
     recorder 只负责落库,不做任何语义判断,与 `sampling` 列由
     `canonical_sampling_json()` 在 emitter 侧定型是同一先例。
     """
@@ -298,4 +301,5 @@ class TelemetryRecorder(Protocol):
         reasoning_tokens: int | None,
         tenant_id: str,
         meta: str,
+        thinking_observation: str,
     ) -> None: ...
diff --git a/src/polygateway/telemetry/schema.py b/src/polygateway/telemetry/schema.py
index dd8f34e..93e3363 100644
--- a/src/polygateway/telemetry/schema.py
+++ b/src/polygateway/telemetry/schema.py
@@ -5,8 +5,8 @@
 多处各存一份必然漂移,而漂移的表现是"下游照打印的 SQL 建完表,库仍报缺列"。
 
 **`COLUMNS` 是 INSERT 字段序,不是物理列序**: 数据库自填的 `created_at` 不在其中(它带
-`DEFAULT now()` / `datetime('now')`,库从不显式写它)。物理表列 = 24 个 INSERT 字段 +
-`created_at` = 25;列数断言一律按物理列数写,两套口径混用是最易错处。
+`DEFAULT now()` / `datetime('now')`,库从不显式写它)。物理表列 = 25 个 INSERT 字段 +
+`created_at` = 26;列数断言一律按物理列数写,两套口径混用是最易错处。
 
 本模块只依赖标准库: `telemetry/` 与 `backends/`、`transports/`、`structured/` 同层且
 互不依赖(import-linter 契约执法)。
@@ -50,7 +50,8 @@ CREATE TABLE IF NOT EXISTS llm_calls (
     sampling             TEXT,
     reasoning_tokens     INTEGER,
     tenant_id            TEXT    NOT NULL DEFAULT '',
-    meta                 TEXT    NOT NULL DEFAULT '{}'
+    meta                 TEXT    NOT NULL DEFAULT '{}',
+    thinking_observation TEXT
 );
 """
 
@@ -80,7 +81,8 @@ CREATE TABLE IF NOT EXISTS llm_calls (
     sampling             TEXT,
     reasoning_tokens     INTEGER,
     tenant_id            TEXT NOT NULL DEFAULT '',
-    meta                 JSONB NOT NULL DEFAULT '{}'::jsonb
+    meta                 JSONB NOT NULL DEFAULT '{}'::jsonb,
+    thinking_observation TEXT
 );
 """
 
@@ -95,6 +97,9 @@ SQLITE_BACKFILL = (
     # ("Cannot add a NOT NULL column with default value NULL"),补列全盘失败。
     ("tenant_id", "TEXT NOT NULL DEFAULT ''"),
     ("meta", "TEXT NOT NULL DEFAULT '{}'"),
+    # 可空: 补列之前的行没有裁定结果,NULL 如实表达"这行根本没记过这件事",
+    # 与哨兵串 'unknown'(库确实裁过但判不出来)是两回事,不得混同
+    ("thinking_observation", "TEXT"),
 )
 
 # PG 补列的列定义。语句由此派生成两份文本(见下),使"库内执行的那份"与"打印给
@@ -107,6 +112,8 @@ _PG_BACKFILL_DECLS = (
     # 两个默认值都是非易失常量,PG 11+ 只改 catalog 不重写全表,故大表补列亦是秒级
     ("tenant_id", "TEXT NOT NULL DEFAULT ''"),
     ("meta", "JSONB NOT NULL DEFAULT '{}'::jsonb"),
+    # 可空,理由同 SQLITE_BACKFILL 同名项
+    ("thinking_observation", "TEXT"),
 )
 
 # 新列排在 created_at 之后: 与旧表 ALTER 追加的位置一致(见 SQLITE_BACKFILL 同款注释)。
@@ -143,6 +150,7 @@ COLUMNS = (
     "reasoning_tokens",
     "tenant_id",
     "meta",
+    "thinking_observation",
 )
 
 _COLUMN_SET = frozenset(COLUMNS)
diff --git a/src/polygateway/telemetry/sqlite.py b/src/polygateway/telemetry/sqlite.py
index e737dda..cc46f2c 100644
--- a/src/polygateway/telemetry/sqlite.py
+++ b/src/polygateway/telemetry/sqlite.py
@@ -143,7 +143,7 @@ class SQLiteRecorder:
                     logger.warning("SQLite 遥测补列失败(写入将逐行降级): {}", exc)
 
     async def record_llm_call(self, **fields: object) -> None:
-        """写一行遥测;字段集合即 24 字段冻结签名(ports.TelemetryRecorder)。
+        """写一行遥测;字段集合即 25 字段冻结签名(ports.TelemetryRecorder)。
 
         取值按 `self._columns`(manual 档可能已被裁剪),与 `self._insert` 的
         占位符同序——两者必须一起改,分开改就是把值写进错位的列。
diff --git a/tests/integration/test_postgres_telemetry.py b/tests/integration/test_postgres_telemetry.py
index b645e51..3875404 100644
--- a/tests/integration/test_postgres_telemetry.py
+++ b/tests/integration/test_postgres_telemetry.py
@@ -52,6 +52,7 @@ _EXPECTED_COLUMNS = [
     "reasoning_tokens",
     "tenant_id",
     "meta",
+    "thinking_observation",
 ]
 
 # run 级前缀: 同库并存的其他运行(迁移批跑/另一开发机)互不可见
@@ -125,6 +126,8 @@ async def _record_minimal(
         # 到达 recorder 时已由 emitter 归一化: None → '',空 dict → '{}'
         "tenant_id": "",
         "meta": "{}",
+        # 同样已由 emitter 归一化: 枚举取 .value 后才下沉,recorder 只见裸 str
+        "thinking_observation": "unknown",
     }
     fields.update(overrides)
     await recorder.record_llm_call(**fields)
@@ -602,10 +605,12 @@ _PRE_TENANT_INSERT = (
 )
 
 
-# `_PRE_TENANT_DDL` 的物理列(23 个): 由 `_EXPECTED_COLUMNS` 去掉 issue #11 的两个新维度
+# `_PRE_TENANT_DDL` 的物理列(23 个): 由 `_EXPECTED_COLUMNS` 去掉此后新增的三列
 # 派生而非另抄一份——两份常量必然漂移,而漂移的表现是"manual 档没补列"这条断言假绿。
-# 去掉后的顺序与 DDL 逐字一致(tenant_id/meta 在 DDL 里本就排在末尾)。
-_PRE_TENANT_COLUMNS = [c for c in _EXPECTED_COLUMNS if c not in ("tenant_id", "meta")]
+# 去掉后的顺序与 DDL 逐字一致(这三列在 DDL 里本就排在末尾)。
+_PRE_TENANT_COLUMNS = [
+    c for c in _EXPECTED_COLUMNS if c not in ("tenant_id", "meta", "thinking_observation")
+]
 
 # 回读要逐列比对的字段: 物理列去掉库从不显式写的 created_at,恰好 22 个
 _PRE_TENANT_WRITTEN_COLUMNS = [c for c in _PRE_TENANT_COLUMNS if c != "created_at"]
@@ -761,7 +766,7 @@ class TestCallerDimensionsAcceptance:
                 "WHERE table_schema = $1 AND table_name = 'llm_calls' ORDER BY ordinal_position",
                 schema,
             )
-            # 22 → 24 个 recorder 字段(加 created_at 共 25 个物理列),且新列追加在末尾
+            # 22 → 25 个 recorder 字段(加 created_at 共 26 个物理列),且新列追加在末尾
             assert [r["column_name"] for r in cols] == _EXPECTED_COLUMNS
             rows = await _fetch(
                 schema_dsn,
@@ -1060,7 +1065,7 @@ class TestPublishedSchemaScript:
 
         await _execute_script(fresh_dsn, script)
         actual = [r["column_name"] for r in await _fetch(fresh_dsn, _PHYSICAL_COLUMNS_SQL, schema)]
-        # 物理列 = 24 个 INSERT 字段 + 库从不显式写的 created_at;对着库常量比,不另抄一份
+        # 物理列 = 25 个 INSERT 字段 + 库从不显式写的 created_at;对着库常量比,不另抄一份
         assert set(actual) == set(COLUMNS) | {"created_at"}
         # 列序也不许漂: 新列必须排在 created_at 之后,否则新建库与 ALTER 升级的列序分叉
         assert actual == _EXPECTED_COLUMNS
diff --git a/tests/unit/test_ports.py b/tests/unit/test_ports.py
index abcdd75..89aea7c 100644
--- a/tests/unit/test_ports.py
+++ b/tests/unit/test_ports.py
@@ -244,7 +244,7 @@ class TestTelemetryRecorderSignature:
         params = inspect.signature(TelemetryRecorder.record_llm_call).parameters
         assert {"tenant_id", "meta"} <= set(params)
 
-    @pytest.mark.parametrize("name", ["tenant_id", "meta"])
+    @pytest.mark.parametrize("name", ["tenant_id", "meta", "thinking_observation"])
     def test_caller_dimensions_have_no_default(self, name):
         import inspect
 
diff --git a/tests/unit/test_telemetry.py b/tests/unit/test_telemetry.py
index 408f7ab..e2294e0 100644
--- a/tests/unit/test_telemetry.py
+++ b/tests/unit/test_telemetry.py
@@ -30,6 +30,7 @@ from polygateway.types import (
     OcrTextTransportResult,
     RetryPolicy,
     SourceConfig,
+    ThinkingObservation,
 )
 
 _REQ = ChatRequest(messages=[{"role": "user", "content": "hi"}], session_id="sess-1")
@@ -60,6 +61,7 @@ _EXPECTED_COLUMNS = [
     "reasoning_tokens",
     "tenant_id",
     "meta",
+    "thinking_observation",
 ]
 
 
@@ -127,6 +129,8 @@ async def _record_minimal(recorder, call_id="c1", **overrides):
         # 到达 recorder 时已由 emitter 归一化: None → '',空 dict → '{}'
         "tenant_id": "",
         "meta": "{}",
+        # 同样已由 emitter 归一化: 枚举取 .value 后才下沉,recorder 只见裸 str
+        "thinking_observation": "unknown",
     }
     fields.update(overrides)
     await recorder.record_llm_call(**fields)
@@ -172,16 +176,16 @@ _FROZEN_SQLITE_INSERT = (
     "INSERT OR IGNORE INTO llm_calls (call_id, parent_call_id, session_id, model, provider, "
     "source_name, messages, response, thinking, prompt_tokens, completion_tokens, usage_source, "
     "latency_ms, ttft_ms, max_inter_token_ms, cache_hit, error, cost, cached_prompt_tokens, "
-    "model_reported, sampling, reasoning_tokens, tenant_id, meta) "
-    "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"
+    "model_reported, sampling, reasoning_tokens, tenant_id, meta, thinking_observation) "
+    "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"
 )
 _FROZEN_PG_INSERT = (
     "INSERT INTO llm_calls (call_id, parent_call_id, session_id, model, provider, source_name, "
     "messages, response, thinking, prompt_tokens, completion_tokens, usage_source, latency_ms, "
     "ttft_ms, max_inter_token_ms, cache_hit, error, cost, cached_prompt_tokens, model_reported, "
-    "sampling, reasoning_tokens, tenant_id, meta) "
+    "sampling, reasoning_tokens, tenant_id, meta, thinking_observation) "
     "VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, "
-    "$19, $20, $21, $22, $23, $24) "
+    "$19, $20, $21, $22, $23, $24, $25) "
     # 无冲突目标(issue #13 Task 2): 带 `(call_id)` 的版本在按 created_at 分区、
     # 主键为 (call_id, created_at) 的表上匹配不到约束,PG 直接拒收整条写入
     "ON CONFLICT DO NOTHING"
@@ -207,7 +211,7 @@ class TestSchemaModule:
 
         # COLUMNS 是 INSERT 字段序,不含数据库自填的 created_at
         assert list(COLUMNS) == [c for c in _EXPECTED_COLUMNS if c != "created_at"]
-        assert len(COLUMNS) == 24
+        assert len(COLUMNS) == 25
         # 两端 DDL 的列出现顺序 == 物理列序(created_at 在第 19 位)
         for ddl in (SQLITE_DDL, PG_DDL):
             assert _first_occurrence_order(ddl, _EXPECTED_COLUMNS) == _EXPECTED_COLUMNS
@@ -221,8 +225,8 @@ class TestSchemaModule:
             "ALTER TABLE llm_calls ADD COLUMN cached_prompt_tokens INTEGER",
         )
         assert PG_BACKFILL[-1] == (
-            "meta",
-            "ALTER TABLE llm_calls ADD COLUMN meta JSONB NOT NULL DEFAULT '{}'::jsonb",
+            "thinking_observation",
+            "ALTER TABLE llm_calls ADD COLUMN thinking_observation TEXT",
         )
         assert all("IF NOT EXISTS" not in stmt for _, stmt in PG_BACKFILL)
 
@@ -270,7 +274,7 @@ class TestSchemaModule:
         pg = telemetry_schema_sql("postgres")
         lite = telemetry_schema_sql("sqlite")
         for script in (pg, lite):
-            # 24 个 INSERT 字段 + created_at 全在,且首次出现顺序与建表 DDL 一致
+            # 25 个 INSERT 字段 + created_at 全在,且首次出现顺序与建表 DDL 一致
             assert _first_occurrence_order(script, _EXPECTED_COLUMNS) == _EXPECTED_COLUMNS
             assert "CREATE TABLE IF NOT EXISTS llm_calls" in script
         # 人执行的那份必须幂等: PG 用 ADD COLUMN IF NOT EXISTS(与库内那份有意不同)
@@ -304,11 +308,11 @@ class TestBackendColumnParity:
         assert sqlite.COLUMNS is COLUMNS
         assert postgres.COLUMNS is COLUMNS
 
-    def test_caller_dimensions_are_appended_last(self):
+    def test_new_columns_are_appended_last(self):
         """新列只能追加在末尾: 旧表经 ALTER 补列必落末尾,插在中间会让两条路径分叉。"""
         from polygateway.telemetry.schema import COLUMNS
 
-        assert COLUMNS[-2:] == ("tenant_id", "meta")
+        assert COLUMNS[-3:] == ("tenant_id", "meta", "thinking_observation")
 
 
 class TestSQLiteRecorder:
@@ -378,6 +382,28 @@ class TestSQLiteRecorder:
         assert rows["r-zero"] == 0  # 上报了且确实没推理
         assert rows["r-none"] is None  # 本次调用未上报
 
+    async def test_thinking_observation_column_round_trips(self, tmp_path):
+        """issue #16: 三态裁定结果落库,事后才能按"这次到底推没推理"分组统计。
+
+        断言的是裸字符串 `"observed"` 而非枚举: 归一化在 emitter 侧完成
+        (`_record` 取 `.value`),recorder 拿到的必须已经是 `str`——`StrEnum`
+        虽是 `str` 子类,asyncpg 的参数编码对子类不保证接受,而遥测写失败只
+        降级成一条 warning,PG 那一路会悄无声息地少一列数据。
+        """
+        recorder = SQLiteRecorder(tmp_path / "t.db", auto_migrate=True)
+        await _record_minimal(recorder, call_id="t-obs", thinking_observation="observed")
+        await _record_minimal(recorder, call_id="t-absent", thinking_observation="absent")
+        await _record_minimal(recorder, call_id="t-unknown")
+        recorder.close()
+        rows = dict(
+            sqlite3.connect(tmp_path / "t.db")
+            .execute("SELECT call_id, thinking_observation FROM llm_calls")
+            .fetchall()
+        )
+        assert rows["t-obs"] == "observed"
+        assert rows["t-absent"] == "absent"  # 观测到"确实没推理",与"看不出来"不是一回事
+        assert rows["t-unknown"] == "unknown"
+
     async def test_sampling_column_round_trips(self, tmp_path):
         """issue #4: 采样参数落库,否则事后无法证明某批数据跑在什么温度下。"""
         recorder = SQLiteRecorder(tmp_path / "t.db", auto_migrate=True)
@@ -541,7 +567,7 @@ class TestSQLiteCallerDimensionsAcceptance:
 
         conn = sqlite3.connect(db)
         cols = [r[1] for r in conn.execute("PRAGMA table_info(llm_calls)")]
-        assert cols == _EXPECTED_COLUMNS  # 22 → 24 个 recorder 字段(+ created_at 共 25 物理列)
+        assert cols == _EXPECTED_COLUMNS  # 22 → 25 个 recorder 字段(+ created_at 共 26 物理列)
         rows = dict(conn.execute("SELECT call_id, tenant_id FROM llm_calls").fetchall())
         assert rows["new-row"] == "tenant-a"
         assert rows["old-row"] == ""  # 不是 None: NULL 会被 RLS 静默吞掉
@@ -583,7 +609,7 @@ class TestSQLiteCallerDimensionsAcceptance:
 
         stale = sqlite3.connect(db)
         assert [r[1] for r in stale.execute("PRAGMA table_info(llm_calls)")] == (
-            _EXPECTED_COLUMNS[:-2]
+            _EXPECTED_COLUMNS[:-3]
         )  # 补列确实没成功,用例不是在只读库上空转
 
 
@@ -591,7 +617,7 @@ class TestSQLiteSchemaMode:
     """issue #13: `auto_migrate` 两档——auto 保持自动补列,manual 只裁剪写入不发 DDL。
 
     列数断言一律按**物理列数**写: 旧表 22 个 INSERT 字段 + `created_at` = 23,
-    补齐后 24 + `created_at` = 25。混用 INSERT 字段数与物理列数是本处最易错的地方。
+    补齐后 25 + `created_at` = 26。混用 INSERT 字段数与物理列数是本处最易错的地方。
     """
 
     def _physical_columns(self, db: Path) -> list[str]:
@@ -630,7 +656,7 @@ class TestSQLiteSchemaMode:
         assert "ALTER TABLE" in message  # 给出可直接执行的补列 SQL
 
     async def test_auto_mode_still_upgrades_the_legacy_table(self, tmp_path):
-        """auto + 同款旧表: 现状回归,补列后物理列数 23 → 25。"""
+        """auto + 同款旧表: 现状回归,补列后物理列数 23 → 26。"""
         db = tmp_path / "auto_legacy.db"
         _make_pre_tenant_db(db)
 
@@ -639,10 +665,10 @@ class TestSQLiteSchemaMode:
         recorder.close()
 
         assert self._physical_columns(db) == _EXPECTED_COLUMNS
-        assert len(self._physical_columns(db)) == 25
+        assert len(self._physical_columns(db)) == 26
 
     async def test_manual_mode_still_creates_a_fresh_table(self, tmp_path):
-        """manual 只管 ALTER,不管 CREATE: 全新库照建,25 个物理列齐全(设计 §4.2)。"""
+        """manual 只管 ALTER,不管 CREATE: 全新库照建,26 个物理列齐全(设计 §4.2)。"""
         db = tmp_path / "manual_fresh.db"
         recorder = SQLiteRecorder(db, auto_migrate=False)
         await _record_minimal(recorder, call_id="c-fresh", tenant_id="tenant-a")
@@ -866,6 +892,7 @@ class TestPostgresBackfillDiscipline:
         "reasoning_tokens",
         "tenant_id",
         "meta",
+        "thinking_observation",
     ]
 
     def _recorder(self, conn):
@@ -926,6 +953,7 @@ class TestPostgresTableProbe:
         "reasoning_tokens",
         "tenant_id",
         "meta",
+        "thinking_observation",
     ]
 
     def _recorder(self, conn):
@@ -1118,6 +1146,62 @@ class TestEmitterRecorderContract:
         assert set(rec.rows[0]) == set(COLUMNS)
 
 
+class TestEmitterThinkingObservation:
+    """issue #16: 三态裁定经 emitter 落库,且落的是**裸 str** 而非枚举实例。
+
+    类型断言不是洁癖: `StrEnum` 虽是 `str` 子类,asyncpg 的参数编码对 `str`
+    子类不保证接受,而遥测写失败只降级成一条 warning——PG 那一路会静默少一列
+    数据,本地 SQLite 测试全绿也发现不了。归一化因此固定在 emitter 侧,与
+    `tenant_id`/`meta`/`sampling` 同一先例。
+    """
+
+    async def test_attempt_carries_the_verdict_as_a_plain_string(self):
+        rec = _MemoryRecorder()
+        await TelemetryEmitter(rec, text_cap=None).emit_attempt(
+            request=_REQ,
+            source=_source(),
+            call_id="c",
+            latency_ms=1,
+            response=_resp(thinking_observation=ThinkingObservation.OBSERVED),
+            error=None,
+        )
+        value = rec.rows[0]["thinking_observation"]
+        assert value == "observed"
+        assert type(value) is str  # 不是 ThinkingObservation: 子类实例不得下沉到 recorder
+
+    async def test_cache_hit_replays_the_recorded_verdict(self):
+        """缓存命中回放历史那次的裁定: 与 model/prompt_tokens 同一口径。"""
+        rec = _MemoryRecorder()
+        await TelemetryEmitter(rec, text_cap=None).emit_cache_hit(
+            request=_REQ,
+            response=_resp(cache_hit=True, thinking_observation=ThinkingObservation.ABSENT),
+        )
+        assert rec.rows[0]["thinking_observation"] == "absent"
+
+    async def test_terminal_failure_records_unknown(self):
+        """终态失败无响应可言,记 `unknown`——它恰好就是"观测不到",不撒谎。"""
+        rec = _MemoryRecorder()
+        await TelemetryEmitter(rec, text_cap=None).emit_terminal_failure(
+            request=_REQ, call_id="c", latency_ms=1, error="dead"
+        )
+        value = rec.rows[0]["thinking_observation"]
+        assert value == "unknown"
+        assert type(value) is str
+
+    async def test_failed_attempt_records_unknown(self):
+        """失败尝试(response=None)同理: 默认视图即 UNKNOWN。"""
+        rec = _MemoryRecorder()
+        await TelemetryEmitter(rec, text_cap=None).emit_attempt(
+            request=_REQ,
+            source=_source(),
+            call_id="c",
+            latency_ms=1,
+            response=None,
+            error="boom",
+        )
+        assert rec.rows[0]["thinking_observation"] == "unknown"
+
+
 class TestEmitterObservabilityFields:
     """issue #3: 三个入口各自的取值口径(设计 §5 表)。"""