docs(aibox): update README for v2.2 prompt system fixes

- Expand prompts directory listing to show all 10 skill files explicitly
- Update prompt system table: '1_*.md' → 'N_*.md (N=1~10)' with numeric ordering
- Document load order guarantee (identity → skills 1-10 → soul → status)
- Document is_continuation behavior (no repeated pre_prompt on recursive calls)
- Add v2.2 changelog: prompt ordering fix, token waste reduction,
  recursion pre_prompt dedup, large-file threshold, memory search tuning

aibox/README.md

@@ -85,11 +85,20 @@ aibox/
 ```
 ~/.aibox/
 ├── config.json              # 配置文件
-├── prompts/                 # 系统提示词目录
+├── prompts/                 # 系统提示词目录（按文件名数字前缀排序加载）
-│   ├── 00_IDENTITY.md       # 用户身份信息
+│   ├── 00_IDENTITY.md       # 用户身份信息（最先加载）
-│   ├── 1_*.md               # 基础技能提示词（文件操作、网页访问等）
+│   ├── 1_memory_search.md   # 基础技能提示词（1-10，按数字升序加载）
+│   ├── 2_memory_save.md
+│   ├── 3_file_operations.md
+│   ├── 4_web_access.md
+│   ├── 5_memo_management.md
+│   ├── 6_command_execution.md
+│   ├── 7_extension_skills.md
+│   ├── 8_identity_management.md
+│   ├── 9_language_instruction.md
+│   ├── 10_task_handling.md
 │   ├── 99_SOUL.md           # AI 灵魂定义（角色、性格、能力边界）
-│   └── 100_STATUS.md        # AI 状态记录
+│   └── 100_STATUS.md        # AI 状态记录（最后加载）
 ├── skills/                  # 扩展技能目录（运行时从仓库 skills/ 复制或用户自建）
 ├── memories.db              # 记忆和任务数据库（SQLite）
 ├── current_conversation.json # 当前对话
@@ -364,15 +373,17 @@ AI 可以通过 `update_soul` 工具动态更新 `99_SOUL.md`，通过 `update_i
 
 ## 🧩 提示词系统
 
-提示词文件位于 `~/.aibox/prompts/`：
+提示词文件位于 `~/.aibox/prompts/`，按文件名**数字前缀升序**加载（identity → 技能 1~10 → soul → status）：
 
 | 文件 | 用途 | 更新方式 |
 |-----|------|---------|
 | `00_IDENTITY.md` | 用户身份信息 | AI 通过 `update_identity` 工具更新 |
-| `1_*.md` | 基础技能提示词 | 手动编辑，AI 通过 `reload_prompts` 重载 |
+| `N_*.md`（N=1~10） | 基础技能提示词（记忆搜索/保存、文件/网页/命令/任务操作、技能扩展、身份管理、语言、任务规则） | 手动编辑，AI 通过 `/reload_prompts` 重载 |
-| `99_SOUL.md` | AI 灵魂定义（角色、性格） | AI 通过 `update_soul` 工具更新 |
+| `99_SOUL.md` | AI 灵魂定义（角色、性格、能力边界、行为准则） | AI 通过 `update_soul` 工具更新 |
 | `100_STATUS.md` | AI 状态记录 | AI 通过 `save_memory_and_end_conversation` 更新 |
 
+> **加载顺序保证**：`00_IDENTITY.md` 最先加载（AI 先知道用户是谁），然后是 1~10 号技能提示词（按数字升序），接着是 `99_SOUL.md`（AI 的角色定位），最后是 `100_STATUS.md`（当前状态）。每轮用户输入前会附加一段简洁的工作流程提醒（不重复 system prompt 中的详细说明），工具调用后递归继续时不再重复注入。
+
 ## ⚙️ 配置说明
 
 主要配置项（`~/.aibox/config.json`，首次启动自动生成）：
@@ -673,6 +684,14 @@ elif command == "/mycommand":
 
 ## 📄 更新日志
 
+### v2.2
+- ✅ 修复提示词系统排序 bug：`SKILL_PREFIX = "1_"` 只匹配 `1_` 开头文件，导致 `2_`~`10_` 技能提示词排序错乱（`10_task_handling.md` 排在 `1_memory_search.md` 前面）；改用正则按数字前缀排序
+- ✅ 修复 `_init_system_prompts` 使用未排序列表的问题，确保 system prompt 按 identity → 技能 → soul → status 正确顺序加载
+- ✅ 精简每轮用户输入的 pre_prompt：原来约 30 行重复指令（与 system prompt 内容重复），每轮浪费 ~300 token；精简为 5 条简洁提醒
+- ✅ 修复工具调用后递归继续时重复注入工作流指令的问题：新增 `is_continuation` 参数，递归时跳过 pre_prompt
+- ✅ 优化默认大文件阈值：50KB/200行 → 500KB/2000行，增加 `tail=true` 和二进制文件处理指导
+- ✅ 优化记忆搜索习惯："每次都要做"→"根据话题相关性判断"，避免简单问候也触发无意义搜索
+
 ### v2.1
 - ✅ 修复工具调用结果破坏对话导致 API 持续请求失败的问题
   - 不再回传 `reasoning_content`（API 响应字段）到请求中

aibox/aibox.py

@@ -436,7 +436,7 @@ class PromptManager:
     
     文件命名规范:
     - 00_IDENTITY.md: 用户身份信息
-    - 1_*.md/txt: 基础技能（必须从1_开头）
+    - N_*.md/txt: 基础技能提示词（N 为数字，如 1_memory_search.md、10_task_handling.md），按数字升序加载
     - 99_SOUL.md: 智能体灵魂文件（角色定位、身份定义、性格等）
     - 100_STATUS.md: 智能体状态记录
     """
@@ -450,8 +450,8 @@ class PromptManager:
     STATUS_FILE = "100_STATUS.md"
     SOUL_FILE = "99_SOUL.md"  # 智能体灵魂文件 - 所有角色定位内容
     
-    # 基础技能前缀
+    # 基础技能文件名模式：N_xxx.md（N 为数字，如 1_memory_search.md、10_task_handling.md）
-    SKILL_PREFIX = "1_"  # 基础技能文件必须以这个开头
+    _SKILL_FILE_RE = re.compile(r'^(\d+)_')
     
     def __new__(cls):
         if cls._instance is None:
@@ -642,14 +642,17 @@ class PromptManager:
                 print(f"⚠️ 创建默认灵魂文件失败: {e}")
     
     def _ensure_skill_files(self):
-        """确保基础技能文件存在 - 以1_开头的文件"""
+        """确保基础技能文件存在 - N_ 开头的文件（N 为数字，按数字顺序加载）"""
         skill_files = [
             {
                 'filename': '1_memory_search.md',
                 'content': """# 记忆搜索习惯
 
-## 重要习惯
+## 核心习惯
-在每次回答用户问题前，**必须**先使用 `memory search` 工具搜索记忆库中与当前话题相关的记忆。
+在回答用户问题前，**应根据话题相关性判断是否需要搜索记忆**：
+- 涉及过往讨论、用户偏好、之前做过的事 → 必须搜索
+- 全新话题、简单问候（"你好"）、确认回应（"好的""明白了"）→ 无需搜索
+- 不确定时，搜索一下也无妨
 
 ## 为什么要这样做？
 - 保持对话的连贯性
