From 43c34d290c81f822a7ca5ca680cfae6e4bca55a1 Mon Sep 17 00:00:00 2001 From: Caldalis Date: Wed, 26 Aug 2026 15:25:53 +0800 Subject: [PATCH] fix(channels): route handoff reply ack by channel --- backend/app/channels/service_intake.py | 16 +- backend/tests/test_channel_handoff_reply.py | 216 ++++++++++++++++++++ backend/tests/test_feishu_handoff.py | 34 ++- 3 files changed, 252 insertions(+), 14 deletions(-) create mode 100644 backend/tests/test_channel_handoff_reply.py diff --git a/backend/app/channels/service_intake.py b/backend/app/channels/service_intake.py index eada2edb8..bce10ce41 100644 --- a/backend/app/channels/service_intake.py +++ b/backend/app/channels/service_intake.py @@ -973,8 +973,9 @@ def _run_handoff_reply_command( binding: ChannelBinding, inbound: ChannelInbound, command: ChannelCommand, + target: dict, ) -> str: - """/回复反馈 指令处理:处理人通过飞书发送 /回复反馈 <内容> 回复人工转接通知。 + """/回复反馈 指令处理:处理人发送 /回复反馈 <内容> 回复人工转接通知。 匹配策略(按优先级): 1. 引用通知(parent_id):按 handoff.notify_message_id == parent_id 精确匹配。 @@ -983,6 +984,10 @@ def _run_handoff_reply_command( 提示处理人回复对应通知消息。 命中后复用 _apply_handoff_reply 置 answered + 恢复 SOP,并给处理人回确认。 无 ChannelIdentity 时拒绝。 + + 确认回执沿用调用方解析出的渠道 target(暂存事件的 target_json;微信走兜底值), + 与其余指令回复一致。此处不能写死飞书的 receive_id 字段:企微/微信的 send() 读 + to_user_id,钉钉读 session_webhook,收到飞书格式会直接抛错、投递失败。 """ reply_text = command.query.strip() if not reply_text: @@ -1067,7 +1072,7 @@ def _run_handoff_reply_command( handoff, reply_text, answered_by_user_id=answered_by, - source="feishu", + source=binding.channel, ) # 给处理人回一条确认(经 outbox 投递) db.add( @@ -1076,10 +1081,7 @@ def _run_handoff_reply_command( binding_id=binding.id, session_id=f"handoff:{handoff.id}", message_id=None, - target_json={ - "receive_id_type": "open_id", - "receive_id": inbound.from_user_id, - }, + target_json=dict(target), kind="handoff_ack", text=f"已收到你的回复，正在恢复 SOP 执行。回复预览：{reply_text[:120]}", status="pending", @@ -1225,7 +1227,7 @@ def process_inbound( if command.kind in {"bind", "unbind"}: reply = _run_bind_command(db, binding, inbound, command) elif command.kind == "handoff_reply": - reply = _run_handoff_reply_command(db, binding, inbound, command) + reply = _run_handoff_reply_command(db, binding, inbound, command, target) elif binding.team_id: # 团队绑定:消息直路由团队 TL,员工列表/切换等指令无意义 reply = "该渠道已接入团队，消息由团队 TL 统一接收，员工切换类指令不可用。" diff --git a/backend/tests/test_channel_handoff_reply.py b/backend/tests/test_channel_handoff_reply.py new file mode 100644 index 000000000..2622679cc --- /dev/null +++ b/backend/tests/test_channel_handoff_reply.py @@ -0,0 +1,216 @@ +"""/回复反馈 指令的确认回执必须按渠道构造投递目标 +回归 backdrop:确认回执曾写死飞书的 receive_id/receive_id_type,而该指令对四个 +渠道都开放。微信的 send() 读 to_user_id、钉钉读 session_webhook,收到飞书格式会直接抛错,处理人拿不到"已收到你的回复"。 +""" + +from __future__ import annotations + +import pytest +from sqlalchemy.pool import StaticPool +from sqlmodel import Session, SQLModel, create_engine, select + +import app.channels.service_intake as intake_mod +from app.channels.adapters.base import ChannelInbound +from app.channels.crypto import encrypt_channel_secret +from app.channels.service_intake import _run_handoff_reply_command +from app.channels.service_routing import ChannelCommand +from app.db.models import ( + ChannelBinding, + ChannelDelivery, + ChannelIdentity, + ChatSession, + HumanHandoffRequest, + Tenant, + User, + utc_now, +) + +# 各渠道入站时构造并暂存到 ChannelInboundEvent.target_json 的投递目标, +# 与 feishu_runtime / service_wecom_inbox / service_dingtalk_inbox 保持一致; +# 微信不进暂存守护,走 process_inbound 的兜底 target +# required_key 是该渠道 send() 用来定位收件人的字段,缺失即投递失败 +CHANNEL_TARGETS = { + "feishu": ( + { + "message_id": "evt_1", + "reply_in_thread": False, + "receive_id_type": "open_id", + "receive_id": "u_assignee", + }, + "receive_id", + ), + "wecom": ({"to_user_id": "u_assignee", "context_token": "ctx_1"}, "to_user_id"), + "wechat": ({"to_user_id": "u_assignee", "context_token": "ctx_1"}, "to_user_id"), + "dingtalk": ( + { + "to_user_id": "u_assignee", + "context_token": "https://oapi.dingtalk.com/robot/send?access_token=t", + "session_webhook": "https://oapi.dingtalk.com/robot/send?access_token=t", + "session_webhook_expired_time": 0, + "conversation_id": "cid_1", + "conversation_type": "1", + "message_id": "evt_1", + }, + "session_webhook", + ), +} + +def _test_engine(): + engine = create_engine( + "sqlite://", + connect_args={"check_same_thread": False}, + poolclass=StaticPool, + ) + SQLModel.metadata.create_all(engine) + return engine + + +def _seed(db: Session, channel: str) -> ChannelBinding: + db.add(Tenant(id="tenant_demo", name="Demo")) + db.add( + User( + id="assignee_user", + tenant_id="tenant_demo", + username="assignee", + display_name="指派人", + password_hash="x", + ) + ) + binding = ChannelBinding( + id=f"binding_{channel}", + tenant_id="tenant_demo", + agent_id="agent_demo", + channel=channel, + status="active", + config_json={}, + credentials_enc=encrypt_channel_secret("secret-value"), + external_account_key=f"{channel}:test", + config_revision=1, + ) + db.add(binding) + db.add( + ChannelIdentity( + tenant_id="tenant_demo", + channel=channel, + external_account_scope="", + external_user_id="u_assignee", + staffdeck_user_id="assignee_user", + ) + ) + db.add( + ChatSession( + id="session_1", + tenant_id="tenant_demo", + agent_id="agent_demo", + status="handoff", + ) + ) + db.add( + HumanHandoffRequest( + id="handoff_1", + tenant_id="tenant_demo", + session_id="session_1", + agent_id="agent_demo", + assignee_user_id="assignee_user", + pending_question="网络故障", + context_summary="user: 网络断了", + status="pending", + ) + ) + db.commit() + return binding + +def _inbound(channel: str) -> ChannelInbound: + return ChannelInbound( + channel=channel, + event_id="evt_1", + from_user_id="u_assignee", + to_user_id="u_bot", + session_id="u_assignee", + group_id="", + context_token="ctx_1", + text="/回复反馈 已修复网络", + is_group=False, + raw={}, + ) + +def _run(db: Session, binding: ChannelBinding, channel: str, monkeypatch): + """执行 /回复反馈 指令,返回 (回执投递, _apply_handoff_reply 收到的 source)""" + seen: list[str] = [] + def fake_apply(db_arg, row, reply, *, answered_by_user_id, source="web"): + row.status = "answered" + row.human_reply = reply + row.answered_at = utc_now() + db_arg.add(row) + db_arg.commit() + seen.append(source) + + + import app.api.chat as chat_api + monkeypatch.setattr(chat_api, "_apply_handoff_reply", fake_apply) + monkeypatch.setattr(intake_mod, "external_account_scope", lambda _db, _b: "") + target, _ = CHANNEL_TARGETS[channel] + result = _run_handoff_reply_command( + db, + binding, + _inbound(channel), + ChannelCommand(kind="handoff_reply", query="已修复网络"), + dict(target), + ) + assert result is intake_mod._HANDOFF_REPLY_HANDLED + ack = db.exec( + select(ChannelDelivery).where(ChannelDelivery.kind == "handoff_ack") + ).first() + assert ack is not None + return ack, seen + + +@pytest.mark.parametrize("channel", sorted(CHANNEL_TARGETS)) +def test_handoff_reply_ack_uses_channel_target(channel: str, monkeypatch) -> None: + """回执投递目标沿用本渠道的target，而非写死飞书字段""" + engine = _test_engine() + with Session(engine) as db: + binding = _seed(db, channel) + ack, _ = _run(db, binding, channel, monkeypatch) + + target, required_key = CHANNEL_TARGETS[channel] + assert ack.target_json == target + # send靠该字段定位收件人,缺失时适配器直接抛错,回执永远送不达 + assert ack.target_json.get(required_key) + assert "已收到你的回复" in ack.text +@pytest.mark.parametrize("channel", sorted(CHANNEL_TARGETS)) +def test_handoff_reply_records_originating_channel(channel: str, monkeypatch) -> None: + """人工答复来源记为实际渠道,不再一律记成feishu""" + engine = _test_engine() + with Session(engine) as db: + binding = _seed(db, channel) + _, seen = _run(db, binding, channel, monkeypatch) + + assert seen == [channel] + assert db.get(HumanHandoffRequest, "handoff_1").status == "answered" + +@pytest.mark.parametrize("channel", ["wecom", "wechat", "dingtalk"]) +def test_feishu_shaped_target_is_undeliverable_elsewhere(channel: str) -> None: + """写死的飞书目标在其他渠道上不可投递——这正是本次修复要避免的回归 + 三个适配器都在触网前校验目标字段,故此断言与网络无关 + """ + from app.channels.adapters.dingtalk import DingTalkAdapter + from app.channels.adapters.wechat import WeChatAdapter + from app.channels.adapters.wecom import WeComAdapter + adapter = { + "wecom": WeComAdapter, + "wechat": WeChatAdapter, + "dingtalk": DingTalkAdapter, + }[channel]() + binding = ChannelBinding( + id=f"binding_{channel}", + tenant_id="tenant_demo", + agent_id="agent_demo", + channel=channel, + status="active", + config_json={}, + ) + feishu_target = {"receive_id_type": "open_id", "receive_id": "u_assignee"} + with pytest.raises(Exception) as excinfo: + adapter.send(binding, feishu_target, "已收到你的回复") + assert "目标" in str(excinfo.value) diff --git a/backend/tests/test_feishu_handoff.py b/backend/tests/test_feishu_handoff.py index 2c6941fae..ad0d5cfec 100644 --- a/backend/tests/test_feishu_handoff.py +++ b/backend/tests/test_feishu_handoff.py @@ -142,6 +142,14 @@ def _inbound( parent_id=parent_id, ) +def _feishu_target(inbound: ChannelInbound) -> dict: + """飞书入站在 feishu_runtime 侧构造并暂存的投递 target。""" + return { + "message_id": inbound.event_id, + "reply_in_thread": False, + "receive_id_type": "open_id", + "receive_id": inbound.from_user_id, + } def _inbound_event( *, @@ -1731,7 +1739,9 @@ def fake_apply(db_arg, row, reply, *, answered_by_user_id, source="web"): original = intake_mod.external_account_scope intake_mod.external_account_scope = lambda _db, _b: "" try: - result = _run_handoff_reply_command(db, binding, inbound, command) + result = _run_handoff_reply_command( + db, binding, inbound, command, _feishu_target(inbound) + ) assert resumed == [("handoff_hr1", "feishu")] assert result is intake_mod._HANDOFF_REPLY_HANDLED assert db.get(HumanHandoffRequest, "handoff_hr1").status == "answered" @@ -1777,7 +1787,9 @@ def test_run_handoff_reply_command_rejects_without_identity() -> None: original = intake_mod.external_account_scope intake_mod.external_account_scope = lambda _db, _b: "" try: - result = _run_handoff_reply_command(db, binding, inbound, command) + result = _run_handoff_reply_command( + db, binding, inbound, command, _feishu_target(inbound) + ) assert "未找到" in result or "未绑定" in result assert db.get(HumanHandoffRequest, "handoff_hr2").status == "pending" finally: @@ -1803,7 +1815,9 @@ def test_run_handoff_reply_command_no_pending_handoff_returns_error() -> None: original = intake_mod.external_account_scope intake_mod.external_account_scope = lambda _db, _b: "" try: - result = _run_handoff_reply_command(db, binding, inbound, command) + result = _run_handoff_reply_command( + db, binding, inbound, command, _feishu_target(inbound) + ) assert "未找到" in result finally: intake_mod.external_account_scope = original @@ -1823,7 +1837,7 @@ def test_run_handoff_reply_command_empty_query_returns_usage() -> None: inbound = _inbound(event_id="om_hr_4", text="/回复反馈") command = ChannelCommand(kind="handoff_reply", query="") - result = _run_handoff_reply_command(db, binding, inbound, command) + result = _run_handoff_reply_command(db, binding, inbound, command, _feishu_target(inbound)) assert "用法" in result @@ -1899,7 +1913,9 @@ def fake_apply(db_arg, row, reply, *, answered_by_user_id, source="web"): original = intake_mod.external_account_scope intake_mod.external_account_scope = lambda _db, _b: "" try: - result = _run_handoff_reply_command(db, binding, inbound, command) + result = _run_handoff_reply_command( + db, binding, inbound, command, _feishu_target(inbound) + ) assert resumed == [] assert "多个待处理" in result finally: @@ -1928,7 +1944,9 @@ def test_run_handoff_reply_command_rejects_unknown_parent_id() -> None: original = intake_mod.external_account_scope intake_mod.external_account_scope = lambda _db, _b: "" try: - result = _run_handoff_reply_command(db, binding, inbound, command) + result = _run_handoff_reply_command( + db, binding, inbound, command, _feishu_target(inbound) + ) assert "未找到" in result assert db.get(HumanHandoffRequest, handoff.id).status == "pending" finally: @@ -2026,7 +2044,9 @@ def fake_apply(db_arg, row, reply, *, answered_by_user_id, source="web"): original = intake_mod.external_account_scope intake_mod.external_account_scope = lambda _db, _b: "" try: - result = _run_handoff_reply_command(db, binding, inbound, command) + result = _run_handoff_reply_command( + db, binding, inbound, command, _feishu_target(inbound) + ) assert resumed == ["handoff_p1"] assert result is intake_mod._HANDOFF_REPLY_HANDLED finally: