from __future__ import annotations import asyncio import json from datetime import datetime, timedelta, timezone from pathlib import Path import httpx from fastapi.testclient import TestClient from investibook import config as cfg from investibook.api import deps as deps_api from investibook.api import lark as lark_api from investibook.api import lark_feedback as lark_feedback_api from investibook.core.agent_runtime import AskResult from investibook.core.feedback_analysis import analyze_feedback_and_persist from investibook.core import lark_bot as lark_core from investibook.core.lark_bot import ( ResolvedLarkEmail, extract_issue_id, resolve_issue_id_from_event, ) from investibook.db import queries as dbq from investibook.main import app def _configure_test_runtime(monkeypatch, tmp_path: Path) -> list[dict]: data_dir = tmp_path / "data" monkeypatch.setattr(dbq, "DB_PATH", data_dir / "db" / "investibook.db") monkeypatch.setattr(cfg, "WORKSPACES_DIR", data_dir / "workspaces") monkeypatch.setattr(cfg, "TEMPLATES_DIR", data_dir / "templates") monkeypatch.setattr(cfg, "RUNTIME_CONFIG_DIR", data_dir / "config") monkeypatch.setattr(cfg, "LARK_APP_ID", "cli_test_app") monkeypatch.setattr(cfg, "LARK_APP_SECRET", "test_secret") monkeypatch.setattr(cfg, "LARK_VERIFICATION_TOKEN", "test_token") monkeypatch.setattr(cfg, "LARK_BOT_NAME", "investibook-bot-test") monkeypatch.setattr(cfg, "LARK_SESSION_TIMEOUT_MINUTES", 30) monkeypatch.setattr(cfg, "LARK_AGENT_IDLE_TIMEOUT_SECONDS", 600) monkeypatch.setattr(cfg, "INVESTIBOOK_WEB_BASE_URL", "http://testserver") from investibook import main as main_mod monkeypatch.setattr(main_mod, "_seed_templates", lambda: None) monkeypatch.setattr( main_mod, "_db_migrate_messages_v1_to_v2", lambda: {"migrated": 0} ) async def _fake_scheduler(_shutdown_event): return None monkeypatch.setattr( "investibook.core.scheduler.start_daily_evaluation_scheduler", _fake_scheduler, ) chat_calls: list[dict] = [] async def _fake_ask_and_wait(*args, **kwargs): chat_calls.append(dict(kwargs)) return "chat-session-test", AskResult( text="测试诊断回复", finish_reason="completed", error=None ) monkeypatch.setattr(lark_api, "run_chat_and_wait", _fake_ask_and_wait) async def _fake_reply_interactive_message(*, message_id: str, card: dict): return { "code": 0, "data": { "message_id": f"card-{message_id}", "body": {"content": '{"text":"feedback card"}'}, }, } monkeypatch.setattr( lark_api, "reply_interactive_message", _fake_reply_interactive_message ) return chat_calls def test_build_workspace_url_uses_configured_web_base_url(monkeypatch): monkeypatch.setattr(cfg, "INVESTIBOOK_WEB_BASE_URL", "https://dev.investibook.com") assert ( lark_api._build_workspace_url("ws-1775829959034-0805f7ae9") == "https://dev.investibook.com/workspace/ws-1775829959034-0805f7ae9" ) def test_build_agent_timeout_fallback_keeps_current_workspace(): reply = lark_api._build_agent_fallback_text(finish_reason="timeout") assert "后台仍在处理中" in reply assert "无需重新开始" in reply assert "可以直接输入新的 issue id" not in reply def test_build_agent_error_fallback_reports_failure(): reply = lark_api._build_agent_fallback_text(finish_reason="error") assert "未能正常完成" in reply assert "后台仍在处理中" not in reply def test_resolve_issue_id_from_topic_payload(): event = { "message": { "chat_type": "group", "content": '{"text":"@Bot 帮我看下根因"}', }, "topic": { "title": "O11Y-2343244 TiDB CPU 抖动告警", "card_fields": [{"label": "Issue", "value": "O11Y-2343244"}], }, } issue_id = asyncio.run(resolve_issue_id_from_event(event, "@Bot 帮我看下根因")) assert issue_id == "O11Y-2343244" def test_extract_issue_id_prefers_o11y(): text = "关联 TCOC-12345，同时告警单是 O11Y-2343244，需要优先诊断" assert extract_issue_id(text) == "O11Y-2343244" def test_lark_webhook_private_chat_creates_session(monkeypatch, tmp_path: Path): chat_calls = _configure_test_runtime(monkeypatch, tmp_path) replies: list[tuple[str, str]] = [] cards: list[tuple[str, dict]] = [] reactions: list[str] = [] async def _fake_add_message_reaction(*, message_id: str, emoji_type=None): reactions.append(message_id) return {"code": 0} async def _fake_reply_text_message(*, message_id: str, text: str): replies.append((message_id, text)) return {"code": 0} async def _fake_reply_interactive_message(*, message_id: str, card: dict): cards.append((message_id, card)) return { "code": 0, "data": { "message_id": "om_bot_private_1", }, } monkeypatch.setattr(lark_api, "add_message_reaction", _fake_add_message_reaction) monkeypatch.setattr(lark_api, "reply_text_message", _fake_reply_text_message) monkeypatch.setattr(lark_api, "reply_interactive_message", _fake_reply_interactive_message) with TestClient(app) as client: payload = { "header": { "event_id": "evt_private_1", "event_type": "im.message.receive_v1", "token": "test_token", }, "event": { "sender": {"sender_id": {"user_id": "u_private_1"}}, "message": { "message_id": "om_private_1", "chat_id": "oc_private_1", "chat_type": "p2p", "message_type": "text", "content": '{"text":"最近告警怎么样"}', }, }, } resp = client.post("/api/lark/events", json=payload) assert resp.status_code == 200 body = resp.json() assert body["ok"] is True assert body["issue_id"] is None assert body["reused"] is False assert reactions == ["om_private_1"] assert cards and cards[0][0] == "om_private_1" assert cards[0][1]["elements"][0]["elements"][0]["content"].startswith("Workspace:") assert cards[0][1]["elements"][1]["content"].startswith( "Workspace link: http://testserver/workspace/" ) assert "测试诊断回复" in cards[0][1]["elements"][1]["content"] assert chat_calls[0]["diagnosis_mode"] == "copilot" assert chat_calls[0]["trigger_type"] == "human_message" assert chat_calls[0]["supports_followup"] is True assert chat_calls[0]["idle_timeout_s"] == 600 assert cards[0][1]["elements"][3]["actions"][0]["text"]["content"] == "评价" row = dbq._db_get_latest_feishu_session("u_private_1", "oc_private_1") assert row is not None assert row["chat_type"] == "p2p" assert row["issue_id"] is None assert row["user_email"] == "" assert row["workspace_id"] outsider = dbq._db_create_user("lark-workspace-viewer@pingcap.com", "pass-123") access_row, is_owner, can_edit = asyncio.run( deps_api.require_workspace_access(row["workspace_id"], outsider) ) assert access_row["id"] == row["workspace_id"] assert is_owner is False assert can_edit is False def test_lark_webhook_falls_back_to_text_when_interactive_reply_fails(monkeypatch, tmp_path: Path): _configure_test_runtime(monkeypatch, tmp_path) replies: list[tuple[str, str]] = [] async def _fake_add_message_reaction(*, message_id: str, emoji_type=None): return {"code": 0} async def _fake_reply_text_message(*, message_id: str, text: str): replies.append((message_id, text)) return {"code": 0, "data": {"message_id": "om_bot_text_fallback_1"}} async def _fake_reply_interactive_message(*, message_id: str, card: dict): request = httpx.Request( "POST", f"https://open.feishu.cn/open-apis/im/v1/messages/{message_id}/reply", ) response = httpx.Response( 400, request=request, text='{"code":1234,"msg":"invalid interactive card"}', ) raise httpx.HTTPStatusError( "Client error '400 Bad Request' for url", request=request, response=response, ) monkeypatch.setattr(lark_api, "add_message_reaction", _fake_add_message_reaction) monkeypatch.setattr(lark_api, "reply_text_message", _fake_reply_text_message) monkeypatch.setattr(lark_api, "reply_interactive_message", _fake_reply_interactive_message) with TestClient(app) as client: payload = { "header": { "event_id": "evt_private_text_fallback", "event_type": "im.message.receive_v1", "token": "test_token", }, "event": { "sender": {"sender_id": {"user_id": "u_private_fallback"}}, "message": { "message_id": "om_private_fallback_1", "chat_id": "oc_private_fallback_1", "chat_type": "p2p", "message_type": "text", "content": '{"text":"最近告警怎么样"}', }, }, } resp = client.post("/api/lark/events", json=payload) assert resp.status_code == 200 assert resp.json()["reply_sent"] is True assert replies assert replies[0][0] == "om_private_fallback_1" assert replies[0][1].startswith("[Workspace:") def test_lark_feedback_action_and_comment_flow(monkeypatch, tmp_path: Path): _configure_test_runtime(monkeypatch, tmp_path) replies: list[tuple[str, str]] = [] cards: list[tuple[str, dict]] = [] async def _fake_add_message_reaction(*, message_id: str, emoji_type=None): return {"code": 0} async def _fake_reply_text_message(*, message_id: str, text: str): replies.append((message_id, text)) return {"code": 0} async def _fake_reply_interactive_message(*, message_id: str, card: dict): cards.append((message_id, card)) return { "code": 0, "data": { "message_id": "om_bot_feedback_1", }, } monkeypatch.setattr(lark_api, "add_message_reaction", _fake_add_message_reaction) monkeypatch.setattr(lark_api, "reply_text_message", _fake_reply_text_message) monkeypatch.setattr(lark_api, "reply_interactive_message", _fake_reply_interactive_message) with TestClient(app) as client: message_payload = { "header": { "event_id": "evt_private_feedback_1", "event_type": "im.message.receive_v1", "token": "test_token", }, "event": { "sender": { "sender_id": {"user_id": "u_feedback_1"}, "name": "测试用户", }, "message": { "message_id": "om_private_feedback_1", "chat_id": "oc_private_feedback_1", "chat_type": "p2p", "message_type": "text", "content": '{"text":"这个告警怎么看"}', }, }, } msg_resp = client.post("/api/lark/events", json=message_payload) assert msg_resp.status_code == 200 assert cards and cards[0][0] == "om_private_feedback_1" assert cards[0][1]["elements"][0]["elements"][0]["content"].startswith("Workspace:") assert cards[0][1]["elements"][1]["content"].startswith( "Workspace link: http://testserver/workspace/" ) assert cards[0][1]["elements"][3]["actions"][0]["text"]["content"] == "评价" expand_payload = { "token": "test_token", "uuid": "feedback-expand-1", "open_message_id": "om_bot_feedback_1", "operator": { "operator_id": {"user_id": "u_feedback_1"}, "name": "测试用户", }, "action": { "text": {"content": "评价"}, "value": { "action": "feedback_expand", }, }, } expand_resp = client.post("/api/lark/feedback/actions", json=expand_payload) assert expand_resp.status_code == 200 assert expand_resp.json()["card"]["type"] == "raw" assert expand_resp.json()["card"]["data"]["elements"][3]["actions"][0]["text"]["content"] == "正确" assert ( expand_resp.json()["card"]["data"]["elements"][3]["actions"][0]["value"]["assistant_message_id"] == "om_bot_feedback_1" ) assert ( expand_resp.json()["card"]["data"]["elements"][3]["actions"][1]["text"]["content"] == "不正确（自动收集上下文反馈）" ) feedback_payload = { "token": "test_token", "uuid": "feedback-action-1", "open_message_id": "om_bot_feedback_1", "operator": { "operator_id": {"user_id": "u_feedback_1"}, "name": "测试用户", }, "action": { "text": {"content": "不正确（自动收集上下文反馈）"}, "value": { "action": "feedback_mark", "workspace_id": "lark-ws-test", "session_id": "sess-test", "chat_id": "oc_private_feedback_1", "chat_type": "p2p", "thread_root_id": "om_private_feedback_1", "issue_id": "", "feedback_value": "incorrect", } }, } feedback_resp = client.post("/api/lark/feedback/actions", json=feedback_payload) assert feedback_resp.status_code == 200 assert feedback_resp.json()["card"]["type"] == "raw" assert "不正确" in feedback_resp.json()["card"]["data"]["elements"][-1]["elements"][0]["content"] feedback_row = dbq._db_get_feishu_message_feedback( "om_bot_feedback_1", "u_feedback_1" ) assert feedback_row is not None assert feedback_row["feedback_value"] == "incorrect" assert feedback_row["user_email"] == "" assert feedback_row["interaction_scene"] == "private_chat" assert feedback_row["awaiting_comment"] == 0 session_row = dbq._db_get_latest_feishu_session( "u_feedback_1", "oc_private_feedback_1" ) assert session_row is not None workspace_feedback = dbq._db_get_feedback( session_row["workspace_id"], "u_feedback_1", source="lark", ) assert workspace_feedback is not None assert workspace_feedback["rating"] == "failure" assert workspace_feedback["source"] == "lark" feedback_row = dbq._db_get_feishu_message_feedback( "om_bot_feedback_1", "u_feedback_1" ) assert feedback_row is not None assert feedback_row["comment"] is None assert feedback_row["awaiting_comment"] == 0 def test_lark_feedback_persists_canonical_feishu_user_id(monkeypatch, tmp_path: Path): _configure_test_runtime(monkeypatch, tmp_path) cards: list[tuple[str, dict]] = [] async def _fake_add_message_reaction(*, message_id: str, emoji_type=None): return {"code": 0} async def _fake_reply_text_message(*, message_id: str, text: str): return {"code": 0} async def _fake_reply_interactive_message(*, message_id: str, card: dict): cards.append((message_id, card)) return { "code": 0, "data": { "message_id": "om_bot_feedback_open_id_1", }, } async def _fake_resolve_lark_user_id(*, sender_id, collapsed_user_id: str): if collapsed_user_id == "ou_feedback_1": return "u_feedback_canonical_1" return collapsed_user_id monkeypatch.setattr(lark_api, "add_message_reaction", _fake_add_message_reaction) monkeypatch.setattr(lark_api, "reply_text_message", _fake_reply_text_message) monkeypatch.setattr(lark_api, "reply_interactive_message", _fake_reply_interactive_message) monkeypatch.setattr(lark_api, "resolve_lark_user_id", _fake_resolve_lark_user_id) monkeypatch.setattr( lark_feedback_api, "resolve_lark_user_id", _fake_resolve_lark_user_id, ) with TestClient(app) as client: message_payload = { "header": { "event_id": "evt_private_feedback_open_id_1", "event_type": "im.message.receive_v1", "token": "test_token", }, "event": { "sender": { "sender_id": {"open_id": "ou_feedback_1"}, "name": "测试用户", }, "message": { "message_id": "om_private_feedback_open_id_1", "chat_id": "oc_private_feedback_open_id_1", "chat_type": "p2p", "message_type": "text", "content": '{"text":"这个告警怎么看"}', }, }, } msg_resp = client.post("/api/lark/events", json=message_payload) assert msg_resp.status_code == 200 feedback_payload = { "token": "test_token", "uuid": "feedback-action-open-id-1", "open_message_id": "om_bot_feedback_open_id_1", "operator": { "operator_id": {"open_id": "ou_feedback_1"}, "name": "测试用户", }, "action": { "text": {"content": "不正确（自动收集上下文反馈）"}, "value": { "action": "feedback_mark", "workspace_id": "lark-ws-test", "session_id": "sess-test", "chat_id": "oc_private_feedback_open_id_1", "chat_type": "p2p", "thread_root_id": "om_private_feedback_open_id_1", "issue_id": "", "feedback_value": "incorrect", }, }, } feedback_resp = client.post("/api/lark/feedback/actions", json=feedback_payload) assert feedback_resp.status_code == 200 session_row = dbq._db_get_latest_feishu_session( "u_feedback_canonical_1", "oc_private_feedback_open_id_1" ) assert session_row is not None feedback_row = dbq._db_get_feishu_message_feedback( "om_bot_feedback_open_id_1", "u_feedback_canonical_1" ) assert feedback_row is not None workspace_feedback = dbq._db_get_feedback( session_row["workspace_id"], "u_feedback_canonical_1", source="lark", ) assert workspace_feedback is not None assert workspace_feedback["rating"] == "failure" assert dbq._db_get_feedback( session_row["workspace_id"], "ou_feedback_1", source="lark", ) is None def test_lark_feedback_metric_uses_workspace_feedback_scene(monkeypatch, tmp_path: Path): _configure_test_runtime(monkeypatch, tmp_path) dbq._db_init() metric_calls: list[dict[str, str | None]] = [] unified_metric_calls: list[dict[str, str | None]] = [] monkeypatch.setattr( lark_feedback_api, "record_lark_feedback_mark", lambda **kwargs: metric_calls.append(kwargs), ) monkeypatch.setattr( lark_feedback_api, "record_feedback_mark", lambda **kwargs: unified_metric_calls.append(kwargs), ) asyncio.run( lark_feedback_api._sync_workspace_feedback_from_lark( { "workspace_id": "lark-ws-scene-1", "feedback_user_id": "u_scene_1", "feedback_value": "correct", "interaction_scene": "thread_comment", "chat_type": "group", "thread_root_id": "om_thread_1", "user_email": "scene-user@example.com", "analysis_summary": "", "analysis_updated_at": "", } ) ) workspace_feedback = dbq._db_get_feedback( "lark-ws-scene-1", "u_scene_1", source="lark", ) assert workspace_feedback is not None assert workspace_feedback["rating"] == "success" assert metric_calls == [ { "feedback_value": "correct", "interaction_scene": "thread_comment", "user_email": "scene-user@example.com", } ] assert unified_metric_calls == [ { "source": "lark", "rating": "success", "interaction_scene": "thread_comment", "user_email": "scene-user@example.com", } ] def test_lark_feedback_metric_is_not_duplicated_by_analysis_sync(monkeypatch, tmp_path: Path): _configure_test_runtime(monkeypatch, tmp_path) dbq._db_init() metric_calls: list[dict[str, str | None]] = [] unified_metric_calls: list[dict[str, str | None]] = [] monkeypatch.setattr( lark_feedback_api, "record_lark_feedback_mark", lambda **kwargs: metric_calls.append(kwargs), ) monkeypatch.setattr( lark_feedback_api, "record_feedback_mark", lambda **kwargs: unified_metric_calls.append(kwargs), ) base_row = { "workspace_id": "lark-ws-analysis-1", "feedback_user_id": "u_analysis_metric_1", "feedback_value": "incorrect", "interaction_scene": "private_chat", "chat_type": "p2p", "thread_root_id": "om_private_analysis_metric_1", "user_email": "analysis-user@example.com", } asyncio.run( lark_feedback_api._sync_workspace_feedback_from_lark( { **base_row, "analysis_summary": "", "analysis_updated_at": "", } ) ) asyncio.run( lark_feedback_api._sync_workspace_feedback_from_lark( { **base_row, "analysis_summary": "缺少针对当前问题的具体诊断", "analysis_updated_at": "2026-04-16T00:00:00+00:00", } ) ) workspace_feedback = dbq._db_get_feedback( "lark-ws-analysis-1", "u_analysis_metric_1", source="lark", ) assert workspace_feedback is not None assert workspace_feedback["rating"] == "failure" assert workspace_feedback["failure_reason"] == "缺少针对当前问题的具体诊断" assert metric_calls == [ { "feedback_value": "incorrect", "interaction_scene": "private_chat", "user_email": "analysis-user@example.com", } ] assert unified_metric_calls == [ { "source": "lark", "rating": "failure", "interaction_scene": "private_chat", "user_email": "analysis-user@example.com", } ] def test_lark_chat_metrics_uses_real_feishu_email_and_preserves_it(monkeypatch, tmp_path: Path): _configure_test_runtime(monkeypatch, tmp_path) metrics_calls: list[dict[str, str | None]] = [] async def _fake_add_message_reaction(*, message_id: str, emoji_type=None): return {"code": 0} async def _fake_reply_text_message(*, message_id: str, text: str): return {"code": 0} async def _fake_reply_interactive_message(*, message_id: str, card: dict): return { "code": 0, "data": { "message_id": f"om_bot_{message_id}", }, } async def _fake_run_chat_and_wait(**kwargs): metrics_calls.append( { "user_email": kwargs.get("user_email"), "metrics_user_email": kwargs.get("metrics_user_email"), } ) return "chat-session-test", AskResult( text="测试诊断回复", finish_reason="completed", error=None ) call_count = {"value": 0} async def _fake_resolve_lark_user_email(value, *, sender_id, collapsed_user_id): call_count["value"] += 1 if call_count["value"] == 1: return ResolvedLarkEmail("hongmeng.lv@pingcap.com", "contact_api") return ResolvedLarkEmail("", "none") monkeypatch.setattr(lark_api, "add_message_reaction", _fake_add_message_reaction) monkeypatch.setattr(lark_api, "reply_text_message", _fake_reply_text_message) monkeypatch.setattr(lark_api, "reply_interactive_message", _fake_reply_interactive_message) monkeypatch.setattr(lark_api, "run_chat_and_wait", _fake_run_chat_and_wait) monkeypatch.setattr(lark_api, "resolve_lark_user_email", _fake_resolve_lark_user_email) with TestClient(app) as client: first_payload = { "header": { "event_id": "evt_metrics_email_1", "event_type": "im.message.receive_v1", "token": "test_token", }, "event": { "sender": {"sender_id": {"user_id": "u_metrics_email_1"}}, "message": { "message_id": "om_metrics_email_1", "chat_id": "oc_metrics_email_1", "chat_type": "p2p", "message_type": "text", "content": '{"text":"第一次提问"}', }, }, } first_resp = client.post("/api/lark/events", json=first_payload) assert first_resp.status_code == 200 second_payload = { "header": { "event_id": "evt_metrics_email_2", "event_type": "im.message.receive_v1", "token": "test_token", }, "event": { "sender": {"sender_id": {"user_id": "u_metrics_email_1"}}, "message": { "message_id": "om_metrics_email_2", "chat_id": "oc_metrics_email_1", "chat_type": "p2p", "message_type": "text", "content": '{"text":"第二次提问"}', }, }, } second_resp = client.post("/api/lark/events", json=second_payload) assert second_resp.status_code == 200 assert [call["metrics_user_email"] for call in metrics_calls] == [ "hongmeng.lv@pingcap.com", "hongmeng.lv@pingcap.com", ] assert [call["user_email"] for call in metrics_calls] == [ "hongmeng.lv@pingcap.com", "hongmeng.lv@pingcap.com", ] session_row = dbq._db_get_latest_feishu_session( "u_metrics_email_1", "oc_metrics_email_1" ) assert session_row is not None assert session_row["user_email"] == "hongmeng.lv@pingcap.com" def test_lark_feedback_expand_returns_toast_when_message_context_missing(monkeypatch, tmp_path: Path): _configure_test_runtime(monkeypatch, tmp_path) with TestClient(app) as client: expand_payload = { "token": "test_token", "uuid": "feedback-expand-missing-1", "open_message_id": "om_missing_feedback_1", "operator": { "operator_id": {"user_id": "u_feedback_missing_1"}, "name": "测试用户", }, "action": { "text": {"content": "评价"}, "value": { "action": "feedback_expand", }, }, } expand_resp = client.post("/api/lark/feedback/actions", json=expand_payload) assert expand_resp.status_code == 200 assert expand_resp.json()["toast"]["type"] == "warning" assert "暂时无法加载评价面板" in expand_resp.json()["toast"]["content"] def test_incorrect_feedback_analysis_is_persisted(monkeypatch, tmp_path: Path): _configure_test_runtime(monkeypatch, tmp_path) dbq._db_init() now = datetime.now(timezone.utc).isoformat() dbq._db_append_feishu_chat_message( message_id="om_parent_analysis_1", chat_id="oc_private_analysis_1", chat_type="p2p", sender_id="u_analysis_1", sender_name="测试用户", role="user", content="帮我判断 O11Y-123 的 TiKV CPU 告警是不是和昨天变更有关", created_at=now, thread_root_id="om_parent_analysis_1", ) dbq._db_append_feishu_chat_message( message_id="om_assistant_analysis_1", chat_id="oc_private_analysis_1", chat_type="p2p", sender_id="bot", sender_name="bot", role="assistant", content="[Workspace: lark-ws-test]\n已继续当前会话 f056bf2b。\n如果你想切到新的问题，可以直接输入新的 issue id，或者说“重新开始”。", created_at=now, thread_root_id="om_parent_analysis_1", parent_id="om_parent_analysis_1", ) dbq._db_upsert_feishu_message_feedback( assistant_message_id="om_assistant_analysis_1", parent_user_message_id="om_parent_analysis_1", workspace_id="lark-ws-test", session_id="sess-test", chat_id="oc_private_analysis_1", chat_type="p2p", interaction_scene="private_chat", thread_root_id="om_parent_analysis_1", issue_id="O11Y-123", feedback_user_id="u_analysis_1", user_email="", feedback_user_name="测试用户", feedback_value="incorrect", source="card_action", ) stored = asyncio.run( analyze_feedback_and_persist( assistant_message_id="om_assistant_analysis_1", feedback_user_id="u_analysis_1", ) ) assert stored is not None assert "缺少针对当前问题的具体诊断" in stored["analysis_summary"] details = json.loads(stored["analysis_details_json"]) issue_codes = {item["code"] for item in details["detected_issues"]} assert "template_reply" in issue_codes assert details["parent_user_excerpt"].startswith("帮我判断 O11Y-123") def test_incorrect_feedback_analysis_includes_following_user_context(monkeypatch, tmp_path: Path): _configure_test_runtime(monkeypatch, tmp_path) dbq._db_init() now = datetime.now(timezone.utc) dbq._db_append_feishu_chat_message( message_id="om_parent_analysis_2", chat_id="oc_private_analysis_2", chat_type="p2p", sender_id="u_analysis_2", sender_name="测试用户", role="user", content="帮我看下这次 CPU 告警", created_at=now.isoformat(), thread_root_id="om_parent_analysis_2", ) dbq._db_append_feishu_chat_message( message_id="om_assistant_analysis_2", chat_id="oc_private_analysis_2", chat_type="p2p", sender_id="bot", sender_name="bot", role="assistant", content="先继续当前会话，如果你想切到新的问题，可以直接输入新的 issue id。", created_at=(now + timedelta(seconds=1)).isoformat(), thread_root_id="om_parent_analysis_2", parent_id="om_parent_analysis_2", ) dbq._db_append_feishu_chat_message( message_id="om_followup_user_2", chat_id="oc_private_analysis_2", chat_type="p2p", sender_id="u_analysis_2", sender_name="测试用户", role="user", content="我看了监控，不是 CPU，本次是变更后连接数暴涨。", created_at=(now + timedelta(seconds=2)).isoformat(), thread_root_id="om_parent_analysis_2", ) dbq._db_upsert_feishu_message_feedback( assistant_message_id="om_assistant_analysis_2", parent_user_message_id="om_parent_analysis_2", workspace_id="lark-ws-test", session_id="sess-test", chat_id="oc_private_analysis_2", chat_type="p2p", interaction_scene="private_chat", thread_root_id="om_parent_analysis_2", issue_id=None, feedback_user_id="u_analysis_2", user_email="", feedback_user_name="测试用户", feedback_value="incorrect", source="card_action", ) stored = asyncio.run( analyze_feedback_and_persist( assistant_message_id="om_assistant_analysis_2", feedback_user_id="u_analysis_2", ) ) assert stored is not None assert "用户在标记不正确后补充" in stored["analysis_summary"] details = json.loads(stored["analysis_details_json"]) assert details["following_context"][0]["content"].startswith("我看了监控") def test_lark_webhook_group_requires_mention(monkeypatch, tmp_path: Path): _configure_test_runtime(monkeypatch, tmp_path) async def _fake_reply_text_message(*, message_id: str, text: str): raise AssertionError("reply_text_message should not be called") monkeypatch.setattr(lark_api, "reply_text_message", _fake_reply_text_message) with TestClient(app) as client: payload = { "header": { "event_id": "evt_group_1", "event_type": "im.message.receive_v1", "token": "test_token", }, "event": { "sender": {"sender_id": {"user_id": "u_group_1"}}, "message": { "message_id": "om_group_1", "chat_id": "oc_group_1", "chat_type": "group", "message_type": "text", "content": '{"text":"大家帮忙看下"}', }, }, } resp = client.post("/api/lark/events", json=payload) assert resp.status_code == 200 assert resp.json()["ignored"] == "without_mention" assert dbq._db_get_latest_feishu_session("u_group_1", "oc_group_1") is None cached = dbq._db_get_recent_feishu_chat_messages("oc_group_1", limit=5) assert len(cached) == 1 assert cached[0]["content"] == "大家帮忙看下" def test_lark_webhook_group_mentioning_other_user_does_not_trigger(monkeypatch, tmp_path: Path): _configure_test_runtime(monkeypatch, tmp_path) async def _fake_reply_text_message(*, message_id: str, text: str): raise AssertionError("reply_text_message should not be called") monkeypatch.setattr(lark_api, "reply_text_message", _fake_reply_text_message) with TestClient(app) as client: payload = { "header": { "event_id": "evt_group_other_mention", "event_type": "im.message.receive_v1", "token": "test_token", }, "event": { "sender": {"sender_id": {"user_id": "u_group_other_mention"}}, "message": { "message_id": "om_group_other_mention", "chat_id": "oc_group_other_mention", "chat_type": "group", "message_type": "text", "content": '{"zh_cn":{"content":[[{"tag":"at","user_id":"ou_someone_else","user_name":"倪哲鸣"},{"tag":"text","text":" 机器人上线了"}]]}}', }, }, } resp = client.post("/api/lark/events", json=payload) assert resp.status_code == 200 assert resp.json()["ignored"] == "without_mention" assert dbq._db_get_latest_feishu_session("u_group_other_mention", "oc_group_other_mention") is None def test_lark_webhook_group_text_placeholder_uses_mentions_metadata(monkeypatch, tmp_path: Path): _configure_test_runtime(monkeypatch, tmp_path) cards: list[tuple[str, dict]] = [] async def _fake_add_message_reaction(*, message_id: str, emoji_type=None): return {"code": 0} async def _fake_reply_text_message(*, message_id: str, text: str): return {"code": 0} async def _fake_reply_interactive_message(*, message_id: str, card: dict): cards.append((message_id, card)) return {"code": 0, "data": {"message_id": "om_group_reply_1"}} monkeypatch.setattr(lark_api, "add_message_reaction", _fake_add_message_reaction) monkeypatch.setattr(lark_api, "reply_text_message", _fake_reply_text_message) monkeypatch.setattr(lark_api, "reply_interactive_message", _fake_reply_interactive_message) with TestClient(app) as client: payload = { "header": { "event_id": "evt_group_mentions_metadata", "event_type": "im.message.receive_v1", "token": "test_token", }, "event": { "sender": {"sender_id": {"user_id": "u_group_mentions_metadata"}}, "message": { "message_id": "om_group_mentions_metadata", "chat_id": "oc_group_mentions_metadata", "chat_type": "group", "message_type": "text", "content": '{"text":"@_user_1 hello"}', "mentions": [ { "key": "@_user_1", "name": "investibook-bot-test", } ], }, }, } resp = client.post("/api/lark/events", json=payload) assert resp.status_code == 200 body = resp.json() assert body["ok"] is True assert body["reused"] is False assert cards and cards[0][0] == "om_group_mentions_metadata" assert cards[0][1]["elements"][3]["actions"][0]["text"]["content"] == "评价" def test_lark_webhook_group_summary_uses_cached_chat_history(monkeypatch, tmp_path: Path): _configure_test_runtime(monkeypatch, tmp_path) prompts: list[str] = [] async def _fake_add_message_reaction(*, message_id: str, emoji_type=None): return {"code": 0} async def _fake_reply_text_message(*, message_id: str, text: str): return { "code": 0, "data": { "message_id": f"reply-{message_id}", "body": {"content": '{"text":"群聊总结回复"}'}, }, } async def _fake_run_chat_and_wait(*, workspace_id=None, message=None, user_email=None, **kwargs): prompts.append(str(message)) return "chat-session-test", AskResult( text="群聊总结回复", finish_reason="completed", error=None ) monkeypatch.setattr(lark_api, "add_message_reaction", _fake_add_message_reaction) monkeypatch.setattr(lark_api, "reply_text_message", _fake_reply_text_message) monkeypatch.setattr(lark_api, "run_chat_and_wait", _fake_run_chat_and_wait) with TestClient(app) as client: first_payload = { "header": { "event_id": "evt_group_history_1", "event_type": "im.message.receive_v1", "token": "test_token", }, "event": { "sender": {"sender_id": {"user_id": "u_group_history_a"}, "name": "Alice"}, "message": { "message_id": "om_group_history_1", "chat_id": "oc_group_history", "chat_type": "group", "message_type": "text", "content": '{"text":"这个 issue 先看 TiKV CPU 异常"}', }, }, } second_payload = { "header": { "event_id": "evt_group_history_2", "event_type": "im.message.receive_v1", "token": "test_token", }, "event": { "sender": {"sender_id": {"user_id": "u_group_history_b"}, "name": "Bob"}, "message": { "message_id": "om_group_history_2", "chat_id": "oc_group_history", "chat_type": "group", "message_type": "text", "content": '{"text":"我怀疑和昨天的变更也有关系"}', }, }, } summary_payload = { "header": { "event_id": "evt_group_history_3", "event_type": "im.message.receive_v1", "token": "test_token", }, "event": { "sender": {"sender_id": {"user_id": "u_group_history_a"}, "name": "Alice"}, "message": { "message_id": "om_group_history_3", "chat_id": "oc_group_history", "chat_type": "group", "message_type": "text", "content": '{"zh_cn":{"content":[[{"tag":"at","user_id":"ou_bot","user_name":"investibook-bot-test"},{"tag":"text","text":" 帮我总结下上面聊了什么"}]]}}', }, }, } first_resp = client.post("/api/lark/events", json=first_payload) second_resp = client.post("/api/lark/events", json=second_payload) summary_resp = client.post("/api/lark/events", json=summary_payload) assert first_resp.status_code == 200 assert first_resp.json()["ignored"] == "without_mention" assert second_resp.status_code == 200 assert second_resp.json()["ignored"] == "without_mention" assert summary_resp.status_code == 200 assert prompts assert "Recent group messages:" in prompts[-1] assert "Alice: 这个 issue 先看 TiKV CPU 异常" in prompts[-1] assert "Bob: 我怀疑和昨天的变更也有关系" in prompts[-1] def test_lark_webhook_group_summary_fetches_older_history_from_lark(monkeypatch, tmp_path: Path): _configure_test_runtime(monkeypatch, tmp_path) prompts: list[str] = [] async def _fake_add_message_reaction(*, message_id: str, emoji_type=None): return {"code": 0} async def _fake_reply_text_message(*, message_id: str, text: str): return { "code": 0, "data": { "message_id": f"reply-{message_id}", "body": {"content": '{"text":"结合历史消息的回复"}'}, }, } async def _fake_run_chat_and_wait(*, workspace_id=None, message=None, user_email=None, **kwargs): prompts.append(str(message)) return "chat-session-test", AskResult( text="结合历史消息的回复", finish_reason="completed", error=None ) async def _fake_fetch_chat_history(*, chat_id: str, before_message_time=None, limit: int = 20): assert chat_id == "oc_group_fetch_history" return [ { "message_id": "om_old_1", "chat_id": chat_id, "chat_type": "group", "sender_id": "u_old_1", "sender_name": "Carol", "role": "user", "content": "之前我们已经确认是 OOM 引起的重启", "thread_root_id": None, "parent_id": None, "created_at": "2026-04-09T01:00:00+00:00", }, { "message_id": "om_old_2", "chat_id": chat_id, "chat_type": "group", "sender_id": "u_old_2", "sender_name": "David", "role": "user", "content": "后来大家又怀疑和网络抖动有关", "thread_root_id": None, "parent_id": None, "created_at": "2026-04-09T01:01:00+00:00", }, ] monkeypatch.setattr(lark_api, "add_message_reaction", _fake_add_message_reaction) monkeypatch.setattr(lark_api, "reply_text_message", _fake_reply_text_message) monkeypatch.setattr(lark_api, "run_chat_and_wait", _fake_run_chat_and_wait) monkeypatch.setattr(lark_api, "fetch_chat_history", _fake_fetch_chat_history) with TestClient(app) as client: summary_payload = { "header": { "event_id": "evt_group_fetch_history_1", "event_type": "im.message.receive_v1", "token": "test_token", }, "event": { "sender": {"sender_id": {"user_id": "u_group_fetch_history"}, "name": "Alice"}, "message": { "message_id": "om_group_fetch_history_1", "chat_id": "oc_group_fetch_history", "chat_type": "group", "message_type": "text", "content": '{"zh_cn":{"content":[[{"tag":"at","user_id":"ou_bot","user_name":"investibook-bot-test"},{"tag":"text","text":" 结合上面历史消息帮我总结一下"}]]}}', }, }, } summary_resp = client.post("/api/lark/events", json=summary_payload) assert summary_resp.status_code == 200 assert prompts assert "Carol: 之前我们已经确认是 OOM 引起的重启" in prompts[-1] assert "David: 后来大家又怀疑和网络抖动有关" in prompts[-1] def test_lark_webhook_stale_message_sends_timeout_reply(monkeypatch, tmp_path: Path): _configure_test_runtime(monkeypatch, tmp_path) replies: list[tuple[str, str]] = [] async def _fake_add_message_reaction(*, message_id: str, emoji_type=None): raise AssertionError("add_message_reaction should not be called for stale messages") async def _fake_reply_text_message(*, message_id: str, text: str): replies.append((message_id, text)) return {"code": 0} monkeypatch.setattr(lark_api, "add_message_reaction", _fake_add_message_reaction) monkeypatch.setattr(lark_api, "reply_text_message", _fake_reply_text_message) stale_ms = int((datetime.now(timezone.utc) - timedelta(hours=8)).timestamp() * 1000) with TestClient(app) as client: payload = { "header": { "event_id": "evt_stale_1", "event_type": "im.message.receive_v1", "token": "test_token", }, "event": { "sender": {"sender_id": {"user_id": "u_stale_1"}}, "message": { "message_id": "om_stale_1", "chat_id": "oc_stale_1", "chat_type": "p2p", "message_type": "text", "create_time": str(stale_ms), "content": '{"text":"诊断下 O11Y-2343244"}', }, }, } resp = client.post("/api/lark/events", json=payload) assert resp.status_code == 200 assert resp.json()["skipped"] == "timeout_reply" assert replies == [("om_stale_1", "回复超时，请联系管理员")] assert dbq._db_get_latest_feishu_session("u_stale_1", "oc_stale_1") is None def test_lark_webhook_group_inherits_issue_from_topic(monkeypatch, tmp_path: Path): _configure_test_runtime(monkeypatch, tmp_path) cards: list[dict] = [] async def _fake_reply_text_message(*, message_id: str, text: str): return {"code": 0} async def _fake_reply_interactive_message(*, message_id: str, card: dict): cards.append(card) return {"code": 0, "data": {"message_id": "om_group_issue_reply_1"}} monkeypatch.setattr(lark_api, "reply_text_message", _fake_reply_text_message) monkeypatch.setattr(lark_api, "reply_interactive_message", _fake_reply_interactive_message) with TestClient(app) as client: payload = { "header": { "event_id": "evt_group_2", "event_type": "im.message.receive_v1", "token": "test_token", }, "event": { "sender": {"sender_id": {"user_id": "u_group_2"}}, "topic": {"title": "O11Y-2343244 诊断话题"}, "message": { "message_id": "om_group_2", "chat_id": "oc_group_2", "chat_type": "group", "message_type": "text", "parent_id": "om_parent_1", "content": '{"zh_cn":{"content":[[{"tag":"at","user_id":"ou_bot","user_name":"investibook-bot-test"},{"tag":"text","text":" 帮我看下根因"}]]}}', }, }, } resp = client.post("/api/lark/events", json=payload) assert resp.status_code == 200 body = resp.json() assert body["issue_id"] == "O11Y-2343244" assert body["reused"] is False assert cards assert cards[0]["elements"][0]["elements"][0]["content"].startswith("Workspace:") assert "测试诊断回复" in cards[0]["elements"][1]["content"] row = dbq._db_get_latest_feishu_session("u_group_2", "oc_group_2") assert row is not None assert row["issue_id"] == "O11Y-2343244" assert row["latest_thread_id"] == "om_parent_1" assert row["workspace_id"] def test_lark_webhook_threaded_group_comment_still_requires_mention( monkeypatch, tmp_path: Path ): _configure_test_runtime(monkeypatch, tmp_path) async def _fake_reply_text_message(*, message_id: str, text: str): raise AssertionError("reply_text_message should not be called") monkeypatch.setattr(lark_api, "reply_text_message", _fake_reply_text_message) with TestClient(app) as client: payload = { "header": { "event_id": "evt_group_thread_nomention", "event_type": "im.message.receive_v1", "token": "test_token", }, "event": { "sender": {"sender_id": {"user_id": "u_group_thread_nomention"}}, "topic": {"title": "O11Y-2343244 诊断话题"}, "message": { "message_id": "om_group_thread_nomention", "chat_id": "oc_group_thread_nomention", "chat_type": "group", "message_type": "text", "parent_id": "om_parent_thread", "root_id": "om_root_thread", "content": '{"text":"诊断下"}', }, }, } resp = client.post("/api/lark/events", json=payload) assert resp.status_code == 200 assert resp.json()["ignored"] == "without_mention" def test_lark_webhook_non_p2p_comment_requires_mention(monkeypatch, tmp_path: Path): _configure_test_runtime(monkeypatch, tmp_path) async def _fake_reply_text_message(*, message_id: str, text: str): raise AssertionError("reply_text_message should not be called") monkeypatch.setattr(lark_api, "reply_text_message", _fake_reply_text_message) with TestClient(app) as client: payload = { "header": { "event_id": "evt_post_nomention", "event_type": "im.message.receive_v1", "token": "test_token", }, "event": { "sender": {"sender_id": {"user_id": "u_post_nomention"}}, "topic": {"title": "O11Y-2343244 诊断话题"}, "message": { "message_id": "om_post_nomention", "chat_id": "oc_post_nomention", "chat_type": "post", "message_type": "text", "parent_id": "om_parent_post", "root_id": "om_root_post", "content": '{"text":"诊断下"}', }, }, } resp = client.post("/api/lark/events", json=payload) assert resp.status_code == 200 assert resp.json()["ignored"] == "without_mention" def test_lark_webhook_topic_followup_reuses_root_issue(monkeypatch, tmp_path: Path): _configure_test_runtime(monkeypatch, tmp_path) replies: list[str] = [] async def _fake_reply_text_message(*, message_id: str, text: str): replies.append(text) return {"code": 0} async def _fake_fetch_message_context(message_id: str): if message_id == "om_topic_root": return { "message": { "message_id": "om_topic_root", "content": '{"text":"告警 Issue: O11Y-2343244"}', } } if message_id == "om_bot_reply": return { "message": { "message_id": "om_bot_reply", "root_id": "om_topic_root", "parent_id": "om_topic_root", "content": '{"text":"已为 O11Y-2343244 创建新会话"}', } } return None monkeypatch.setattr(lark_api, "reply_text_message", _fake_reply_text_message) monkeypatch.setattr(lark_core, "fetch_message_context", _fake_fetch_message_context) with TestClient(app) as client: first_payload = { "header": { "event_id": "evt_group_3", "event_type": "im.message.receive_v1", "token": "test_token", }, "event": { "sender": {"sender_id": {"user_id": "u_group_3"}}, "message": { "message_id": "om_comment_1", "chat_id": "oc_group_3", "chat_type": "group", "message_type": "text", "root_id": "om_topic_root", "parent_id": "om_topic_root", "content": '{"zh_cn":{"content":[[{"tag":"at","user_id":"ou_bot","user_name":"investibook-bot-test"},{"tag":"text","text":" 帮我看下根因"}]]}}', }, }, } first_resp = client.post("/api/lark/events", json=first_payload) second_payload = { "header": { "event_id": "evt_group_4", "event_type": "im.message.receive_v1", "token": "test_token", }, "event": { "sender": {"sender_id": {"user_id": "u_group_3"}}, "message": { "message_id": "om_comment_2", "chat_id": "oc_group_3", "chat_type": "group", "message_type": "text", "root_id": "om_topic_root", "parent_id": "om_bot_reply", "content": '{"zh_cn":{"content":[[{"tag":"at","user_id":"ou_bot","user_name":"investibook-bot-test"},{"tag":"text","text":" CPU 呢"}]]}}', }, }, } second_resp = client.post("/api/lark/events", json=second_payload) assert first_resp.status_code == 200 assert second_resp.status_code == 200 second_body = second_resp.json() assert second_body["issue_id"] == "O11Y-2343244" assert second_body["reused"] is True assert second_body["reason"] == "same_thread_root" row = dbq._db_get_latest_feishu_session("u_group_3", "oc_group_3") assert row is not None assert row["issue_id"] == "O11Y-2343244" assert row["thread_root_id"] == "om_topic_root" assert row["latest_message_id"] == "om_comment_2" def test_lark_webhook_thread_comment_reuses_topic_session_and_includes_root_context( monkeypatch, tmp_path: Path ): _configure_test_runtime(monkeypatch, tmp_path) chat_calls: list[dict[str, str | None]] = [] async def _fake_add_message_reaction(*, message_id: str, emoji_type=None): return {"code": 0} async def _fake_reply_text_message(*, message_id: str, text: str): return {"code": 0} async def _fake_run_chat_and_wait(*, workspace_id=None, message=None, user_email=None, **kwargs): chat_calls.append({"workspace_id": workspace_id, "message": message}) return "chat-session-test", AskResult( text="线程上下文回复", finish_reason="completed", error=None ) monkeypatch.setattr(lark_api, "add_message_reaction", _fake_add_message_reaction) monkeypatch.setattr(lark_api, "reply_text_message", _fake_reply_text_message) monkeypatch.setattr(lark_api, "run_chat_and_wait", _fake_run_chat_and_wait) with TestClient(app) as client: first_payload = { "header": { "event_id": "evt_thread_ctx_1", "event_type": "im.message.receive_v1", "token": "test_token", }, "event": { "sender": {"sender_id": {"user_id": "u_thread_ctx"}}, "message": { "message_id": "om_thread_ctx_root", "chat_id": "oc_thread_ctx", "chat_type": "group", "message_type": "text", "thread_id": "omt_thread_ctx_topic", "content": '{"zh_cn":{"content":[[{"tag":"at","user_id":"ou_bot","user_name":"investibook-bot-test"},{"tag":"text","text":" 10698648422638660035 这个集群最近在做什么操作"}]]}}', }, }, } second_payload = { "header": { "event_id": "evt_thread_ctx_2", "event_type": "im.message.receive_v1", "token": "test_token", }, "event": { "sender": {"sender_id": {"user_id": "u_thread_ctx"}}, "message": { "message_id": "om_thread_ctx_followup", "chat_id": "oc_thread_ctx", "chat_type": "group", "message_type": "text", "root_id": "om_thread_ctx_root", "parent_id": "om_thread_ctx_root", "content": '{"zh_cn":{"content":[[{"tag":"at","user_id":"ou_bot","user_name":"investibook-bot-test"},{"tag":"text","text":" 谁操作的 tikv 配置变更"}]]}}', }, }, } first_resp = client.post("/api/lark/events", json=first_payload) second_resp = client.post("/api/lark/events", json=second_payload) assert first_resp.status_code == 200 assert second_resp.status_code == 200 assert second_resp.json()["reused"] is True assert second_resp.json()["reason"] == "same_thread_root" assert len(chat_calls) == 2 assert chat_calls[0]["workspace_id"] == chat_calls[1]["workspace_id"] assert "Recent group messages:" in str(chat_calls[1]["message"]) assert "10698648422638660035 这个集群最近在做什么操作" in str( chat_calls[1]["message"] ) def test_lark_webhook_metric_followup_reuses_latest_issue_session( monkeypatch, tmp_path: Path ): _configure_test_runtime(monkeypatch, tmp_path) replies: list[str] = [] chat_calls: list[dict[str, str | None]] = [] async def _fake_reply_text_message(*, message_id: str, text: str): replies.append(text) return {"code": 0} async def _fake_run_chat_and_wait(*, workspace_id=None, message=None, user_email=None, **kwargs): chat_calls.append( { "workspace_id": workspace_id, "message": message, } ) return "chat-session-test", AskResult( text="指标诊断回复", finish_reason="completed", error=None ) monkeypatch.setattr(lark_api, "reply_text_message", _fake_reply_text_message) monkeypatch.setattr(lark_api, "run_chat_and_wait", _fake_run_chat_and_wait) with TestClient(app) as client: first_payload = { "header": { "event_id": "evt_metric_1", "event_type": "im.message.receive_v1", "token": "test_token", }, "event": { "sender": {"sender_id": {"user_id": "u_metric_1"}}, "message": { "message_id": "om_metric_1", "chat_id": "oc_metric_1", "chat_type": "group", "message_type": "text", "content": '{"zh_cn":{"content":[[{"tag":"at","user_id":"ou_bot","user_name":"investibook-bot-test"},{"tag":"text","text":" 诊断下 O11Y-2343244"}]]}}', }, }, } second_payload = { "header": { "event_id": "evt_metric_2", "event_type": "im.message.receive_v1", "token": "test_token", }, "event": { "sender": {"sender_id": {"user_id": "u_metric_1"}}, "message": { "message_id": "om_metric_2", "chat_id": "oc_metric_1", "chat_type": "group", "message_type": "text", "content": '{"zh_cn":{"content":[[{"tag":"at","user_id":"ou_bot","user_name":"investibook-bot-test"},{"tag":"text","text":" 你能看到这个集群 changefeed 的一些指标么"}]]}}', }, }, } first_resp = client.post("/api/lark/events", json=first_payload) second_resp = client.post("/api/lark/events", json=second_payload) assert first_resp.status_code == 200 assert second_resp.status_code == 200 second_body = second_resp.json() assert second_body["reused"] is True assert second_body["reason"] == "recent_session" assert len(chat_calls) == 2 assert "Current issue id: O11Y-2343244" in str(chat_calls[1]["message"]) def test_lark_webhook_followup_without_keywords_still_reuses_recent_issue_session( monkeypatch, tmp_path: Path ): _configure_test_runtime(monkeypatch, tmp_path) async def _fake_reply_text_message(*, message_id: str, text: str): return {"code": 0} async def _fake_run_chat_and_wait(*, workspace_id=None, message=None, user_email=None, **kwargs): return "chat-session-test", AskResult( text="继续对话回复", finish_reason="completed", error=None ) monkeypatch.setattr(lark_api, "reply_text_message", _fake_reply_text_message) monkeypatch.setattr(lark_api, "run_chat_and_wait", _fake_run_chat_and_wait) with TestClient(app) as client: first_payload = { "header": { "event_id": "evt_recent_1", "event_type": "im.message.receive_v1", "token": "test_token", }, "event": { "sender": {"sender_id": {"user_id": "u_recent_1"}}, "message": { "message_id": "om_recent_1", "chat_id": "oc_recent_1", "chat_type": "group", "message_type": "text", "content": '{"zh_cn":{"content":[[{"tag":"at","user_id":"ou_bot","user_name":"investibook-bot-test"},{"tag":"text","text":" 诊断下 O11Y-2343244"}]]}}', }, }, } second_payload = { "header": { "event_id": "evt_recent_2", "event_type": "im.message.receive_v1", "token": "test_token", }, "event": { "sender": {"sender_id": {"user_id": "u_recent_1"}}, "message": { "message_id": "om_recent_2", "chat_id": "oc_recent_1", "chat_type": "group", "message_type": "text", "content": '{"zh_cn":{"content":[[{"tag":"at","user_id":"ou_bot","user_name":"investibook-bot-test"},{"tag":"text","text":" 收到我的消息了么"}]]}}', }, }, } first_resp = client.post("/api/lark/events", json=first_payload) second_resp = client.post("/api/lark/events", json=second_payload) assert first_resp.status_code == 200 assert second_resp.status_code == 200 second_body = second_resp.json() assert second_body["reused"] is True assert second_body["reason"] == "recent_session" def test_lark_webhook_merge_forward_message_is_expanded(monkeypatch, tmp_path: Path): _configure_test_runtime(monkeypatch, tmp_path) prompts: list[str] = [] cards: list[tuple[str, dict]] = [] async def _fake_add_message_reaction(*, message_id: str, emoji_type=None): return {"code": 0} async def _fake_reply_text_message(*, message_id: str, text: str): return {"code": 0} async def _fake_reply_interactive_message(*, message_id: str, card: dict): cards.append((message_id, card)) return {"code": 0, "data": {"message_id": f"om_bot_{message_id}"}} async def _fake_run_chat_and_wait(*, workspace_id=None, message=None, user_email=None, **kwargs): prompts.append(str(message)) return "chat-session-test", AskResult( text="合并转发诊断回复", finish_reason="completed", error=None ) monkeypatch.setattr(lark_api, "add_message_reaction", _fake_add_message_reaction) monkeypatch.setattr(lark_api, "reply_text_message", _fake_reply_text_message) monkeypatch.setattr(lark_api, "reply_interactive_message", _fake_reply_interactive_message) monkeypatch.setattr(lark_api, "run_chat_and_wait", _fake_run_chat_and_wait) merge_forward_content = json.dumps({ "msg_list": [ { "sender": {"name": "Alice"}, "body": {"content": '{"text":"O11Y-2343244 这台机器 CPU 飙高了"}'}, }, { "sender": {"name": "Bob"}, "body": {"content": '{"text":"我看监控也有异常，需要紧急诊断"}'}, }, ] }) with TestClient(app) as client: payload = { "header": { "event_id": "evt_merge_forward_1", "event_type": "im.message.receive_v1", "token": "test_token", }, "event": { "sender": {"sender_id": {"user_id": "u_merge_forward_1"}}, "message": { "message_id": "om_merge_forward_1", "chat_id": "oc_merge_forward_1", "chat_type": "p2p", "message_type": "merge_forward", "content": merge_forward_content, }, }, } resp = client.post("/api/lark/events", json=payload) assert resp.status_code == 200 body = resp.json() assert body["ok"] is True assert cards assert "合并转发诊断回复" in cards[0][1]["elements"][1]["content"] assert prompts assert "Alice: O11Y-2343244 这台机器 CPU 飙高了" in prompts[-1] assert "Bob: 我看监控也有异常，需要紧急诊断" in prompts[-1]