@@ -665,11 +668,9 @@ memory search keywords=["关键词1", "关键词2"]
 ## 示例
 如果用户说："上次我们讨论的那个项目怎么样了？"
 你应该：
-1. 先搜索 "项目" 相关的记忆
+1. 搜索 "项目" 相关的记忆
 2. 根据搜索结果回答
 3. 如果找不到相关记忆，诚实告知用户
-
-记住：这是你的核心习惯，每次都要做！
 """
             },
             {
@@ -711,19 +712,22 @@ save_memory_and_end_conversation(
 - **list_files**: 列出目录
 
 ## 大文件处理规则
-当遇到以下情况时，**必须先告知用户文件过大**：
+当遇到以下情况时，**必须分页读取**，不要一次性读取整个文件：
-1. 文件大小超过 50KB
+1. 文件大小超过 500KB
-2. 文件行数超过 200 行
+2. 文件行数超过 2000 行
 
 对于大文件，应该：
-- 使用 head/tail 参数只读取开头或结尾
+- 先用 head=true 读取前50行了解文件结构
 - 使用 start_line/end_line 参数指定行范围
-- 建议用户提供更具体的查询条件
+- 使用 tail=true 读取文件末尾
+- 对于二进制文件或特殊格式文件，使用 run_command 配合 head/strings 等命令查看
+- 建议用户提供更具体的查询条件以缩小范围
 
 ## 示例
 ```
 read_file filepath="/path/to/file.txt" head=true max_lines=50
-read_file filepath="/path/to/file.txt" start_line=10 end_line=20
+read_file filepath="/path/to/file.txt" start_line=100 end_line=200
+read_file filepath="/path/to/file.txt" tail=true max_lines=50
 ```
 """
             },
@@ -1047,33 +1051,46 @@ update_soul content="我是...（新的自我定义）"
         """创建默认提示词（已由各 _ensure_* 方法处理）"""
         pass
     
+    @classmethod
+    def _skill_sort_key(cls, filename: str):
+        """技能文件按前缀数字排序，非技能文件返回 None。"""
+        m = cls._SKILL_FILE_RE.match(filename)
+        return int(m.group(1)) if m else None
+
+    def _prompt_sort_key(self, p):
+        """排序规则：00_IDENTITY → 技能文件(按数字升序) → 99_SOUL → 100_STATUS → 其他"""
+        file = p['file']
+        if file == self.IDENTITY_FILE:
+            return (0,)
+        skill_num = self._skill_sort_key(file)
+        if skill_num is not None:
+            return (1, skill_num)
+        if file == self.SOUL_FILE:
+            return (2,)
+        if file == self.STATUS_FILE:
+            return (4,)
+        return (3, file)  # 其他文件按名字排序
+
     def get_prompts(self, join_with: str = "\n\n") -> str:
-        """获取所有提示词，按顺序拼接"""
+        """获取所有提示词，按顺序拼接（identity → 技能文件 → soul → status）"""
         if not self._prompts:
             return ""
-        
+        sorted_prompts = sorted(self._prompts, key=self._prompt_sort_key)
-        # 排序规则：先基础技能（1_开头），再其他，最后灵魂文件
-        def sort_key(p):
-            file = p['file']
-            if file == self.SOUL_FILE:
-                return 3  # 灵魂文件最后加载
-            elif file.startswith(self.SKILL_PREFIX):
-                return 1  # 基础技能
-            elif file == self.IDENTITY_FILE:
-                return 0  # 身份文件最先
-            elif file == self.STATUS_FILE:
-                return 4  # 状态文件最后
-            else:
-                return 2
-        
-        sorted_prompts = sorted(self._prompts, key=sort_key)
         contents = [p['content'] for p in sorted_prompts]
         return join_with.join(contents)
 
+    def get_sorted_prompts_list(self) -> List[Dict]:
+        """获取按正确顺序排列的提示词列表（用于 system 消息构建）。"""
+        return sorted(self._prompts, key=self._prompt_sort_key)
+
     def get_prompts_list(self) -> List[Dict]:
-        """获取所有提示词列表"""
+        """获取所有提示词列表（原始加载顺序，已弃用，请用 get_sorted_prompts_list）。"""
         return self._prompts.copy()
 
+    def get_skill_prompts(self) -> List[Dict]:
+        """获取所有基础技能提示词（N_ 开头的文件）。"""
+        return [p for p in self._prompts if self._skill_sort_key(p['file']) is not None]
+
     def reload(self):
         """重新加载所有提示词"""
         self._prompts = []
@@ -6536,19 +6553,19 @@ class DeepSeekChat:
         self._register_tools(exit_callback, command_executor)
     
     def _init_system_prompts(self, system_prompt: str = None):
-        """初始化系统提示词 - 从 PromptManager 加载"""
+        """初始化系统提示词 - 从 PromptManager 加载（按 identity → 技能 → soul → status 顺序）"""
         if system_prompt:
             # 如果明确指定了 system_prompt，使用它
             self._add_message("system", system_prompt)
             self.logger.info("使用指定的系统提示词")
         else:
-            # 从 PromptManager 加载所有提示词
+            # 从 PromptManager 加载所有提示词，按正确顺序排列
-            prompts = self.prompt_manager.get_prompts_list()
+            prompts = self.prompt_manager.get_sorted_prompts_list()
             if prompts:
                 for prompt in prompts:
                     if prompt['content'].strip():
                         self._add_message("system", prompt['content'].strip())
-                self.logger.info(f"已加载 {len(prompts)} 个系统提示词")
+                self.logger.info(f"已加载 {len(prompts)} 个系统提示词（顺序: identity → 技能 → soul → status）")
             else:
                 # 如果没有任何提示词，添加一个默认的
                 default_prompt = "你是一个智能AI助手。"
@@ -6847,8 +6864,15 @@ class DeepSeekChat:
         # 默认情况下，如果记忆库存在，搜索一下也无妨
         return True
     
-    def think_and_respond(self, input_text: str, output_callback: Callable = None) -> Optional[str]:
+    def think_and_respond(self, input_text: str, output_callback: Callable = None, is_continuation: bool = False) -> Optional[str]:
-        """思考并回应 - 支持实时流式输出，支持思考模式"""
+        """思考并回应 - 支持实时流式输出，支持思考模式。
+        
+        Args:
+            input_text: 用户输入或继续提示
+            output_callback: 流式输出回调
+            is_continuation: 是否为工具调用后的递归继续；为 True 时不注入冗长的工作流指令，
+                            因为 system prompt 中已有完整行为规范，继续时只需传递工具结果
+        """
         if not self.conversation_active:
             self.logger.info("对话已结束，重置状态")
             self.conversation_active = True
@@ -6880,54 +6904,14 @@ class DeepSeekChat:
         
         self.stats["api_calls"] += 1
         
