diff --git a/runtime/gateway.py b/runtime/gateway.py index 9955b0ce..5872e29e 100644 --- a/runtime/gateway.py +++ b/runtime/gateway.py @@ -36,6 +36,11 @@ from oclaw.runtime.worker import ensure_worker_started from oclaw.runtime.orchestration.trace import new_span_id, new_trace_id from oclaw.runtime.chat.tool_runtime import compact_turn_tool_messages_for_storage from oclaw.runtime.chat.model_path_audit import ensure_no_tool_or_embedded_image_payload +from oclaw.runtime.session_auto_title import ( + AUTO_TITLE_SYSTEM_PROMPT_EN, + AUTO_TITLE_SYSTEM_PROMPT_ZH, + finalize_auto_title, +) from oclaw.runtime.tools.base import ToolRegistry from oclaw.runtime.tools.local_sdk import local_adapter_startup_self_check @@ -180,7 +185,7 @@ class OclawGateway: } def _maybe_generate_title_on_third_round(self, *, msg: StandardMessage, model: Any | None) -> None: - """Generate title once on round-3 using user text only (no tools/reasoning context).""" + """Generate title once on round-3: one plain model.chat (system+user, no tools).""" if model is None or not callable(getattr(model, "chat", None)): return sid = str(msg.session_id or "").strip() @@ -225,19 +230,12 @@ class OclawGateway: body = body[:_TITLE_BODIES_MAX_CHARS] try: lang_is_en = str(msg.metadata.get("lang") if isinstance(msg.metadata, dict) else "").lower().startswith("en") - sys = ( - "Generate a concise chat title from these user messages only. " - "Use the dominant language used by the user content body. " - "Return title text only, no quotes, no markdown, max 18 chars." - if lang_is_en - else "仅基于以下用户正文生成简短会话标题。请使用对话内容主体语言命名。" - "只返回标题文本，不要引号，不要markdown，最多18个字。" - ) + sys = AUTO_TITLE_SYSTEM_PROMPT_EN if lang_is_en else AUTO_TITLE_SYSTEM_PROMPT_ZH messages = [{"role": "system", "content": sys}, {"role": "user", "content": body}] ensure_no_tool_or_embedded_image_payload(messages=messages, path="gateway.auto_title") resp = model.chat(messages, [], on_token=None) - title = str(getattr(resp, "content", "") or "").strip().replace("\n", " ") - title = title.strip("\"'` ").strip() + raw_title = str(getattr(resp, "content", "") or "").strip().strip("\"'` ") + title = finalize_auto_title(raw=raw_title, fallback=str(bodies[0] or "").strip()) if not title: return self.store.rename_session(sid, title[:_SESSION_TITLE_MAX_LEN]) diff --git a/runtime/session_auto_title.py b/runtime/session_auto_title.py new file mode 100644 index 00000000..5bd8ad21 --- /dev/null +++ b/runtime/session_auto_title.py @@ -0,0 +1,61 @@ +from __future__ import annotations + +AUTO_TITLE_CHAR_MAX = 18 + +# Minimal system prompts: naming-only role (full rules are enforced in finalize_auto_title). +AUTO_TITLE_SYSTEM_PROMPT_EN = ( + "Dedicated session naming agent: output one short title only, " + f"≤{AUTO_TITLE_CHAR_MAX} characters, same language as the user lines." +) +AUTO_TITLE_SYSTEM_PROMPT_ZH = ( + "你是专职会话命名助手：只输出一条短标题，" + f"至多{AUTO_TITLE_CHAR_MAX}个字，语种与用户发言一致。" +) +# One-line titles longer than this are almost never valid (model wrote prose). +_AUTO_TITLE_RAW_SOFT_MAX = 42 + + +def _collapse_ws(raw: str) -> str: + return " ".join(str(raw or "").strip().split()) + + +def should_reject_auto_title(raw: str) -> bool: + s = _collapse_ws(raw) + if not s: + return True + if len(s) > _AUTO_TITLE_RAW_SOFT_MAX: + return True + low = s.lower() + if "claude code" in low: + return True + if "软件工程助手" in s: + return True + if "您好" in s and "我是" in s and len(s) > 16: + return True + if low.startswith("hello! i am") or low.startswith("hi! i am"): + return True + return False + + +def finalize_auto_title(*, raw: str, fallback: str, max_chars: int = AUTO_TITLE_CHAR_MAX) -> str: + """Turn model output into a short session title; reject prose and use fallback.""" + fb = _collapse_ws(fallback) + pick = _collapse_ws(raw) + if not pick or should_reject_auto_title(raw): + base = fb + else: + base = pick + if not base: + return "会话" + if len(base) <= max_chars: + return base + return base[:max_chars] + + +__all__ = [ + "AUTO_TITLE_CHAR_MAX", + "AUTO_TITLE_SYSTEM_PROMPT_EN", + "AUTO_TITLE_SYSTEM_PROMPT_ZH", + "finalize_auto_title", + "should_reject_auto_title", +] diff --git a/runtime/worker.py b/runtime/worker.py index 32709a72..6b163b65 100644 --- a/runtime/worker.py +++ b/runtime/worker.py @@ -12,6 +12,11 @@ from oclaw.runtime.memory_stage import build_memory_context from oclaw.runtime.relay_pointer import build_acp_relay_result, validate_relay_share_envelope from oclaw.runtime.types import StandardMessage from oclaw.runtime.chat.model_path_audit import ensure_no_tool_or_embedded_image_payload +from oclaw.runtime.session_auto_title import ( + AUTO_TITLE_SYSTEM_PROMPT_EN, + AUTO_TITLE_SYSTEM_PROMPT_ZH, + finalize_auto_title, +) _LOCK = threading.Lock() _THREAD: threading.Thread | None = None @@ -68,7 +73,7 @@ def _maybe_rename_from_first_user_message(*, store: Any, session_id: str, user_t def _maybe_generate_title_on_third_round(*, store: Any, msg: StandardMessage, model: Any | None) -> None: - """Generate title once on round-3 using user text only (no tools/reasoning context).""" + """Generate title once on round-3: one plain model.chat (system+user, no tools).""" if model is None or not callable(getattr(model, "chat", None)): return sid = str(msg.session_id or "").strip() @@ -113,19 +118,12 @@ def _maybe_generate_title_on_third_round(*, store: Any, msg: StandardMessage, mo body = body[:_TITLE_BODIES_MAX_CHARS] try: lang_is_en = str(msg.metadata.get("lang") if isinstance(msg.metadata, dict) else "").lower().startswith("en") - sys = ( - "Generate a concise chat title from these user messages only. " - "Use the dominant language used by the user content body. " - "Return title text only, no quotes, no markdown, max 18 chars." - if lang_is_en - else "仅基于以下用户正文生成简短会话标题。请使用对话内容主体语言命名。" - "只返回标题文本，不要引号，不要markdown，最多18个字。" - ) + sys = AUTO_TITLE_SYSTEM_PROMPT_EN if lang_is_en else AUTO_TITLE_SYSTEM_PROMPT_ZH messages = [{"role": "system", "content": sys}, {"role": "user", "content": body}] ensure_no_tool_or_embedded_image_payload(messages=messages, path="worker.auto_title") resp = model.chat(messages, [], on_token=None) - title = str(getattr(resp, "content", "") or "").strip().replace("\n", " ") - title = title.strip("\"'` ").strip() + raw_title = str(getattr(resp, "content", "") or "").strip().strip("\"'` ") + title = finalize_auto_title(raw=raw_title, fallback=str(bodies[0] or "").strip()) if not title: return store.rename_session(sid, title[:_SESSION_TITLE_MAX_LEN]) @@ -284,11 +282,7 @@ def _worker_loop(*, store: Any, worker_id: str, poll_interval_s: float) -> None: user_text=user_text, attachments=list(attachments or []), ) - _maybe_generate_title_on_third_round( - store=store, - msg=msg, - model=getattr(executor, "model", None), - ) + _maybe_generate_title_on_third_round(store=store, msg=msg, model=getattr(executor, "model", None)) run_out = run_agent_core( store=store, data=AgentCoreRunInput( diff --git a/tests/test_oclaw_worker_title.py b/tests/test_oclaw_worker_title.py index 9f74ba1a..9dbbff50 100644 --- a/tests/test_oclaw_worker_title.py +++ b/tests/test_oclaw_worker_title.py @@ -14,8 +14,11 @@ class _DummyResp: class _DummyModel: + def __init__(self, content: str = "第三轮标题") -> None: + self._content = content + def chat(self, _messages, _tools, on_token=None): # noqa: ANN001 - return _DummyResp("第三轮标题") + return _DummyResp(self._content) def test_worker_third_round_title_generation_updates_stage3(tmp_path: Path) -> None: @@ -57,3 +60,45 @@ def test_worker_third_round_title_generation_updates_stage3(tmp_path: Path) -> N assert renamed is not None assert str(getattr(renamed, "title", "") or "") == "第三轮标题" assert str(store.get_setting(f"AIA_SESSION_AUTO_TITLE_STAGE:{sid}") or "") == "3" + + +def test_worker_third_round_rejects_prose_title_uses_fallback(tmp_path: Path) -> None: + store = SqliteStore(str(tmp_path / "worker-title-fb.sqlite")) + tenant = store.create_tenant("Team") + user = store.create_user_account( + tenant_id=str(tenant["id"]), + username="tester", + display_name="Tester", + role="owner", + password_hash=hashlib.sha256("test-pass".encode("utf-8")).hexdigest(), + is_active=True, + ) + session = store.create_session_for_user( + title="第一轮标题", + tenant_id=str(tenant["id"]), + user_id=str(user["id"]), + ) + sid = str(session.id) + store.set_setting(f"AIA_SESSION_AUTO_TITLE_STAGE:{sid}", "1") + store.add_message(session_id=sid, role="user", content="第一轮问题") + store.add_message(session_id=sid, role="assistant", content="第一轮回答") + store.add_message(session_id=sid, role="user", content="第二轮问题") + store.add_message(session_id=sid, role="assistant", content="第二轮回答") + + msg = StandardMessage( + session_id=sid, + tenant_id=str(tenant["id"]), + user_id=str(user["id"]), + role="member", + channel="admin_chat", + text="第三轮问题", + attachments=[], + metadata={"lang": "zh"}, + ) + garbage = "您好！我是 Claude Code，一个软件工程助手。" + "x" * 80 + _maybe_generate_title_on_third_round(store=store, msg=msg, model=_DummyModel(content=garbage)) + + renamed = store.get_session(sid) + assert renamed is not None + assert str(getattr(renamed, "title", "") or "") == "第一轮问题" + assert str(store.get_setting(f"AIA_SESSION_AUTO_TITLE_STAGE:{sid}") or "") == "3" diff --git a/tests/test_session_auto_title.py b/tests/test_session_auto_title.py new file mode 100644 index 00000000..29f722ad --- /dev/null +++ b/tests/test_session_auto_title.py @@ -0,0 +1,27 @@ +from __future__ import annotations + +from oclaw.runtime.session_auto_title import AUTO_TITLE_CHAR_MAX, finalize_auto_title, should_reject_auto_title + + +def test_should_reject_long_prose() -> None: + blob = "您好！我是 Claude Code，一个软件工程助手。" + "x" * 80 + assert should_reject_auto_title(blob) is True + + +def test_should_reject_claude_code_intro() -> None: + assert should_reject_auto_title("Hello! I am Claude Code, a software engineering assistant.") is True + + +def test_should_accept_short_title() -> None: + assert should_reject_auto_title("告警查询配置") is False + + +def test_finalize_truncates_fallback() -> None: + long_fb = "一二三四五六七八九十11121314151617181920" + out = finalize_auto_title(raw="x" * 100, fallback=long_fb) + assert len(out) == AUTO_TITLE_CHAR_MAX + assert out == long_fb[:AUTO_TITLE_CHAR_MAX] + + +def test_finalize_uses_model_when_sane() -> None: + assert finalize_auto_title(raw=" 网管同步 ", fallback="fallback") == "网管同步"