fix: count every dropped SQLite telemetry row

SQLite 的逐行写入失败只发 warning、不计数,磁盘满 / database is locked /
文件被外部改坏时行真的丢了,而 dropped_rows 恒 0、degraded 恒 False——下游
按 README 的口径读快照对账完全看不见,与 issue #15 要消灭的静默失败同型。
同批修掉关闭后的丢行文案: 写死的遥测已降级与此时 degraded=False 的快照
互相矛盾,改为按状态分档(降级中 / 已关闭),与 PG 侧 _drop_reason 同口径。

src/polygateway/telemetry/sqlite.py

@@ -151,13 +151,26 @@ class SQLiteRecorder:
         if self._conn is None:
             # 改前这里是**裸 return**: 初始化失败后每一行都无声消失,长跑进程里
             # 与"遥测正常"外观上完全一致(设计 §1.4 的直接钉子)
-            self._status.record_drop("遥测已降级")
+            self._status.record_drop(self._drop_reason())
             return
         row = tuple(fields[col] for col in self._columns)
         try:
             await asyncio.to_thread(self._write, row)
         except (OSError, sqlite3.Error) as exc:
             logger.warning("SQLite 遥测写入失败(降级不冒泡): {}", exc)
+            # 计数与出声是两件事,少了计数可见性在这条路径上就是假的: 磁盘满 /
+            # database is locked / 文件被外部改坏时行真的丢了,而 `dropped_rows`
+            # 恒 0、`degraded` 恒 False,下游读快照对账完全看不见(PG 侧两件都做)
+            self._status.record_drop("写入失败")
+
+    def _drop_reason(self) -> str:
+        """连接为 None 时说清是**哪一种**写不进去: 降级中 / 调用方自己关了。
+
+        不能写死为"已降级": `close()` 之后 `degraded` 是 False,固定文案会与
+        下游读到的快照互相矛盾,对账的人分不清该等自愈还是修自己的关闭时序。
+        本侧只有这两态(初始化失败必置降级,此外只剩关闭),故不照抄 PG 的三分。
+        """
+        return "遥测已降级" if self._status.snapshot().degraded else "遥测已关闭"
 
     def _write(self, row: tuple) -> None:
         assert self._conn is not None  # 内部不变量: 调用方已判空

tests/unit/test_telemetry.py

@@ -1958,6 +1958,42 @@ class TestSQLiteStatusVisibility:
         assert recorder.telemetry_status.degraded is False
         recorder.close()
 
+    async def test_write_failure_counts_as_a_dropped_row(self, tmp_path, captured_warnings):
+        """逐行写入失败也必须计数,否则可见性在这条路径上是假的。
+
+        只发一条 warning 而不计数,`dropped_rows` 会恒 0、`degraded` 恒 False——
+        磁盘满 / database is locked / 文件被外部改坏时行真的丢了,而下游按
+        README 的口径("不必再靠人工对账")读快照完全看不见,与 issue #15
+        要消灭的静默失败同型。PG 侧两件都做(`postgres.py` 写入失败分支)。
+        """
+        db = tmp_path / "ok.db"
+        recorder = SQLiteRecorder(db, auto_migrate=True)
+        # 真实失败而非 mock: 另一连接把表删掉(等价于库文件被外部改坏),
+        # 此后 recorder 的每次 INSERT 都报 `no such table: llm_calls`
+        side = sqlite3.connect(db)
+        side.execute("DROP TABLE llm_calls")
+        side.commit()
+        side.close()
+        captured_warnings.clear()
+        await _record_minimal(recorder)  # 不得抛: 遥测绝不冒泡
+        assert recorder.telemetry_status.dropped_rows == 1
+        assert captured_warnings  # 丢的第一行必须出声
+        recorder.close()
+
+    async def test_drop_reason_after_close_matches_the_snapshot(self, tmp_path, captured_warnings):
+        """关闭后的丢行原因不得写死为"已降级": 此时快照里的 `degraded` 是 False。
+
+        固定文案与下游读到的快照互相矛盾,对账的人分不清该等后端自愈、还是
+        修自己的关闭时序。PG 侧用 `_drop_reason()` 分档,SQLite 侧必须同口径。
+        """
+        recorder = SQLiteRecorder(tmp_path / "ok.db", auto_migrate=True)
+        recorder.close()
+        captured_warnings.clear()
+        await _record_minimal(recorder)  # 不得抛
+        status = recorder.telemetry_status
+        assert status.degraded is False and status.dropped_rows == 1
+        assert "关闭" in captured_warnings[0] and "降级" not in captured_warnings[0]
+
 
 class TestPostgresStatusVisibility:
     """PG 侧的降级必须能被下游查到(计划 T2 建立可见性,T5 改判据)。"""