-        # 前置提示词
+        if is_continuation:
-        pre_prompt = """【系统指令 - 核心工作流程】
+            # 递归继续（工具调用后）：直接使用输入文本，不重复注入工作流指令
-
+            # system prompt 中已有完整行为规范，此处只需传递工具结果的上下文
-        在开始任何任务之前，你必须严格按照以下优先级执行：
+            enhanced_input = input_text
-
+        else:
-        ## 第一步：搜索可用技能（最高优先级）⭐
+            # 首轮回话：附加简洁的工作流程提醒（不重复 system prompt 中已有的详细说明）
-        当用户提出任何任务、需求或问题时，**必须先使用 `search_skills` 工具**搜索相关技能：
+            pre_prompt = "【工作流程提醒】\n1. 如果用户提出新任务，先用 search_skills 搜索是否有现成技能可用\n2. 根据话题相关性决定是否 memory search（简单问候/确认无需搜索）\n3. 信息不足时用 ask_user 提问，不要猜测\n4. 遇到大文件用分页读取（head/tail/start_line/end_line），不要直接打开大文件\n5. 话题完成时用 save_memory_and_end_conversation 保存记忆并结束"
-        - 使用 `search_skills keyword="[任务关键词]"` 搜索是否有现成的扩展技能可以完成该任务
+            enhanced_input = pre_prompt + "\n\n" + input_text
-        - 例如：用户说"帮我查天气"，你应该先执行 `search_skills keyword="天气"`
-        - 例如：用户说"帮我计算BMI"，你应该先执行 `search_skills keyword="BMI"`
-        - 如果搜索结果中有匹配的技能，优先使用该技能完成任务
-        - 只有在没有合适技能的情况下，才考虑使用其他工具或方法
-
-        ## 第二步：搜索相关记忆
-        在每次回答用户问题前，**必须**使用 `memory search` 工具搜索记忆库中与当前话题相关的记忆，确保回答的连贯性和准确性。
-
-        ## 第三步：积极向用户提问
-        如果你有任何无法确定的事情，那么你需要用户提供额外信息，
-        请务必使用 `ask_user` 工具向用户提问，以便获得更准确的指引。
-
-        ## 第四步：执行任务或回答问题
-        根据搜索结果，选择合适的工具或直接回答。
-
-        【重要提醒】
-        - 技能优先原则：有技能先用技能，不要自己硬编码实现
-        - 如果没有合适的技能，可以使用其他工具或直接回答
-        - 对话结束时，必须使用 `save_memory_and_end_conversation` 工具保存本次对话的重要信息并更新智能体状态
-        - 当了解到用户的身份信息时，使用 `update_identity` 工具更新 00_IDENTITY.md
-        - 当你的角色、性格、能力边界需要调整时，使用 `update_soul` 工具更新 99_SOUL.md 文件来重新定义自己
-        - 绝对不要直接打开大文件来读，应该使用额外的压缩工具
-
-        现在请处理用户的输入："""
-        
-        # 后置提示词
-        post_prompt = """
-        
-                【注意】
-                当本次对话的话题讨论完成时，请使用 `save_memory_and_end_conversation` 工具结束对话并保存重要信息到记忆库，同时更新你的状态到 100_STATUS.md。
-        
-                【自我认知更新】
-                在对话过程中，如果你发现：
-                - 自己的角色定位需要调整
-                - 性格特点需要改变
-                - 能力边界需要重新定义
-                - 行为准则需要优化
-                请使用 `update_soul` 工具更新 99_SOUL.md 文件，这有助于你更好地理解自己并为用户提供更精准的服务。"""
-        
-        enhanced_input = pre_prompt + "\n\n" + input_text + post_prompt
         
         # 记录本轮开始前的消息长度，用于 API 失败时回滚，防止坏消息污染对话历史
         msgs_before_turn = len(self.messages)
@@ -7146,9 +7130,8 @@ class DeepSeekChat:
                 if output_callback:
                     output_callback("line", i18n.get('analyzing_tool_results') + "\n")
                 
-                # 【关键修复4】递归调用，让AI基于工具结果继续回答
+                # 递归调用，让AI基于工具结果继续回答；is_continuation=True 避免重复注入工作流指令
-                # 使用一个新的输入提示，避免重复
+                return self.think_and_respond("请基于以上工具执行结果继续回答。", output_callback, is_continuation=True)
-                return self.think_and_respond("请基于以上工具执行结果继续回答用户的问题。", output_callback)
             
             elif full_response:
                 # 正常添加助手消息，包含 reasoning_content

confs/README.md

@@ -181,6 +181,7 @@ port = resp["data"]["value"]
 
 - **扩展名白名单**：仅允许 `.ini/.conf/.cfg/.json/.yaml/.yml/.toml/.txt`，其余一律拒绝
 - **路径遍历防护**：配置名必须等于 `os.path.basename(name)`，且 realpath 必须位于 `configs/` 目录前缀之下
+- **文件大小限制**：单配置文件上限 1 MiB（`MAX_CONFIG_SIZE`），超过则拒绝读取并返回错误，防止超大文件导致内存溢出
 - **仅监听本机**：默认 `127.0.0.1`，不对外暴露，仅供内部服务调用
 - **无鉴权**：当前无身份校验，依赖"仅本机 + 内部网络"的信任边界；如需承载敏感配置（如 API 密钥），应额外评估鉴权方案
 
@@ -224,5 +225,6 @@ confs/
 - 配置文件需手动放入 `configs/` 目录（或用 `--dir` 指定其他目录），服务不会自动生成业务配置
 - 文件以 UTF-8 读取，非 UTF-8 文件会返回 `配置读取失败` 错误
 - 配置名区分大小写（与文件系统一致）
+- 单配置文件大小上限为 1 MiB（`MAX_CONFIG_SIZE`），超过此限制会返回 `配置文件过大` 错误，防止超大文件读入内存导致 OOM
 - 缓存基于文件 mtime 判断变化，若同秒内修改且 mtime 精度不足（如某些 FAT 文件系统），可能无法及时重载——可手动调 `reload` 路由强制刷新
 - 当前接入情况见上文"集成关系 → 已接入的服务"；新服务接入可参考 static 的三层 fail-safe 模式（内置默认 → confs 优先 → 本地回退）

confs/confs.py

@@ -42,6 +42,8 @@ from log_client import logs  # noqa: E402
 CONFIG_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "configs")
 # 允许分发的配置文件类型（白名单）
 ALLOWED_EXT = {".ini", ".conf", ".cfg", ".json", ".yaml", ".yml", ".toml", ".txt"}
+# 单配置文件大小上限（1 MiB），防止超大文件读入内存导致 OOM
+MAX_CONFIG_SIZE = 1 * 1024 * 1024
 
 
 # ── 异常 ────────────────────────────────────────────────────────────────
@@ -103,13 +105,19 @@ class ConfigStore:
         path = self._resolve(name)
         with self._lock:
             try:
-                mtime = os.path.getmtime(path)
+                st = os.stat(path)
+                mtime = st.st_mtime
+                size = st.st_size
             except OSError:
                 self._cache.pop(name, None)
                 raise ConfigNotFound(f"配置不存在或不可读: {name}") from None
             entry = self._cache.get(name)
             if entry is not None and entry["mtime"] == mtime:
                 return entry
