diff --git a/adapters/telemetry.py b/adapters/telemetry.py
index 8632402..5d24e50 100644
--- a/adapters/telemetry.py
+++ b/adapters/telemetry.py
@@ -10,6 +10,8 @@ import asyncio
 import sqlite3
 from typing import TYPE_CHECKING
 
+from loguru import logger
+
 if TYPE_CHECKING:
     from pathlib import Path
 
@@ -45,7 +47,7 @@ class SQLiteTelemetryRecorder:
     """
 
     _INSERT_SQL = """
-    INSERT INTO llm_calls (
+    INSERT OR IGNORE INTO llm_calls (
         call_id, parent_call_id, session_id,
         model_name, provider, messages, response, thinking,
         prompt_tokens, completion_tokens, latency_ms,
@@ -85,33 +87,46 @@ class SQLiteTelemetryRecorder:
         cache_hit: bool,
         error: str | None,
     ) -> None:
-        """同步写入一条 LLM 调用记录到 SQLite。"""
-        conn = sqlite3.connect(str(self._db_path))
+        """同步写入一条 LLM 调用记录到 SQLite。
+
+        三层防御加固：
+        1. INSERT OR IGNORE — 主键冲突静默忽略
+        2. WAL + busy_timeout — 并发写锁容忍
+        3. try/except sqlite3.Error — DB 错误不冒泡到调用方
+        """
         try:
-            self._ensure_table(conn)
-            conn.execute(
-                self._INSERT_SQL,
-                (
-                    call_id,
-                    parent_call_id,
-                    session_id,
-                    model_name,
-                    provider,
-                    messages,
-                    response,
-                    thinking,
-                    prompt_tokens,
-                    completion_tokens,
-                    latency_ms,
-                    ttft_ms,
-                    max_inter_token_ms,
-                    int(cache_hit),
-                    error,
-                ),
+            conn = sqlite3.connect(str(self._db_path), timeout=10.0)
+            try:
+                conn.execute("PRAGMA journal_mode=WAL")
+                conn.execute("PRAGMA busy_timeout=5000")
+                self._ensure_table(conn)
+                conn.execute(
+                    self._INSERT_SQL,
+                    (
+                        call_id,
+                        parent_call_id,
+                        session_id,
+                        model_name,
+                        provider,
+                        messages,
+                        response,
+                        thinking,
+                        prompt_tokens,
+                        completion_tokens,
+                        latency_ms,
+                        ttft_ms,
+                        max_inter_token_ms,
+                        int(cache_hit),
+                        error,
+                    ),
+                )
+                conn.commit()
+            finally:
+                conn.close()
+        except sqlite3.Error as exc:
+            logger.warning(
+                "遥测写入失败（已降级），call_id={}: {}", call_id, exc
             )
-            conn.commit()
-        finally:
-            conn.close()
 
     async def record_llm_call(
         self,
diff --git a/tests/unit/test_telemetry.py b/tests/unit/test_telemetry.py
index 2580c78..17a4883 100644
--- a/tests/unit/test_telemetry.py
+++ b/tests/unit/test_telemetry.py
@@ -104,3 +104,38 @@ async def test_record_cache_hit(recorder, db_path):
     conn.close()
 
     assert row["cache_hit"] == 1
+
+
+@pytest.mark.asyncio
+async def test_duplicate_call_id_does_not_raise(recorder, db_path):
+    """重复 call_id 写入应静默忽略（INSERT OR IGNORE），不抛异常。"""
+    kwargs = _make_call_kwargs()
+    await recorder.record_llm_call(**kwargs)
+    await recorder.record_llm_call(**kwargs)
+    conn = sqlite3.connect(str(db_path))
+    rows = conn.execute("SELECT COUNT(*) FROM llm_calls").fetchone()
+    conn.close()
+    assert rows[0] == 1
+
+
+@pytest.mark.asyncio
+async def test_db_error_does_not_propagate(tmp_path):
+    """SQLite 写入失败时 record_llm_call 应静默降级，不抛异常。"""
+    bad_recorder = SQLiteTelemetryRecorder(db_path=tmp_path / "nonexistent_dir" / "bad.db")
+    kwargs = _make_call_kwargs()
+    await bad_recorder.record_llm_call(**kwargs)
+
+
+@pytest.mark.asyncio
+async def test_concurrent_writes_no_lock_error(recorder, db_path):
+    """16 路并发 record_llm_call 应全部成功，无 database is locked 错误。"""
+    import asyncio
+    tasks = []
+    for _ in range(16):
+        kwargs = _make_call_kwargs()
+        tasks.append(recorder.record_llm_call(**kwargs))
+    await asyncio.gather(*tasks)
+    conn = sqlite3.connect(str(db_path))
+    count = conn.execute("SELECT COUNT(*) FROM llm_calls").fetchone()[0]
+    conn.close()
+    assert count == 16