"""续办历史工单：模型意图合同 + 回到当时节点。""" from __future__ import annotations import os import sys import unittest ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) if ROOT not in sys.path: sys.path.insert(0, ROOT) os.environ.setdefault("YTD_ENV", "test") os.environ.setdefault("REDIS_BACKEND", "memory") os.environ.setdefault("CHECKPOINT_BACKEND", "memory") os.environ.setdefault("MESSAGE_STORE_BACKEND", "memory") os.environ.setdefault("LLM_DATA_USAGE_CONFIRMED", "false") os.environ.setdefault("LLM_ALLOW_NETWORK", "false") os.environ.setdefault("LEDGER_BACKEND", "memory") from agent.channel.wecom.models import InboundMessage from agent.handlers.continue_thread import handle_continue_thread from agent.ledger.memory_ledger import MemoryLedger from agent.policy import inquiry_copy as copy from agent.policy.air_text_flow import AirTextInquiryFlow from agent.routing import IntentRouter from agent.routing.deepseek_a import DeepSeekARouter from agent.routing.dispatch import dispatch_inbound from agent.routing.ordered_actions import parse_ordered_actions_arguments COMPLETE_AIR = { "起运港": "上海", "目的港": "洛杉矶", "品名": "普货", "件数": "3", "毛重": "180", "体积": "2.5", "包装方式": "纸箱", } class ContinueTicketTests(unittest.TestCase): def setUp(self) -> None: from agent.redis_coord.runtime import shutdown_redis_runtime shutdown_redis_runtime() def test_model_contract_accepts_continue_intent(self) -> None: digest = "a" * 64 parsed = parse_ordered_actions_arguments( { "decision_id": "d1", "intent": "continue_thread", "ordered_action": { "action_type": "CONTINUE_HISTORICAL_TICKET", "order": 1, "target": {"work_order_no": "WO202609140034"}, "evidence": { "source_ref": "inbound:u1", "content_digest": digest, }, }, } ) decision = parsed.to_route_decision() self.assertEqual(decision.intent, "continue_thread") self.assertEqual(decision.thread_id, "WO202609140034") def test_model_junk_digest_still_continues(self) -> None: """测服实测：模型会乱填 content_digest，不能因此整句判失败。""" router = DeepSeekARouter.from_settings() decision = router.route_from_model_args( { "decision_id": "x", "intent": "continue_thread", "ordered_action": { "action_type": "CONTINUE_HISTORICAL_TICKET", "order": "1", "target": {"work_order_no": "WO202609140037"}, "evidence": { "source_ref": "user", "content_digest": "not-a-hash", }, }, }, text="我要继续聊聊WO202609140037", sender_id="WuJiLin", ) self.assertEqual(decision.intent, "continue_thread") self.assertEqual(decision.thread_id, "WO202609140037") def test_stub_natural_language_not_fixed_command(self) -> None: router = DeepSeekARouter.from_settings() hit = router.route_stub_fixture( text="我要继续咨询工单WO202609140034", sender_id="WuJiLin", ) self.assertEqual(hit.intent, "continue_thread") self.assertEqual(hit.thread_id, "WO202609140034") again = router.route_stub_fixture( text="把刚才那单 WO202609140034 再出一下方案", sender_id="WuJiLin", ) self.assertEqual(again.intent, "continue_thread") fresh = router.route_stub_fixture( text="上海到洛杉矶空运，3件 180kg 茶叶", sender_id="WuJiLin", ) self.assertEqual(fresh.intent, "ordinary_text_inquiry") self.assertEqual(fresh.thread_id, "") # 无网夹具认不出「只贴工单号」；正式环境必须由模型认。 bare = router.route_stub_fixture( text="WO202609140034", sender_id="WuJiLin", ) self.assertEqual(bare.intent, "ordinary_text_inquiry") def test_card_click_skips_model(self) -> None: router = IntentRouter(allow_network=False) hit = router.route_inbound( text="adopt_excel:WO202609140034", sender_id="WuJiLin", event="template_card_event", ) self.assertEqual(hit.intent, "card_action") def test_resume_scheme_from_session(self) -> None: flow = AirTextInquiryFlow(MemoryLedger()) replies: list[str] = [] extras: list[dict] = [] def reply(text: str, extra=None) -> None: replies.append(text) extras.append(dict(extra) if extra else {}) flow.on_text( sender_id="u1", text="空运", reply=reply, injected_facts=COMPLETE_AIR, injected_mode="AIR", ) wo = flow.session_of("u1").work_order_no replies.clear() extras.clear() flow.on_button(sender_id="u1", action=copy.BTN_SKIP_COLLAB, reply=reply) self.assertEqual(flow.session_of("u1").phase, "wait_adopt") replies.clear() extras.clear() phase = flow.resume_historical( sender_id="u1", work_order_no=wo, reply=reply ) self.assertEqual(phase, "wait_adopt") self.assertTrue(any(f"{wo}报价方案" in x for x in replies)) self.assertTrue( any((x.get("template_card") or {}).get("main_title", {}).get("title") == f"{wo}报价方案" for x in extras) ) def test_resume_from_ledger_quoted(self) -> None: ledger = MemoryLedger() flow = AirTextInquiryFlow(ledger) created = ledger.create_ticket( sender_id="u2", business_line="AIR", facts=COMPLETE_AIR ) wo = created["work_order_no"] ledger.upsert_quote( work_order_no=wo, quote={ "total": "USD 1,280", "eta": "3-5 天", "valid_until": "2026-09-20", "source_label": "TMS 标准报价", "fee_lines": [{"name": "空运费", "amount": "USD 1,200"}], }, to_status="已报价", ) replies: list[str] = [] def reply(text: str, extra=None) -> None: replies.append(text) _ = extra phase = flow.resume_historical( sender_id="u2", work_order_no=wo, reply=reply ) self.assertEqual(phase, "wait_adopt") self.assertTrue(any("报价方案" in x for x in replies)) def test_resume_after_quote_file_is_deal(self) -> None: ledger = MemoryLedger() created = ledger.create_ticket( sender_id="u6", business_line="AIR", facts=COMPLETE_AIR ) wo = created["work_order_no"] ledger.upsert_quote( work_order_no=wo, quote={"total": "USD 1", "eta": "-", "valid_until": "-"}, to_status="已报价", ) ledger.render_quote( work_order_no=wo, business_line="AIR", facts=COMPLETE_AIR, quote={"total": "USD 1"}, ) flow = AirTextInquiryFlow(ledger) replies: list[str] = [] def reply(text: str, extra=None) -> None: replies.append(text) _ = extra phase = flow.resume_historical( sender_id="u6", work_order_no=wo, reply=reply ) self.assertEqual(phase, "wait_deal") self.assertFalse(any("成交" in x or copy.BTN_DEAL in x for x in replies)) def test_resume_missing_wo_asks(self) -> None: flow = AirTextInquiryFlow(MemoryLedger()) replies: list[str] = [] phase = flow.resume_historical( sender_id="u3", work_order_no="", reply=replies.append ) self.assertEqual(phase, "need_work_order") self.assertTrue(any("工单号" in x for x in replies)) def test_dispatch_passes_work_order(self) -> None: ledger = MemoryLedger() flow = AirTextInquiryFlow(ledger) created = ledger.create_ticket( sender_id="u4", business_line="AIR", facts=COMPLETE_AIR ) wo = created["work_order_no"] ledger.upsert_quote( work_order_no=wo, quote={"total": "USD 1", "eta": "-", "valid_until": "-"}, to_status="已报价", ) from agent.channel.queue import MemoryMessageStore from agent.routing.decision import RouteDecision store = MemoryMessageStore() msg = InboundMessage( sender_id="u4", message_id="m-cont", content="帮我把那单再看一眼", ) phase = handle_continue_thread( msg, thread_id=wo, flow=flow, store=store ) self.assertEqual(phase, "wait_adopt") decision = RouteDecision(intent="continue_thread", thread_id=wo) # dispatch 走全局 flow，这里只断言意图表仍指向 continue self.assertEqual(decision.intent, "continue_thread") _ = dispatch_inbound def test_adopt_excel_after_restart_stays_on_card_ticket(self) -> None: """点 0040 方案卡不能落到刚续办的 0037：进程无书签时按卡上工单号从主账收回。""" ledger = MemoryLedger() live = AirTextInquiryFlow(ledger) sink: list[str] = [] def reply(text: str, extra=None) -> None: sink.append(text) _ = extra live.on_text( sender_id="u5", text="空运", reply=reply, injected_facts=COMPLETE_AIR, injected_mode="AIR", ) wo_old = live.session_of("u5").work_order_no live.on_button(sender_id="u5", action=copy.BTN_SKIP_COLLAB, reply=reply) other = dict(COMPLETE_AIR) other["品名"] = "服装" live.on_text( sender_id="u5", text="空运", reply=reply, injected_facts=other, injected_mode="AIR", ) wo_new = live.session_of("u5").work_order_no self.assertNotEqual(wo_old, wo_new) restarted = AirTextInquiryFlow(ledger) restarted.resume_historical( sender_id="u5", work_order_no=wo_new, reply=reply ) sink.clear() phase = restarted.on_button( sender_id="u5", action=copy.BTN_ADOPT_EXCEL, reply=reply, card_meta={ "event_key": f"adopt_excel:{wo_old}", "task_id": f"scheme-{wo_old}-deadbeef", "response_code": "rc-old-scheme", }, ) self.assertEqual(phase, "wait_deal") joined = "\n".join(sink) self.assertIn(wo_old, joined) self.assertNotIn(wo_new, joined) self.assertIn("Excel", joined) if __name__ == "__main__": unittest.main()