+            if size > MAX_CONFIG_SIZE:
+                raise ConfigError(
+                    f"配置文件过大: {name}（{size} 字节，上限 {MAX_CONFIG_SIZE} 字节）"
+                )
             try:
                 with open(path, "r", encoding="utf-8") as f:
                     content = f.read()

confs/log_client.py

@@ -1,10 +1,11 @@
-"""统一日志客户端 — 各服务共用（dnfs / smtp / static 三份副本内容一致，改动请同步）。
+"""统一日志客户端 — 各服务共用（dnfs / smtp / static / confs 四份副本内容一致，改动请同步）。
 
 设计要点：
   - 通过 pymsg RPC 调用 dbv2 execute 路由，逐条 INSERT 到 Logs 表
   - 异步非阻塞：内存队列 + 后台线程批量推送，不拖慢业务
-  - fail-safe：db 服务不可用时静默丢弃，绝不影响业务
+  - fail-safe：db 服务不可用时静默丢弃，绝不影响业务；后台线程任何异常都不会崩溃
   - 缓冲批量：队列满 50 条或超时 2 秒时自动 flush
+  - 时间戳：日志入队时即记录 ts，反映事件真实发生时间
 
 用法：
     from log_client import logs
@@ -35,6 +36,20 @@ _DBV2_PORT = int(_os.environ.get('DB_PORT', '8005'))
 _BATCH_SIZE = 50         # 队列满 50 条时 flush
 _FLUSH_INTERVAL = 2.0    # 超时 2 秒强制 flush
 _TIMEOUT = 5             # socket 超时（秒）
+_QUEUE_MAXSIZE = 10000   # 队列上限，防止后台线程异常时内存无限增长；满则丢弃最旧条目
+
+
+def _safe_json(obj):
+    """安全序列化 extra 字段，不可序列化对象转为 repr 字符串，保证不抛异常。"""
+    try:
+        return json.dumps(obj, ensure_ascii=False, default=str)
+    except Exception:
+        return repr(obj)
+
+
+def _now_iso():
+    """生成当前 UTC 时间 ISO 字符串（在入队时调用，保证时间戳准确）。"""
+    return datetime.now(timezone.utc).isoformat()
 
 
 class LogClient:
@@ -46,9 +61,10 @@ class LogClient:
         self._service_name = service_name
         self._batch_size = batch_size
         self._flush_interval = flush_interval
-        self._queue = queue.Queue()
+        self._queue = queue.Queue(maxsize=_QUEUE_MAXSIZE)
         self._stop = threading.Event()
         self._flush_signal = threading.Event()
+        self._flush_done = threading.Event()
         self._thread = threading.Thread(target=self._flush_loop,
                                         daemon=True, name='log-client')
         self._thread.start()
@@ -57,14 +73,30 @@ class LogClient:
     # ── 公开接口 ────────────────────────────────────────────────────────
 
     def push(self, service, level, message, trace_id=None, extra=None):
-        """推送一条日志（非阻塞，立即返回）。"""
+        """推送一条日志（非阻塞，立即返回）。
-        self._queue.put({
+
+        队列满时静默丢弃最旧条目再放入新条目，保证不阻塞业务线程、不导致 OOM。
+        """
+        entry = {
+            'ts': _now_iso(),  # 入队时即生成时间戳，反映真实事件时间
             'service': service,
             'level': level,
             'message': message,
             'trace_id': trace_id,
             'extra': extra,
-        })
+        }
+        try:
+            self._queue.put_nowait(entry)
+        except queue.Full:
+            # 队列满：丢弃最旧的一条，再放入新条目（drop-oldest 策略）
+            try:
+                self._queue.get_nowait()
+            except queue.Empty:
+                pass
+            try:
+                self._queue.put_nowait(entry)
+            except queue.Full:
+                pass  # 极端竞争下仍失败则静默丢弃
 
     def debug(self, service, message, **kw):
         self.push(service, 'DEBUG', message, **kw)
@@ -83,8 +115,15 @@ class LogClient:
 
     def flush(self):
         """手动 flush 队列中剩余的日志（含后台线程持有的局部 batch）。"""
+        self._flush_done.clear()
         self._flush_signal.set()  # 通知后台线程立即 flush 其局部 batch
-        time.sleep(0.6)  # 等待后台线程响应
+        # 等待后台线程完成 flush（最多等待 flush_interval + 2 秒余量）
+        self._flush_done.wait(timeout=self._flush_interval + 2.0)
+        # 主线程排空队列中剩余条目
+        self._drain_queue_to_db()
+
+    def _drain_queue_to_db(self):
+        """排空主线程可见的队列中剩余条目，直接发送到 db。"""
         batch = []
         while True:
             try:
@@ -96,60 +135,92 @@ class LogClient:
                 break
         if batch:
             self._send(batch)
-        self._flush_signal.clear()
 
     # ── 内部实现 ────────────────────────────────────────────────────────
 
     def _flush_loop(self):
-        """后台线程：积攒到 batch_size 或超时 flush_interval 时发送。"""
+        """后台线程：积攒到 batch_size 或超时 flush_interval 时发送。
+
+        外层捕获所有异常，保证后台线程在任何情况下（网络故障、序列化错误等）
+        都不会死亡，避免日志队列无限增长导致 OOM。
+        """
         batch = []
         last_flush = time.monotonic()
 
         while not self._stop.is_set():
-            timeout = self._flush_interval - (time.monotonic() - last_flush)
-            chk_timeout = max(0.1, min(timeout, 0.5))
             try:
-                item = self._queue.get(timeout=chk_timeout)
+                timeout = self._flush_interval - (time.monotonic() - last_flush)
-                batch.append(item)
+                chk_timeout = max(0.1, min(timeout, 0.5))
-                while len(batch) < self._batch_size:
+                try:
-                    try:
+                    item = self._queue.get(timeout=chk_timeout)
-                        batch.append(self._queue.get_nowait())
+                    batch.append(item)
-                    except queue.Empty:
+                    while len(batch) < self._batch_size:
-                        break
+                        try:
-            except queue.Empty:
+                            batch.append(self._queue.get_nowait())
-                pass
+                        except queue.Empty:
+                            break
+                except queue.Empty:
+                    pass
 
-            should_flush = (
+                should_flush = (
-                len(batch) >= self._batch_size or
+                    len(batch) >= self._batch_size or
-                self._flush_signal.is_set() or
+                    self._flush_signal.is_set() or
-                (batch and time.monotonic() - last_flush >= self._flush_interval)
+                    (batch and time.monotonic() - last_flush >= self._flush_interval)
-            )
+                )
-            if should_flush and batch:
+                if should_flush and batch:
-                self._send(batch)
+                    try:
-                batch = []
+                        self._send(batch)
-                last_flush = time.monotonic()
+                    except Exception as e:
+                        # _send 内部已有细粒度异常处理，此处是最终安全网
+                        print(f"[log_client] _send 异常（已捕获，线程继续运行）: {e}",
+                              file=_sys.stderr)
+                    batch = []
+                    last_flush = time.monotonic()
+                    if self._flush_signal.is_set():
+                        self._flush_signal.clear()
+                        self._flush_done.set()
+            except Exception as e:
+                # 最外层安全网：任何意外异常都不应杀死日志线程
+                print(f"[log_client] _flush_loop 异常（已捕获，线程继续运行）: {e}",
+                      file=_sys.stderr)
+                time.sleep(0.5)  # 防止异常风暴导致 CPU 满载
 
         # 退出前 flush 剩余
         if batch:
