diff --git a/core/task_title.py b/core/task_title.py index eecba5a..63a7780 100644 --- a/core/task_title.py +++ b/core/task_title.py @@ -101,6 +101,10 @@ def generate_task_title( f"{type(e).__name__}: {e}", flush=True, ) + # 标题是一次性闸:若模型瞬时失败后仍只消费 pending,任务会永久停在 + # “新对话”。直接用首条消息的第一行作本地兜底,保证有效首发总能完成 + # 命名;不保留 pending 给后续消息重试,避免拿第二条消息误命名整段对话。 + title = clean_generated_title("", user_message) # 一次性消费 pending。WHERE pending=true 是与人工 PATCH name 的竞态闸: # 用户先改名时 PATCH 已清 false,此处 rowcount=0,不覆盖。 diff --git a/tests/test_task_title.py b/tests/test_task_title.py index 0ed38b3..c338b34 100644 --- a/tests/test_task_title.py +++ b/tests/test_task_title.py @@ -2,8 +2,15 @@ from __future__ import annotations import unittest +from types import SimpleNamespace +from unittest.mock import MagicMock, patch +from uuid import uuid4 -from core.task_title import clean_generated_title, is_attachment_only_message +from core.task_title import ( + clean_generated_title, + generate_task_title, + is_attachment_only_message, +) class TaskTitleTests(unittest.TestCase): @@ -32,6 +39,46 @@ class TaskTitleTests(unittest.TestCase): )) self.assertFalse(is_attachment_only_message("")) + def test_model_failure_applies_local_fallback_title(self): + task_id = uuid4() + user_id = uuid4() + read_session = MagicMock() + read_session.execute.return_value.first.return_value = SimpleNamespace( + auto_title_pending=True, + auto_title_version=3, + ) + write_session = MagicMock() + write_session.execute.return_value.rowcount = 1 + + read_ctx = MagicMock() + read_ctx.__enter__.return_value = read_session + write_ctx = MagicMock() + write_ctx.__enter__.return_value = write_session + + with ( + patch("core.task_title.session_scope", side_effect=[read_ctx, write_ctx]), + patch( + "core.task_title.load_config", + return_value={ + "default_model": "deepseek_v4.flash", + "models_dir": "config/models", + }, + ), + patch("core.task_title.ModelCapabilities.load", return_value=MagicMock()), + patch("core.task_title.LLM", side_effect=RuntimeError("provider unavailable")), + ): + title = generate_task_title( + task_id=task_id, + user_id=user_id, + user_message="分析手机端自动命名失败\n补充说明", + model_profile="deepseek_v4.flash", + ) + + self.assertEqual(title, "分析手机端自动命名失败") + values = write_session.execute.call_args.args[0].compile().params + self.assertEqual(values["name"], "分析手机端自动命名失败") + self.assertFalse(values["auto_title_pending"]) + if __name__ == "__main__": unittest.main()