-            self._send(batch)
+            try:
+                self._send(batch)
+            except Exception as e:
+                print(f"[log_client] 退出时 flush 异常: {e}", file=_sys.stderr)
+        self._flush_done.set()
 
     def _send(self, batch):
-        """通过 pymsg RPC（execute 路由）逐条 INSERT 到 dbv2 Logs 表。"""
+        """通过 pymsg RPC（execute 路由）批量发送到 dbv2 Logs 表。
-        now = datetime.now(timezone.utc).isoformat()
+
-        client = RpcClient(_DBV2_ADDR, _DBV2_PORT, self._service_name + "-log")
+        整个 batch 复用一个 TCP 连接；网络错误时放弃本批剩余条目（fail-safe），
+        不抛异常给调用方。
+        """
+        try:
+            client = RpcClient(_DBV2_ADDR, _DBV2_PORT, self._service_name + "-log")
+        except Exception as e:
+            print(f"[log_client] 创建 RpcClient 失败: {e}", file=_sys.stderr)
+            return
+
         for entry in batch:
             extra_val = entry.get("extra")
+            try:
+                extra_json = _safe_json(extra_val) if extra_val is not None else None
+            except Exception:
+                extra_json = None
             payload = {
                 "action": "insert",
                 "table": "Logs",
                 "values": {
-                    "ts": now,
+                    "ts": entry.get("ts", _now_iso()),
                     "service": entry.get("service"),
                     "level": entry.get("level"),
                     "message": entry.get("message"),
                     "trace_id": entry.get("trace_id"),
-                    "extra": json.dumps(extra_val, ensure_ascii=False)
+                    "extra": extra_json,
-                    if extra_val is not None else None,
                 },
             }
             try:

dnfs/log_client.py

@@ -1,10 +1,11 @@
-"""统一日志客户端 — 各服务共用（dnfs / smtp / static 三份副本内容一致，改动请同步）。
+"""统一日志客户端 — 各服务共用（dnfs / smtp / static / confs 四份副本内容一致，改动请同步）。
 
 设计要点：
   - 通过 pymsg RPC 调用 dbv2 execute 路由，逐条 INSERT 到 Logs 表
   - 异步非阻塞：内存队列 + 后台线程批量推送，不拖慢业务
-  - fail-safe：db 服务不可用时静默丢弃，绝不影响业务
+  - fail-safe：db 服务不可用时静默丢弃，绝不影响业务；后台线程任何异常都不会崩溃
   - 缓冲批量：队列满 50 条或超时 2 秒时自动 flush
+  - 时间戳：日志入队时即记录 ts，反映事件真实发生时间
 
 用法：
     from log_client import logs
@@ -35,6 +36,20 @@ _DBV2_PORT = int(_os.environ.get('DB_PORT', '8005'))
 _BATCH_SIZE = 50         # 队列满 50 条时 flush
 _FLUSH_INTERVAL = 2.0    # 超时 2 秒强制 flush
 _TIMEOUT = 5             # socket 超时（秒）
+_QUEUE_MAXSIZE = 10000   # 队列上限，防止后台线程异常时内存无限增长；满则丢弃最旧条目
+
+
+def _safe_json(obj):
+    """安全序列化 extra 字段，不可序列化对象转为 repr 字符串，保证不抛异常。"""
+    try:
+        return json.dumps(obj, ensure_ascii=False, default=str)
+    except Exception:
+        return repr(obj)
+
+
+def _now_iso():
+    """生成当前 UTC 时间 ISO 字符串（在入队时调用，保证时间戳准确）。"""
+    return datetime.now(timezone.utc).isoformat()
 
 
 class LogClient:
@@ -46,9 +61,10 @@ class LogClient:
         self._service_name = service_name
         self._batch_size = batch_size
         self._flush_interval = flush_interval
-        self._queue = queue.Queue()
+        self._queue = queue.Queue(maxsize=_QUEUE_MAXSIZE)
         self._stop = threading.Event()
         self._flush_signal = threading.Event()
+        self._flush_done = threading.Event()
         self._thread = threading.Thread(target=self._flush_loop,
                                         daemon=True, name='log-client')
         self._thread.start()
@@ -57,14 +73,30 @@ class LogClient:
     # ── 公开接口 ────────────────────────────────────────────────────────
 
     def push(self, service, level, message, trace_id=None, extra=None):
-        """推送一条日志（非阻塞，立即返回）。"""
+        """推送一条日志（非阻塞，立即返回）。
-        self._queue.put({
+
+        队列满时静默丢弃最旧条目再放入新条目，保证不阻塞业务线程、不导致 OOM。
+        """
+        entry = {
+            'ts': _now_iso(),  # 入队时即生成时间戳，反映真实事件时间
             'service': service,
             'level': level,
             'message': message,
             'trace_id': trace_id,
             'extra': extra,
-        })
+        }
+        try:
+            self._queue.put_nowait(entry)
+        except queue.Full:
+            # 队列满：丢弃最旧的一条，再放入新条目（drop-oldest 策略）
+            try:
+                self._queue.get_nowait()
+            except queue.Empty:
+                pass
+            try:
+                self._queue.put_nowait(entry)
+            except queue.Full:
+                pass  # 极端竞争下仍失败则静默丢弃
 
     def debug(self, service, message, **kw):
         self.push(service, 'DEBUG', message, **kw)
@@ -83,8 +115,15 @@ class LogClient:
 
     def flush(self):
         """手动 flush 队列中剩余的日志（含后台线程持有的局部 batch）。"""
+        self._flush_done.clear()
         self._flush_signal.set()  # 通知后台线程立即 flush 其局部 batch
-        time.sleep(0.6)  # 等待后台线程响应
+        # 等待后台线程完成 flush（最多等待 flush_interval + 2 秒余量）
+        self._flush_done.wait(timeout=self._flush_interval + 2.0)
+        # 主线程排空队列中剩余条目
+        self._drain_queue_to_db()
+
+    def _drain_queue_to_db(self):
+        """排空主线程可见的队列中剩余条目，直接发送到 db。"""
         batch = []
         while True:
             try:
@@ -96,60 +135,92 @@ class LogClient:
                 break
         if batch:
             self._send(batch)
-        self._flush_signal.clear()
 
     # ── 内部实现 ────────────────────────────────────────────────────────
 
     def _flush_loop(self):
-        """后台线程：积攒到 batch_size 或超时 flush_interval 时发送。"""
+        """后台线程：积攒到 batch_size 或超时 flush_interval 时发送。
+
+        外层捕获所有异常，保证后台线程在任何情况下（网络故障、序列化错误等）
+        都不会死亡，避免日志队列无限增长导致 OOM。
+        """
         batch = []
         last_flush = time.monotonic()
 
         while not self._stop.is_set():
-            timeout = self._flush_interval - (time.monotonic() - last_flush)
-            chk_timeout = max(0.1, min(timeout, 0.5))
             try:
-                item = self._queue.get(timeout=chk_timeout)
+                timeout = self._flush_interval - (time.monotonic() - last_flush)
-                batch.append(item)
+                chk_timeout = max(0.1, min(timeout, 0.5))
-                while len(batch) < self._batch_size:
+                try:
-                    try:
+                    item = self._queue.get(timeout=chk_timeout)
-                        batch.append(self._queue.get_nowait())
+                    batch.append(item)
-                    except queue.Empty:
+                    while len(batch) < self._batch_size:
-                        break
+                        try:
-            except queue.Empty:
+                            batch.append(self._queue.get_nowait())
-                pass
+                        except queue.Empty:
+                            break
+                except queue.Empty:
+                    pass
 
-            should_flush = (
+                should_flush = (
-                len(batch) >= self._batch_size or
+                    len(batch) >= self._batch_size or
-                self._flush_signal.is_set() or
+                    self._flush_signal.is_set() or
-                (batch and time.monotonic() - last_flush >= self._flush_interval)
+                    (batch and time.monotonic() - last_flush >= self._flush_interval)
-            )
+                )
-            if should_flush and batch:
+                if should_flush and batch:
-                self._send(batch)
+                    try:
-                batch = []
+                        self._send(batch)
-                last_flush = time.monotonic()
+                    except Exception as e:
+                        # _send 内部已有细粒度异常处理，此处是最终安全网
+                        print(f"[log_client] _send 异常（已捕获，线程继续运行）: {e}",
+                              file=_sys.stderr)
+                    batch = []
+                    last_flush = time.monotonic()
+                    if self._flush_signal.is_set():
+                        self._flush_signal.clear()
+                        self._flush_done.set()
+            except Exception as e:
+                # 最外层安全网：任何意外异常都不应杀死日志线程
+                print(f"[log_client] _flush_loop 异常（已捕获，线程继续运行）: {e}",
+                      file=_sys.stderr)
+                time.sleep(0.5)  # 防止异常风暴导致 CPU 满载
 
         # 退出前 flush 剩余
         if batch:
-            self._send(batch)
+            try:
+                self._send(batch)
+            except Exception as e:
+                print(f"[log_client] 退出时 flush 异常: {e}", file=_sys.stderr)
+        self._flush_done.set()
 
     def _send(self, batch):
-        """通过 pymsg RPC（execute 路由）逐条 INSERT 到 dbv2 Logs 表。"""
+        """通过 pymsg RPC（execute 路由）批量发送到 dbv2 Logs 表。
-        now = datetime.now(timezone.utc).isoformat()
+
-        client = RpcClient(_DBV2_ADDR, _DBV2_PORT, self._service_name + "-log")
+        整个 batch 复用一个 TCP 连接；网络错误时放弃本批剩余条目（fail-safe），
+        不抛异常给调用方。
+        """
+        try:
+            client = RpcClient(_DBV2_ADDR, _DBV2_PORT, self._service_name + "-log")
+        except Exception as e:
+            print(f"[log_client] 创建 RpcClient 失败: {e}", file=_sys.stderr)
+            return
+
         for entry in batch:
             extra_val = entry.get("extra")
+            try:
+                extra_json = _safe_json(extra_val) if extra_val is not None else None
+            except Exception:
+                extra_json = None
             payload = {
                 "action": "insert",
                 "table": "Logs",
                 "values": {
-                    "ts": now,
+                    "ts": entry.get("ts", _now_iso()),
                     "service": entry.get("service"),
                     "level": entry.get("level"),
                     "message": entry.get("message"),
                     "trace_id": entry.get("trace_id"),
-                    "extra": json.dumps(extra_val, ensure_ascii=False)
+                    "extra": extra_json,
-                    if extra_val is not None else None,
                 },
             }
             try:

smtp/log_client.py

@@ -1,10 +1,11 @@
-"""统一日志客户端 — 各服务共用（dnfs / smtp / static 三份副本内容一致，改动请同步）。
+"""统一日志客户端 — 各服务共用（dnfs / smtp / static / confs 四份副本内容一致，改动请同步）。
 
 设计要点：
   - 通过 pymsg RPC 调用 dbv2 execute 路由，逐条 INSERT 到 Logs 表
   - 异步非阻塞：内存队列 + 后台线程批量推送，不拖慢业务
-  - fail-safe：db 服务不可用时静默丢弃，绝不影响业务
+  - fail-safe：db 服务不可用时静默丢弃，绝不影响业务；后台线程任何异常都不会崩溃
   - 缓冲批量：队列满 50 条或超时 2 秒时自动 flush
+  - 时间戳：日志入队时即记录 ts，反映事件真实发生时间
 
 用法：
     from log_client import logs
@@ -35,6 +36,20 @@ _DBV2_PORT = int(_os.environ.get('DB_PORT', '8005'))
 _BATCH_SIZE = 50         # 队列满 50 条时 flush
 _FLUSH_INTERVAL = 2.0    # 超时 2 秒强制 flush
 _TIMEOUT = 5             # socket 超时（秒）
+_QUEUE_MAXSIZE = 10000   # 队列上限，防止后台线程异常时内存无限增长；满则丢弃最旧条目
+
+
+def _safe_json(obj):
+    """安全序列化 extra 字段，不可序列化对象转为 repr 字符串，保证不抛异常。"""
+    try:
+        return json.dumps(obj, ensure_ascii=False, default=str)
+    except Exception:
+        return repr(obj)
+
+
+def _now_iso():
+    """生成当前 UTC 时间 ISO 字符串（在入队时调用，保证时间戳准确）。"""
+    return datetime.now(timezone.utc).isoformat()
 
 
 class LogClient:
@@ -46,9 +61,10 @@ class LogClient:
         self._service_name = service_name
         self._batch_size = batch_size
         self._flush_interval = flush_interval
-        self._queue = queue.Queue()
+        self._queue = queue.Queue(maxsize=_QUEUE_MAXSIZE)
         self._stop = threading.Event()
         self._flush_signal = threading.Event()
+        self._flush_done = threading.Event()
         self._thread = threading.Thread(target=self._flush_loop,
                                         daemon=True, name='log-client')
         self._thread.start()
@@ -57,14 +73,30 @@ class LogClient:
     # ── 公开接口 ────────────────────────────────────────────────────────
 
     def push(self, service, level, message, trace_id=None, extra=None):
-        """推送一条日志（非阻塞，立即返回）。"""
+        """推送一条日志（非阻塞，立即返回）。
-        self._queue.put({
+
+        队列满时静默丢弃最旧条目再放入新条目，保证不阻塞业务线程、不导致 OOM。
+        """
+        entry = {
+            'ts': _now_iso(),  # 入队时即生成时间戳，反映真实事件时间
             'service': service,
             'level': level,
             'message': message,
             'trace_id': trace_id,
             'extra': extra,
-        })
+        }
+        try:
+            self._queue.put_nowait(entry)
+        except queue.Full:
+            # 队列满：丢弃最旧的一条，再放入新条目（drop-oldest 策略）
+            try:
+                self._queue.get_nowait()
+            except queue.Empty:
+                pass
+            try:
+                self._queue.put_nowait(entry)
+            except queue.Full:
+                pass  # 极端竞争下仍失败则静默丢弃
 
     def debug(self, service, message, **kw):
         self.push(service, 'DEBUG', message, **kw)
@@ -83,8 +115,15 @@ class LogClient:
 
     def flush(self):
         """手动 flush 队列中剩余的日志（含后台线程持有的局部 batch）。"""
+        self._flush_done.clear()
         self._flush_signal.set()  # 通知后台线程立即 flush 其局部 batch
-        time.sleep(0.6)  # 等待后台线程响应
+        # 等待后台线程完成 flush（最多等待 flush_interval + 2 秒余量）
+        self._flush_done.wait(timeout=self._flush_interval + 2.0)
+        # 主线程排空队列中剩余条目
+        self._drain_queue_to_db()
+
+    def _drain_queue_to_db(self):
+        """排空主线程可见的队列中剩余条目，直接发送到 db。"""
         batch = []
         while True:
             try:
@@ -96,60 +135,92 @@ class LogClient:
                 break
         if batch:
             self._send(batch)
-        self._flush_signal.clear()
 
     # ── 内部实现 ────────────────────────────────────────────────────────
 
     def _flush_loop(self):
-        """后台线程：积攒到 batch_size 或超时 flush_interval 时发送。"""
+        """后台线程：积攒到 batch_size 或超时 flush_interval 时发送。
+
+        外层捕获所有异常，保证后台线程在任何情况下（网络故障、序列化错误等）
+        都不会死亡，避免日志队列无限增长导致 OOM。
+        """
         batch = []
         last_flush = time.monotonic()
 
         while not self._stop.is_set():
-            timeout = self._flush_interval - (time.monotonic() - last_flush)
-            chk_timeout = max(0.1, min(timeout, 0.5))
             try:
-                item = self._queue.get(timeout=chk_timeout)
+                timeout = self._flush_interval - (time.monotonic() - last_flush)
-                batch.append(item)
+                chk_timeout = max(0.1, min(timeout, 0.5))
-                while len(batch) < self._batch_size:
+                try:
-                    try:
+                    item = self._queue.get(timeout=chk_timeout)
-                        batch.append(self._queue.get_nowait())
+                    batch.append(item)
-                    except queue.Empty:
+                    while len(batch) < self._batch_size:
-                        break
+                        try:
-            except queue.Empty:
+                            batch.append(self._queue.get_nowait())
-                pass
+                        except queue.Empty:
+                            break
+                except queue.Empty:
+                    pass
 
-            should_flush = (
+                should_flush = (
-                len(batch) >= self._batch_size or
+                    len(batch) >= self._batch_size or
-                self._flush_signal.is_set() or
+                    self._flush_signal.is_set() or
-                (batch and time.monotonic() - last_flush >= self._flush_interval)
+                    (batch and time.monotonic() - last_flush >= self._flush_interval)
-            )
+                )
-            if should_flush and batch:
+                if should_flush and batch:
-                self._send(batch)
+                    try:
-                batch = []
+                        self._send(batch)
-                last_flush = time.monotonic()
+                    except Exception as e:
+                        # _send 内部已有细粒度异常处理，此处是最终安全网
+                        print(f"[log_client] _send 异常（已捕获，线程继续运行）: {e}",
+                              file=_sys.stderr)
+                    batch = []
+                    last_flush = time.monotonic()
+                    if self._flush_signal.is_set():
+                        self._flush_signal.clear()
+                        self._flush_done.set()
+            except Exception as e:
+                # 最外层安全网：任何意外异常都不应杀死日志线程
+                print(f"[log_client] _flush_loop 异常（已捕获，线程继续运行）: {e}",
+                      file=_sys.stderr)
+                time.sleep(0.5)  # 防止异常风暴导致 CPU 满载
 
         # 退出前 flush 剩余
         if batch:
-            self._send(batch)
+            try:
+                self._send(batch)
+            except Exception as e:
+                print(f"[log_client] 退出时 flush 异常: {e}", file=_sys.stderr)
+        self._flush_done.set()
 
     def _send(self, batch):
-        """通过 pymsg RPC（execute 路由）逐条 INSERT 到 dbv2 Logs 表。"""
+        """通过 pymsg RPC（execute 路由）批量发送到 dbv2 Logs 表。
-        now = datetime.now(timezone.utc).isoformat()
+
-        client = RpcClient(_DBV2_ADDR, _DBV2_PORT, self._service_name + "-log")
+        整个 batch 复用一个 TCP 连接；网络错误时放弃本批剩余条目（fail-safe），
+        不抛异常给调用方。
+        """
+        try:
+            client = RpcClient(_DBV2_ADDR, _DBV2_PORT, self._service_name + "-log")
+        except Exception as e:
+            print(f"[log_client] 创建 RpcClient 失败: {e}", file=_sys.stderr)
+            return
+
         for entry in batch:
             extra_val = entry.get("extra")
+            try:
+                extra_json = _safe_json(extra_val) if extra_val is not None else None
+            except Exception:
+                extra_json = None
             payload = {
                 "action": "insert",
                 "table": "Logs",
                 "values": {
-                    "ts": now,
+                    "ts": entry.get("ts", _now_iso()),
                     "service": entry.get("service"),
                     "level": entry.get("level"),
                     "message": entry.get("message"),
                     "trace_id": entry.get("trace_id"),
-                    "extra": json.dumps(extra_val, ensure_ascii=False)
+                    "extra": extra_json,
-                    if extra_val is not None else None,
                 },
             }
             try:

static/log_client.py

@@ -1,10 +1,11 @@
-"""统一日志客户端 — 各服务共用（dnfs / smtp / static 三份副本内容一致，改动请同步）。
+"""统一日志客户端 — 各服务共用（dnfs / smtp / static / confs 四份副本内容一致，改动请同步）。
 
 设计要点：
   - 通过 pymsg RPC 调用 dbv2 execute 路由，逐条 INSERT 到 Logs 表
   - 异步非阻塞：内存队列 + 后台线程批量推送，不拖慢业务
-  - fail-safe：db 服务不可用时静默丢弃，绝不影响业务
+  - fail-safe：db 服务不可用时静默丢弃，绝不影响业务；后台线程任何异常都不会崩溃
   - 缓冲批量：队列满 50 条或超时 2 秒时自动 flush
+  - 时间戳：日志入队时即记录 ts，反映事件真实发生时间
 
 用法：
     from log_client import logs
@@ -35,6 +36,20 @@ _DBV2_PORT = int(_os.environ.get('DB_PORT', '8005'))
 _BATCH_SIZE = 50         # 队列满 50 条时 flush
 _FLUSH_INTERVAL = 2.0    # 超时 2 秒强制 flush
 _TIMEOUT = 5             # socket 超时（秒）
+_QUEUE_MAXSIZE = 10000   # 队列上限，防止后台线程异常时内存无限增长；满则丢弃最旧条目
+
+
+def _safe_json(obj):
+    """安全序列化 extra 字段，不可序列化对象转为 repr 字符串，保证不抛异常。"""
+    try:
+        return json.dumps(obj, ensure_ascii=False, default=str)
+    except Exception:
+        return repr(obj)
+
+
+def _now_iso():
+    """生成当前 UTC 时间 ISO 字符串（在入队时调用，保证时间戳准确）。"""
+    return datetime.now(timezone.utc).isoformat()
 
 
 class LogClient:
@@ -46,9 +61,10 @@ class LogClient:
         self._service_name = service_name
         self._batch_size = batch_size
         self._flush_interval = flush_interval
-        self._queue = queue.Queue()
+        self._queue = queue.Queue(maxsize=_QUEUE_MAXSIZE)
         self._stop = threading.Event()
         self._flush_signal = threading.Event()
+        self._flush_done = threading.Event()
         self._thread = threading.Thread(target=self._flush_loop,
                                         daemon=True, name='log-client')
         self._thread.start()
@@ -57,14 +73,30 @@ class LogClient:
     # ── 公开接口 ────────────────────────────────────────────────────────
 
     def push(self, service, level, message, trace_id=None, extra=None):
-        """推送一条日志（非阻塞，立即返回）。"""
+        """推送一条日志（非阻塞，立即返回）。
-        self._queue.put({
+
+        队列满时静默丢弃最旧条目再放入新条目，保证不阻塞业务线程、不导致 OOM。
+        """
+        entry = {
+            'ts': _now_iso(),  # 入队时即生成时间戳，反映真实事件时间
             'service': service,
             'level': level,
             'message': message,
             'trace_id': trace_id,
             'extra': extra,
-        })
+        }
+        try:
+            self._queue.put_nowait(entry)
+        except queue.Full:
+            # 队列满：丢弃最旧的一条，再放入新条目（drop-oldest 策略）
+            try:
+                self._queue.get_nowait()
+            except queue.Empty:
+                pass
+            try:
+                self._queue.put_nowait(entry)
+            except queue.Full:
+                pass  # 极端竞争下仍失败则静默丢弃
 
     def debug(self, service, message, **kw):
         self.push(service, 'DEBUG', message, **kw)
@@ -83,8 +115,15 @@ class LogClient:
 
     def flush(self):
         """手动 flush 队列中剩余的日志（含后台线程持有的局部 batch）。"""
+        self._flush_done.clear()
         self._flush_signal.set()  # 通知后台线程立即 flush 其局部 batch
-        time.sleep(0.6)  # 等待后台线程响应
+        # 等待后台线程完成 flush（最多等待 flush_interval + 2 秒余量）
+        self._flush_done.wait(timeout=self._flush_interval + 2.0)
+        # 主线程排空队列中剩余条目
+        self._drain_queue_to_db()
+
+    def _drain_queue_to_db(self):
+        """排空主线程可见的队列中剩余条目，直接发送到 db。"""
         batch = []
         while True:
             try:
@@ -96,60 +135,92 @@ class LogClient:
                 break
         if batch:
             self._send(batch)
-        self._flush_signal.clear()
 
     # ── 内部实现 ────────────────────────────────────────────────────────
 
     def _flush_loop(self):
-        """后台线程：积攒到 batch_size 或超时 flush_interval 时发送。"""
+        """后台线程：积攒到 batch_size 或超时 flush_interval 时发送。
+
+        外层捕获所有异常，保证后台线程在任何情况下（网络故障、序列化错误等）
+        都不会死亡，避免日志队列无限增长导致 OOM。
+        """
         batch = []
         last_flush = time.monotonic()
 
         while not self._stop.is_set():
-            timeout = self._flush_interval - (time.monotonic() - last_flush)
-            chk_timeout = max(0.1, min(timeout, 0.5))
             try:
-                item = self._queue.get(timeout=chk_timeout)
+                timeout = self._flush_interval - (time.monotonic() - last_flush)
-                batch.append(item)
+                chk_timeout = max(0.1, min(timeout, 0.5))
-                while len(batch) < self._batch_size:
+                try:
-                    try:
+                    item = self._queue.get(timeout=chk_timeout)
-                        batch.append(self._queue.get_nowait())
+                    batch.append(item)
-                    except queue.Empty:
+                    while len(batch) < self._batch_size:
-                        break
+                        try:
-            except queue.Empty:
+                            batch.append(self._queue.get_nowait())
-                pass
+                        except queue.Empty:
+                            break
+                except queue.Empty:
+                    pass
 
-            should_flush = (
+                should_flush = (
-                len(batch) >= self._batch_size or
+                    len(batch) >= self._batch_size or
-                self._flush_signal.is_set() or
+                    self._flush_signal.is_set() or
-                (batch and time.monotonic() - last_flush >= self._flush_interval)
+                    (batch and time.monotonic() - last_flush >= self._flush_interval)
-            )
+                )
-            if should_flush and batch:
+                if should_flush and batch:
-                self._send(batch)
+                    try:
-                batch = []
+                        self._send(batch)
-                last_flush = time.monotonic()
+                    except Exception as e:
+                        # _send 内部已有细粒度异常处理，此处是最终安全网
+                        print(f"[log_client] _send 异常（已捕获，线程继续运行）: {e}",
+                              file=_sys.stderr)
+                    batch = []
+                    last_flush = time.monotonic()
+                    if self._flush_signal.is_set():
+                        self._flush_signal.clear()
+                        self._flush_done.set()
+            except Exception as e:
+                # 最外层安全网：任何意外异常都不应杀死日志线程
+                print(f"[log_client] _flush_loop 异常（已捕获，线程继续运行）: {e}",
+                      file=_sys.stderr)
+                time.sleep(0.5)  # 防止异常风暴导致 CPU 满载
 
         # 退出前 flush 剩余
         if batch:
-            self._send(batch)
+            try:
+                self._send(batch)
+            except Exception as e:
+                print(f"[log_client] 退出时 flush 异常: {e}", file=_sys.stderr)
+        self._flush_done.set()
 
     def _send(self, batch):
-        """通过 pymsg RPC（execute 路由）逐条 INSERT 到 dbv2 Logs 表。"""
+        """通过 pymsg RPC（execute 路由）批量发送到 dbv2 Logs 表。
-        now = datetime.now(timezone.utc).isoformat()
+
-        client = RpcClient(_DBV2_ADDR, _DBV2_PORT, self._service_name + "-log")
+        整个 batch 复用一个 TCP 连接；网络错误时放弃本批剩余条目（fail-safe），
+        不抛异常给调用方。
+        """
+        try:
+            client = RpcClient(_DBV2_ADDR, _DBV2_PORT, self._service_name + "-log")
+        except Exception as e:
+            print(f"[log_client] 创建 RpcClient 失败: {e}", file=_sys.stderr)
+            return
+
         for entry in batch:
             extra_val = entry.get("extra")
+            try:
+                extra_json = _safe_json(extra_val) if extra_val is not None else None
+            except Exception:
+                extra_json = None
             payload = {
                 "action": "insert",
                 "table": "Logs",
                 "values": {
-                    "ts": now,
+                    "ts": entry.get("ts", _now_iso()),
                     "service": entry.get("service"),
                     "level": entry.get("level"),
                     "message": entry.get("message"),
                     "trace_id": entry.get("trace_id"),
-                    "extra": json.dumps(extra_val, ensure_ascii=False)
+                    "extra": extra_json,
-                    if extra_val is not None else None,
                 },
             }
             try: