落地陆运群内协同：按填写样例拉群并只列本单协同项，进群把私聊已抽到的字段带上。

Co-authored-by: Cursor <cursoragent@cursor.com>

inquiry-agent/tests/test_air_group_collab.py

@@ -226,6 +226,37 @@ class AirGroupCollabTests(unittest.TestCase):
         self.assertEqual(phase, "activated")
         self.assertEqual(self.ledger.find_by_collab_chat(chat_id="wr_air_1").work_order_no, wo2)
 
+    def test_activate_prefills_magnet(self) -> None:
+        created = self.ledger.create_ticket(
+            sender_id="sales1",
+            business_line="AIR",
+            facts={
+                "起运地": "上海",
+                "目的地": "洛杉矶",
+                "品名": "配件",
+                "件数": "3",
+                "毛重": "100",
+                "体积": "2",
+                "包装类型": "纸箱",
+                "报价日期": "2026-09-16",
+                "是否含磁": "是",
+            },
+        )
+        wo = created["work_order_no"]
+        self.ledger.upsert_quote(
+            work_order_no=wo,
+            quote={"total": "USD 800", "source": "TMS", "quoteId": "QT-AIR-1"},
+            to_status="已报价",
+        )
+        handle_air_group(
+            self._msg(sender="sales1", text=wo, chat="wr_air_1", mid="pre1"),
+            flow=self.flow,
+            ledger=self.ledger,
+        )
+        joined = "\n".join(self._texts())
+        self.assertIn("是否含磁：是", joined)
+        self.assertIn("是否含电：-", joined)
+
     def test_second_group_occupied(self) -> None:
         wo = self._create_air()
         handle_air_group(

inquiry-agent/tests/test_land_collab_fields.py

@@ -0,0 +1,102 @@
+"""
+陆运群协同项：六套名单、展示名、预填、口语危险品。
+不连企微 / TMS / 主账。
+"""
+
+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)
+
+from agent.llm.extract_text import extract_collab_fields
+from agent.schema.field_validate import (
+    collab_display_name,
+    collab_fields_for_ticket,
+    harvest_oral_collab,
+    land_collab_fields,
+    pick_collab_from_facts,
+)
+
+
+class LandCollabFieldsTest(unittest.TestCase):
+    def test_domestic_ltl_keys(self):
+        keys = land_collab_fields("国内运输拼车", "国内长途/零担")
+        self.assertEqual(keys, ("客户名称", "包装方式", "货值", "是否为危险品"))
+
+    def test_domestic_ftl_longhaul_keys(self):
+        keys = land_collab_fields("国内运输整车", "国内长途/零担")
+        self.assertEqual(keys, ("客户名称", "货值", "包装方式", "是否为危险品"))
+
+    def test_domestic_ftl_order(self):
+        keys = land_collab_fields("国内运输整车", "国内城配/拖车/打包")
+        self.assertEqual(keys, ("客户名称", "货值", "包装方式", "是否为危险品"))
+
+    def test_cross_ftl_no_vehicle(self):
+        keys = land_collab_fields("跨境整车", "东南亚")
+        self.assertIn("HS编码", keys)
+        self.assertNotIn("车型/数量", keys)
+
+    def test_hk_ftl_has_vehicle(self):
+        keys = land_collab_fields("中港整车", "中港/中亚/中欧")
+        self.assertIn("车型/数量", keys)
+        self.assertEqual(keys[0], "客户名称")
+
+    def test_unlisted_pair_empty(self):
+        self.assertEqual(land_collab_fields("国内运输拼车", "东南亚"), ())
+
+    def test_land_hs_display(self):
+        self.assertEqual(collab_display_name("HS编码", "LAND"), "海关编码")
+        self.assertEqual(collab_display_name("HS编码", "SEA"), "商品海关编码")
+
+    def test_ticket_sea_air(self):
+        self.assertIn("是否含电", collab_fields_for_ticket({}, "SEA"))
+        self.assertEqual(
+            collab_fields_for_ticket({}, "AIR"),
+            ("是否含电", "是否含磁"),
+        )
+
+    def test_pick_skips_inquiry_vehicle(self):
+        got = pick_collab_from_facts(
+            {"包装方式": "纸箱", "车型/数量": "9米/1", "海关编码": "8479"},
+            land_collab_fields("跨境整车", "东南亚"),
+        )
+        self.assertEqual(got.get("包装方式"), "纸箱")
+        self.assertEqual(got.get("HS编码"), "8479")
+        self.assertNotIn("车型/数量", got)
+
+    def test_dg_oral(self):
+        self.assertEqual(harvest_oral_collab("普货")["是否为危险品"], "否")
+        self.assertEqual(harvest_oral_collab("危险品")["是否为危险品"], "是")
+        self.assertEqual(harvest_oral_collab("不是危险品")["是否为危险品"], "否")
+
+    def test_labeled_without_colon(self):
+        allowed = land_collab_fields("国内运输拼车", "国内长途/零担")
+        got = extract_collab_fields(
+            "客户名称张三 包装方式编织袋 货值 10万",
+            allowed_keys=allowed,
+        )
+        self.assertEqual(got.get("客户名称"), "张三")
+        self.assertEqual(got.get("包装方式"), "编织袋")
+        self.assertEqual(got.get("货值"), "10万")
+
+    def test_bare_pack_ignored(self):
+        allowed = land_collab_fields("国内运输拼车", "国内长途/零担")
+        self.assertEqual(extract_collab_fields("编织袋", allowed_keys=allowed), {})
+
+    def test_trade_terms_filtered_for_ltl(self):
+        allowed = land_collab_fields("国内运输拼车", "国内长途/零担")
+        got = extract_collab_fields("客户名称张三，贸易条款 FOB", allowed_keys=allowed)
+        self.assertEqual(got, {"客户名称": "张三"})
+
+    def test_sea_labeled_still_works(self):
+        got = extract_collab_fields("贸易条款：FOB")
+        self.assertEqual(got.get("贸易条款"), "FOB")
+
+
+if __name__ == "__main__":
+    unittest.main()

inquiry-agent/tests/test_land_copy.py

@@ -53,13 +53,12 @@ class LandCopyTest(unittest.TestCase):
         )
         self.assertIn("通关口岸（非必填）：", text)
 
-    def test_hk_shows_load(self):
+    def test_hk_lcl_shows_vehicle_not_load_field(self):
         text = copy.land_confirm_card(
             {
-                "运输类型": "中港",
+                "运输类型": "中港零担/集拼",
                 "线路类别": "东南亚",
                 "运输分类": "急件",
-                "中港装载方式": "零担/集拼",
                 "起运港": "广州",
                 "目的港": "香港",
                 "品名": "衣服",
@@ -67,25 +66,26 @@ class LandCopyTest(unittest.TestCase):
                 "报价日期": "2026-09-17",
             }
         )
-        self.assertIn("零担/集拼", text)
+        self.assertIn("运输类型（必填）：中港零担/集拼", text)
         self.assertIn("车型/数量（必填）：9米6/1", text)
-        self.assertNotIn("\n数量（必填）", text)
+        self.assertNotIn("中港装载方式", text)
+        self.assertIn("通关口岸（非必填）", text)
 
     def test_hk_full_truck_hides_vehicle(self):
         text = copy.land_confirm_card(
             {
-                "运输类型": "中港",
+                "运输类型": "中港整车",
                 "线路类别": "东南亚",
                 "运输分类": "急件",
-                "中港装载方式": "整车",
                 "起运港": "广州",
                 "目的港": "香港",
                 "品名": "衣服",
                 "报价日期": "2026-09-17",
             }
         )
-        self.assertIn("整车", text)
+        self.assertIn("运输类型（必填）：中港整车", text)
         self.assertNotIn("车型/数量", text)
+        self.assertNotIn("中港装载方式", text)
 
     def test_domestic_ftl_one_field(self):
         text = copy.land_confirm_card(
@@ -134,7 +134,7 @@ class LandCopyTest(unittest.TestCase):
         facts = {
             "运输类型": "国内运输拼车",
             "线路类别": "国内城配/拖车/打包",
-            "运输分类": "普件，常用线路/常规车型",
+            "运输分类": "普件(东南亚、中港常用线路/常规车型)",
             "起运港": "泉州市",
             "目的港": "南京市",
             "品名": "通信机柜及线缆配件",
@@ -201,6 +201,66 @@ class LandCopyTest(unittest.TestCase):
         self.assertIn("9米6/1", rows["货物"])
         self.assertNotIn("CBM", rows["货物"])
 
+    def test_land_brief_only_ltl_collab(self):
+        text = copy.sea_group_brief(
+            work_order_no="WO1",
+            facts={
+                "运输类型": "国内运输拼车",
+                "线路类别": "国内长途/零担",
+                "运输分类": "急件",
+                "起运港": "广州",
+                "目的港": "深圳",
+                "品名": "衣服",
+                "体积": "1",
+                "毛重": "1",
+                "报价日期": "2026-09-18",
+            },
+            quote=None,
+            collab_facts={"包装方式": "纸箱"},
+            sales_name="销",
+            product_names=["产"],
+            has_price=False,
+            transport_mode="陆运",
+        )
+        self.assertIn("包装方式：纸箱", text)
+        self.assertIn("客户名称：-", text)
+        self.assertIn("货值：-", text)
+        self.assertIn("是否为危险品：-", text)
+        self.assertNotIn("贸易条款", text)
+        self.assertNotIn("商品海关编码", text)
+        self.assertNotIn("海关编码", text)
+
+    def test_land_brief_ftl_longhaul_order(self):
+        text = copy.sea_group_brief(
+            work_order_no="WO2",
+            facts={
+                "运输类型": "国内运输整车",
+                "线路类别": "国内长途/零担",
+                "运输分类": "急件",
+                "起运港": "广州",
+                "目的港": "深圳",
+                "品名": "衣服",
+                "车型/数量": "9米/1",
+                "报价日期": "2026-09-18",
+            },
+            quote=None,
+            collab_facts={},
+            sales_name="销",
+            product_names=["产"],
+            has_price=False,
+            transport_mode="陆运",
+        )
+        collab = text.split("协同补充字段")[1]
+        self.assertLess(collab.find("客户名称"), collab.find("货值"))
+        self.assertLess(collab.find("货值"), collab.find("包装方式"))
+        self.assertNotIn("贸易条款", text)
+
+    def test_land_unrecognized_lists_keys(self):
+        keys = ("客户名称", "包装方式", "货值", "是否为危险品")
+        text = copy.group_fields_not_recognized(keys=keys, business_line="LAND")
+        self.assertIn("客户名称", text)
+        self.assertNotIn("货好时间", text)
+
 
 if __name__ == "__main__":
     unittest.main()

inquiry-agent/tests/test_land_field_validate.py

@@ -33,10 +33,9 @@ class LandFieldValidateTest(unittest.TestCase):
                 "品名": "衣服",
                 "体积": "10CBM",
                 "毛重": "50kg",
-                "中港装载方式": "零担/集拼",
             },
             business_line="LAND",
-            land_subtype="中港",
+            land_subtype="中港零担/集拼",
         )
         self.assertIn("车型/数量", got["missing"])
         self.assertNotIn("数量", got["missing"])
@@ -48,10 +47,9 @@ class LandFieldValidateTest(unittest.TestCase):
                 "起运港": "广州",
                 "目的港": "香港",
                 "品名": "衣服",
-                "中港装载方式": "整车",
             },
             business_line="LAND",
-            land_subtype="中港",
+            land_subtype="中港整车",
         )
         self.assertTrue(got["ok"])
         self.assertNotIn("车型/数量", got["missing"])

inquiry-agent/tests/test_land_group_collab.py

@@ -0,0 +1,305 @@
+"""
+陆运群内协同：摘要只列本单协同项、补字段过滤、预填。
+不连企微 / 真实 TMS / 8180。
+"""
+
+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.group_collab import handle_group_collab
+from agent.ledger.memory_ledger import MemoryLedger
+from agent.policy import inquiry_copy as copy
+from agent.policy.land_text_flow import LandTextInquiryFlow
+from agent.routing.group_intent import INTENT_FIELDS, INTENT_QUOTE
+
+
+class CountingLedger(MemoryLedger):
+    def __init__(self) -> None:
+        super().__init__()
+        self.last_created_no = ""
+
+    def create_ticket(self, **kwargs):
+        out = super().create_ticket(**kwargs)
+        self.last_created_no = str(out.get("work_order_no") or "")
+        return out
+
+
+class FakeGroupClient:
+    def __init__(self) -> None:
+        self.created: list[dict] = []
+        self.sent: list[dict] = []
+
+    def create_group(self, *, name: str, userids: list[str]) -> dict:
+        self.created.append({"name": name, "userids": list(userids)})
+        return {"ok": True, "chat_id": "wr_land_collab", "error": ""}
+
+    def add_group_members(self, *, chat_id: str, userids: list[str]) -> dict:
+        return {"ok": True}
+
+    def send_group(self, *, chat_id: str, content: str, mention_userids: list | None = None) -> dict:
+        self.sent.append({"chat_id": chat_id, "content": content, "kind": "text"})
+        return {"ok": True}
+
+    def send_group_file(self, *, chat_id: str, filename: str, filepath: str = "", file_b64: str = "") -> dict:
+        self.sent.append({"chat_id": chat_id, "filename": filename, "kind": "file"})
+        return {"ok": True}
+
+
+class LandGroupCollabTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.ledger = CountingLedger()
+        self.ledger.set_quote_templates(
+            [
+                {
+                    "templateId": "mem-default-land",
+                    "name": "陆运默认",
+                    "bizType": "陆运",
+                    "keywords": [],
+                    "isDefault": True,
+                }
+            ]
+        )
+        self.ledger.set_land_staff(
+            [
+                {
+                    "roleCode": "land",
+                    "status": "active",
+                    "wecomId": "p_dom",
+                    "name": "乙",
+                    "routes": "国内长途/零担,国内城配/拖车/打包,中港/中亚/中欧,东南亚",
+                }
+            ]
+        )
+        self.groups = FakeGroupClient()
+        self.flow = LandTextInquiryFlow(self.ledger, group_client=self.groups)
+        self.replies: list[str] = []
+
+    def _reply(self, text: str, extra=None) -> str:
+        self.replies.append(text)
+        return ""
+
+    def _group_texts(self) -> str:
+        return "\n".join(x.get("content") or "" for x in self.groups.sent if x.get("kind") == "text")
+
+    def _open(
+        self,
+        *,
+        option_text: str,
+        extra_facts: dict | None = None,
+        cargo: dict | None = None,
+    ) -> tuple[str, str]:
+        facts = {
+            "起运港": "广州",
+            "目的港": "深圳",
+            "品名": "衣服",
+            "体积": "100CBM",
+            "毛重": "50kg",
+        }
+        facts.update(cargo or {})
+        facts.update(extra_facts or {})
+        self.flow.on_text(
+            sender_id="u1",
+            text=option_text,
+            reply=self._reply,
+            injected_facts=facts,
+            injected_mode="LAND",
+        )
+        self.flow.on_text(sender_id="u1", text="确定", reply=self._reply)
+        self.replies.clear()
+        phase = self.flow.on_button(
+            sender_id="u1",
+            action=copy.BTN_PULL_COLLAB,
+            reply=self._reply,
+        )
+        self.assertEqual(phase, "collab_group")
+        wo = self.flow.session_of("u1").work_order_no
+        return wo, "wr_land_collab"
+
+    def _msg(self, *, sender: str, text: str, mid: str) -> InboundMessage:
+        return InboundMessage(
+            sender_id=sender,
+            message_id=mid,
+            content=text,
+            msg_type="text",
+            chat_id="wr_land_collab",
+            chat_type="group",
+        )
+
+    def test_ltl_brief_only_four_collab_and_prefill_pack(self) -> None:
+        self._open(
+            option_text="国内运输拼车 国内长途/零担 急件 包装方式纸箱",
+            extra_facts={"包装方式": "纸箱"},
+        )
+        brief = self._group_texts()
+        self.assertIn("包装方式：纸箱", brief)
+        self.assertIn("客户名称：-", brief)
+        self.assertIn("货值：-", brief)
+        self.assertIn("是否为危险品：-", brief)
+        self.assertNotIn("贸易条款", brief)
+        self.assertNotIn("商品海关编码", brief)
+        self.assertNotIn("海关编码", brief)
+
+    def test_domestic_ftl_longhaul_can_pull_and_brief_order(self) -> None:
+        """国内整车配长途：允许拉群；协同顺序是客户名称、货值、包装方式、是否为危险品。"""
+        self._open(
+            option_text="国内运输整车 国内长途/零担 急件",
+            cargo={"车型/数量": "9米/1"},
+        )
+        self.assertTrue(self.groups.created)
+        brief = self._group_texts()
+        collab_part = brief.split("协同补充字段")[1]
+        self.assertIn("客户名称：-", collab_part)
+        self.assertIn("货值：-", collab_part)
+        self.assertIn("包装方式：-", collab_part)
+        self.assertIn("是否为危险品：-", collab_part)
+        self.assertLess(collab_part.find("客户名称"), collab_part.find("货值"))
+        self.assertLess(collab_part.find("货值"), collab_part.find("包装方式"))
+        self.assertLess(collab_part.find("包装方式"), collab_part.find("是否为危险品"))
+        self.assertNotIn("贸易条款", brief)
+        self.assertNotIn("海关编码", brief)
+        # 车型/数量只在询价字段，不进协同。
+        self.assertIn("车型/数量：9米/1", brief.split("协同补充字段")[0])
+        self.assertNotIn("车型/数量", collab_part)
+
+    def test_cross_ftl_vehicle_stays_inquiry(self) -> None:
+        wo, _ = self._open(
+            option_text="跨境整车 东南亚 急件",
+            cargo={
+                "目的港": "河内",
+                "车型/数量": "9米/1",
+            },
+        )
+        brief = self._group_texts()
+        self.assertIn("车型/数量：9米/1", brief.split("协同补充字段")[0])
+        self.assertNotIn("车型/数量", brief.split("协同补充字段")[1])
+        self.groups.sent.clear()
+        phase = handle_group_collab(
+            self._msg(sender="u1", text="车型/数量 12米/1", mid="v1"),
+            flow=self.flow,
+            ledger=self.ledger,
+            injected_intent=INTENT_FIELDS,
+        )
+        self.assertEqual(phase, "collab_not_recognized")
+        ticket = self.ledger.get_ticket(work_order_no=wo)
+        self.assertEqual(ticket.facts.get("车型/数量"), "9米/1")
+        self.assertFalse(ticket.collab_facts.get("车型/数量"))
+
+    def test_hk_ftl_vehicle_is_collab(self) -> None:
+        wo, _ = self._open(
+            option_text="中港整车 中港/中亚/中欧 急件",
+            cargo={"目的港": "香港"},
+        )
+        brief = self._group_texts()
+        collab_part = brief.split("协同补充字段")[1]
+        self.assertIn("车型/数量：-", collab_part)
+        self.assertNotIn("车型/数量", brief.split("协同补充字段")[0])
+        self.groups.sent.clear()
+        phase = handle_group_collab(
+            self._msg(sender="u1", text="车型/数量 9米/1", mid="hk1"),
+            flow=self.flow,
+            ledger=self.ledger,
+            injected_intent=INTENT_FIELDS,
+        )
+        self.assertEqual(phase, "collab_field_saved")
+        ticket = self.ledger.get_ticket(work_order_no=wo)
+        self.assertEqual(ticket.collab_facts.get("车型/数量"), "9米/1")
+        self.assertFalse(ticket.facts.get("车型/数量"))
+        self.assertIn("车型/数量：9米/1", self._group_texts())
+
+    def test_ltl_rejects_trade_terms_only(self) -> None:
+        self._open(option_text="国内运输拼车 国内长途/零担 急件")
+        self.groups.sent.clear()
+        phase = handle_group_collab(
+            self._msg(sender="u1", text="贸易条款 FOB", mid="t1"),
+            flow=self.flow,
+            ledger=self.ledger,
+            injected_intent=INTENT_FIELDS,
+        )
+        self.assertEqual(phase, "collab_not_recognized")
+        text = self._group_texts()
+        self.assertIn("客户名称", text)
+        self.assertNotIn("货好时间", text)
+
+    def test_ltl_keeps_name_drops_trade_terms(self) -> None:
+        wo, _ = self._open(option_text="国内运输拼车 国内长途/零担 急件")
+        self.groups.sent.clear()
+        phase = handle_group_collab(
+            self._msg(sender="u1", text="客户名称张三，贸易条款 FOB", mid="t2"),
+            flow=self.flow,
+            ledger=self.ledger,
+            injected_intent=INTENT_FIELDS,
+        )
+        self.assertEqual(phase, "collab_field_saved")
+        ticket = self.ledger.get_ticket(work_order_no=wo)
+        self.assertEqual(ticket.collab_facts.get("客户名称"), "张三")
+        self.assertFalse(ticket.collab_facts.get("贸易条款"))
+        self.assertIn("客户名称：张三", self._group_texts())
+
+    def test_dg_and_value_and_bare_pack(self) -> None:
+        wo, _ = self._open(option_text="国内运输拼车 国内长途/零担 急件")
+        handle_group_collab(
+            self._msg(sender="u1", text="普货", mid="dg1"),
+            flow=self.flow,
+            ledger=self.ledger,
+            injected_intent=INTENT_FIELDS,
+        )
+        ticket = self.ledger.get_ticket(work_order_no=wo)
+        self.assertEqual(ticket.collab_facts.get("是否为危险品"), "否")
+        handle_group_collab(
+            self._msg(sender="u1", text="危险品", mid="dg2"),
+            flow=self.flow,
+            ledger=self.ledger,
+            injected_intent=INTENT_FIELDS,
+        )
+        ticket = self.ledger.get_ticket(work_order_no=wo)
+        self.assertEqual(ticket.collab_facts.get("是否为危险品"), "是")
+        handle_group_collab(
+            self._msg(sender="u1", text="货值 10万 包装方式编织袋", mid="v1"),
+            flow=self.flow,
+            ledger=self.ledger,
+            injected_intent=INTENT_FIELDS,
+        )
+        ticket = self.ledger.get_ticket(work_order_no=wo)
+        self.assertEqual(ticket.collab_facts.get("货值"), "10万")
+        self.assertEqual(ticket.collab_facts.get("包装方式"), "编织袋")
+        self.groups.sent.clear()
+        phase = handle_group_collab(
+            self._msg(sender="u1", text="编织袋", mid="v2"),
+            flow=self.flow,
+            ledger=self.ledger,
+            injected_intent=INTENT_FIELDS,
+        )
+        self.assertEqual(phase, "collab_not_recognized")
+
+    def test_product_can_quote_without_collab(self) -> None:
+        wo, _ = self._open(option_text="国内运输拼车 国内长途/零担 急件")
+        phase = handle_group_collab(
+            self._msg(sender="p_dom", text="运费 1000", mid="q1"),
+            flow=self.flow,
+            ledger=self.ledger,
+            injected_intent=INTENT_QUOTE,
+            injected_quote={"total": "1000", "source_label": "产品报价"},
+        )
+        self.assertIn(phase, {"product_quote_sent", "wait_tms_choice"})
+        ticket = self.ledger.get_ticket(work_order_no=wo)
+        self.assertTrue(ticket.quote)
+
+
+if __name__ == "__main__":
+    unittest.main()

inquiry-agent/tests/test_land_options.py

@@ -1,5 +1,5 @@
 """
-陆运选项联动单测：类型/线路/分类/中港装载，选错只纠正该项。
+陆运选项单测：三项各自必填、不联动；中港拆进运输类型；样例表跟类型/线路缺项走。
 
 不连企微、主账、TMS。
 """
@@ -18,12 +18,14 @@ from agent.schema.land_options import (
     harvest_land_options,
     looks_like_confirm,
     option_prompt,
+    option_prompt_needs_sample,
+    sample_pair_allowed,
     validate_land_options,
 )
 
 
 class LandOptionsTest(unittest.TestCase):
-    def test_domestic_ltl_rejects_sea_route(self):
+    def test_any_type_route_combo_ok(self):
         got = validate_land_options(
             {
                 "运输类型": "国内运输拼车",
@@ -31,28 +33,70 @@ class LandOptionsTest(unittest.TestCase):
                 "运输分类": "急件",
             }
         )
-        self.assertFalse(got["ok"])
-        msg = got["errors"][0]["message"]
-        self.assertIn("国内长途/零担", msg)
-        self.assertIn("国内城配/拖车/打包", msg)
+        self.assertTrue(got["ok"])
+        self.assertEqual(got["normalized"]["运输类型"], "国内运输拼车")
+        self.assertEqual(got["normalized"]["线路类别"], "东南亚")
+
+    def test_bare_hk_does_not_fill_type_or_route(self):
+        cur = harvest_land_options("中港 急件")
+        self.assertFalse(cur.get("运输类型"))
+        self.assertFalse(cur.get("线路类别"))
+        self.assertEqual(cur.get("运输分类"), "急件")
+
+    def test_hk_full_and_lcl_types(self):
+        self.assertEqual(
+            harvest_land_options("中港整车")["运输类型"], "中港整车"
+        )
+        self.assertEqual(
+            harvest_land_options("中港零担/集拼 东南亚")["运输类型"],
+            "中港零担/集拼",
+        )
+        self.assertEqual(
+            harvest_land_options("中港零担")["运输类型"], "中港零担/集拼"
+        )
+
+    def test_new_class_names(self):
+        cur = harvest_land_options(
+            "普件(东南亚、中港常用线路/常规车型)"
+        )
+        self.assertEqual(
+            cur["运输分类"], "普件(东南亚、中港常用线路/常规车型)"
+        )
+        self.assertFalse(
+            harvest_land_options("普件，常用线路/常规车型").get("运输分类")
+        )
+
+    def test_no_hk_load_field(self):
+        got = validate_land_options(
+            {
+                "运输类型": "中港整车",
+                "线路类别": "东南亚",
+                "运输分类": "急件",
+            }
+        )
+        self.assertTrue(got["ok"])
+        self.assertNotIn("中港装载方式", got["missing"])
+        self.assertFalse(got["normalized"].get("中港装载方式"))
 
     def test_followup_route_class_keeps_type(self):
         cur = harvest_land_options(
-            "国内城配/拖车/打包；普件，常用线路/常规车型",
+            "国内城配/拖车/打包；普件(东南亚、中港常用线路/常规车型)",
             current={"运输类型": "国内运输拼车"},
         )
         self.assertEqual(cur["运输类型"], "国内运输拼车")
         self.assertEqual(cur["线路类别"], "国内城配/拖车/打包")
-        self.assertEqual(cur["运输分类"], "普件，常用线路/常规车型")
+        self.assertEqual(
+            cur["运输分类"], "普件(东南亚、中港常用线路/常规车型)"
+        )
 
     def test_model_land_does_not_wipe_type(self):
         dropped = harvest_land_options(
-            "国内城配/拖车/打包；普件，常用线路/常规车型",
+            "国内城配/拖车/打包；急件",
             current={"运输类型": "陆运"},
         )
         self.assertFalse(dropped.get("运输类型"))
         cur = harvest_land_options(
-            "国内城配/拖车/打包；普件，常用线路/常规车型",
+            "国内城配/拖车/打包；急件",
             current={"运输类型": "国内运输拼车"},
         )
         junk = harvest_land_options("陆运", current=cur)
@@ -64,36 +108,66 @@ class LandOptionsTest(unittest.TestCase):
 
     def test_full_prompt_when_empty(self):
         text = option_prompt(current={})
-        self.assertIn("运输类型", text)
-        self.assertIn("线路类别", text)
-        self.assertIn("运输分类", text)
-        self.assertIn("国内运输拼车", text)
-        self.assertIn("整车", text)
-        self.assertNotIn("请补充陆运类型：", text)
+        self.assertIn("请一次选择：运输类型、线路类别、运输分类", text)
+        self.assertIn("运输类型（必填）：可选择回复，", text)
+        self.assertIn("中港整车", text)
+        self.assertIn("中港零担/集拼", text)
+        self.assertIn("线路类别（必填）：可选择回复，", text)
+        self.assertIn("国内长途/零担", text)
+        self.assertIn("1.普件(东南亚、中港常用线路/常规车型)", text)
+        self.assertIn("填写样例见图", text)
+        self.assertNotIn("国内运输整车 + 国内城配/拖车/打包", text)
+        self.assertNotIn("┌", text)
+        self.assertNotIn("| 国内运输整车 |", text)
+        self.assertNotIn("仅供参考", text)
+        self.assertNotIn("中港装载方式", text)
+        self.assertNotIn("跟运输类型联动", text)
 
-    def test_copy_alias_lists_all_three(self):
+    def test_copy_alias_lists_new_types(self):
         from agent.policy import inquiry_copy as copy
 
         text = copy.ASK_LAND_SUBTYPE
-        self.assertIn("运输类型", text)
-        self.assertIn("线路类别", text)
-        self.assertIn("运输分类", text)
+        self.assertIn("中港整车", text)
+        self.assertIn("填写样例", text)
         self.assertNotIn("请补充陆运类型：国内运输拼车", text)
 
-    def test_narrow_prompt_when_type_known(self):
+    def test_sample_table_when_route_missing(self):
         text = option_prompt(current={"运输类型": "国内运输拼车"})
-        self.assertIn("国内长途/零担", text)
-        self.assertNotIn("请选择运输类型", text)
+        self.assertIn("线路类别（必填）", text)
+        self.assertIn("填写样例见图", text)
+        self.assertTrue(option_prompt_needs_sample(current={"运输类型": "国内运输拼车"}))
+        self.assertNotIn("国内运输拼车 + 国内长途/零担", text)
+        self.assertNotIn("请一次选择：运输类型", text)
+        self.assertNotIn("运输类型（必填）", text)
 
-    def test_hk_needs_load(self):
-        got = validate_land_options(
-            {
-                "运输类型": "中港",
+    def test_sample_pairs_go_to_image_not_text(self):
+        from agent.policy import inquiry_copy as copy
+
+        text = option_prompt(current={})
+        self.assertIn("填写样例见图", text)
+        self.assertNotIn("国内运输拼车 + 国内长途/零担", text)
+        extra = copy.land_sample_image_payload()
+        self.assertEqual(extra.get("msgtype"), "image")
+        self.assertTrue(os.path.isfile(str(extra.get("filepath") or "")))
+        self.assertGreater(os.path.getsize(str(extra["filepath"])), 1000)
+
+    def test_no_sample_when_only_class_missing(self):
+        text = option_prompt(
+            current={
+                "运输类型": "国内运输拼车",
                 "线路类别": "东南亚",
-                "运输分类": "急件",
             }
         )
-        self.assertIn("中港装载方式", got["missing"])
+        self.assertIn("运输分类（必填）", text)
+        self.assertNotIn("填写样例", text)
+
+    def test_sample_pair_gate(self):
+        self.assertTrue(sample_pair_allowed("国内运输拼车", "国内长途/零担"))
+        self.assertTrue(sample_pair_allowed("国内运输整车", "国内城配/拖车/打包"))
+        self.assertTrue(sample_pair_allowed("国内运输整车", "国内长途/零担"))
+        self.assertTrue(sample_pair_allowed("中港整车", "中港/中亚/中欧"))
+        self.assertFalse(sample_pair_allowed("国内运输拼车", "东南亚"))
+        self.assertFalse(sample_pair_allowed("中港整车", "东南亚"))
 
     def test_confirm_intent(self):
         self.assertTrue(looks_like_confirm("确认"))

inquiry-agent/tests/test_land_text_inquiry.py

@@ -47,9 +47,11 @@ class LandTextInquiryFrontTests(unittest.TestCase):
         self.ledger = CountingLedger()
         self.flow = LandTextInquiryFlow(self.ledger)
         self.replies: list[str] = []
+        self.extras: list[dict] = []
 
     def _reply(self, text: str, extra=None) -> str:
         self.replies.append(text)
+        self.extras.append(dict(extra or {}))
         return ""
 
     def _joined(self) -> str:
@@ -69,7 +71,12 @@ class LandTextInquiryFrontTests(unittest.TestCase):
         self.assertIn("国内运输拼车", text)
         self.assertIn("线路类别", text)
         self.assertIn("运输分类", text)
+        self.assertIn("中港整车", text)
+        self.assertIn("填写样例见图", text)
         self.assertNotIn("请补充陆运类型：", text)
+        img = [x for x in self.extras if x.get("msgtype") == "image"]
+        self.assertEqual(len(img), 1)
+        self.assertTrue(os.path.isfile(str(img[0].get("filepath") or "")))
 
     def test_known_type_does_not_reask_type(self) -> None:
         phase = self.flow.on_text(
@@ -82,8 +89,10 @@ class LandTextInquiryFrontTests(unittest.TestCase):
         self.assertEqual(phase, "need_land_options")
         self.assertNotIn("请选择运输类型", self._joined())
         self.assertIn("国内长途/零担", self._joined())
+        self.assertIn("填写样例见图", self._joined())
+        self.assertTrue(any(x.get("msgtype") == "image" for x in self.extras))
 
-    def test_wrong_route_keeps_type(self) -> None:
+    def test_unlinked_route_still_keeps_type(self) -> None:
         phase = self.flow.on_text(
             sender_id="u1",
             text="国内运输拼车，线路东南亚",
@@ -92,14 +101,16 @@ class LandTextInquiryFrontTests(unittest.TestCase):
             injected_mode="LAND",
         )
         self.assertEqual(phase, "need_land_options")
-        self.assertIn("国内长途/零担", self._joined())
+        self.assertIn("运输分类", self._joined())
+        self.assertNotIn("填写样例", self._joined())
         sess = self.flow.session_of("u1")
         self.assertEqual(sess.facts.get("运输类型"), "国内运输拼车")
+        self.assertEqual(sess.facts.get("线路类别"), "东南亚")
 
     def test_hk_lcl_asks_vehicle(self) -> None:
         phase = self.flow.on_text(
             sender_id="u1",
-            text="中港 东南亚 急件 零担",
+            text="中港零担/集拼 东南亚 急件",
             reply=self._reply,
             injected_facts={
                 "起运港": "广州",
@@ -119,7 +130,7 @@ class LandTextInquiryFrontTests(unittest.TestCase):
     def test_hk_full_truck_skips_vehicle(self) -> None:
         phase = self.flow.on_text(
             sender_id="u1",
-            text="中港 东南亚 急件 整车",
+            text="中港整车 东南亚 急件",
             reply=self._reply,
             injected_facts={
                 "起运港": "广州",
@@ -403,7 +414,7 @@ class LandTextConfirmTests(unittest.TestCase):
         self.replies.clear()
         phase = self.flow.on_text(
             sender_id="u1",
-            text="国内城配/拖车/打包；普件，常用线路/常规车型",
+            text="国内城配/拖车/打包；普件(东南亚、中港常用线路/常规车型)",
             reply=self._reply,
             injected_facts={"品名": "通信机柜及线缆配件", "运输类型": "陆运"},
             injected_mode="LAND",
@@ -413,7 +424,7 @@ class LandTextConfirmTests(unittest.TestCase):
         self.assertNotIn("请一次选择：运输类型", text)
         self.assertIn("国内运输拼车", text)
         self.assertIn("国内城配/拖车/打包", text)
-        self.assertIn("普件，常用线路/常规车型", text)
+        self.assertIn("普件(东南亚、中港常用线路/常规车型)", text)
 
 
 class FakeGroupClient:
@@ -513,6 +524,36 @@ class LandTextPullTests(unittest.TestCase):
         self.assertEqual(self.groups.created, [])
         self.assertEqual(phase, "wait_collab")
 
+    def test_pull_rejects_unlisted_pair(self) -> None:
+        self.ledger.set_land_staff(
+            [
+                {"roleCode": "land", "status": "active", "wecomId": "p_sea", "name": "甲", "routes": "东南亚"},
+            ]
+        )
+        self.flow.on_text(
+            sender_id="u1",
+            text="国内运输拼车 东南亚 急件",
+            reply=self._reply,
+            injected_facts={
+                "起运港": "广州",
+                "目的港": "河内",
+                "品名": "衣服",
+                "体积": "100CBM",
+                "毛重": "50kg",
+            },
+            injected_mode="LAND",
+        )
+        self.flow.on_text(sender_id="u1", text="确定", reply=self._reply)
+        self.replies.clear()
+        phase = self.flow.on_button(
+            sender_id="u1",
+            action=copy.BTN_PULL_COLLAB,
+            reply=self._reply,
+        )
+        self.assertEqual(self.groups.created, [])
+        self.assertIn("拉群失败，请排查填写内容，并重新发起工单。", "\n".join(self.replies))
+        self.assertNotEqual(phase, "collab_group")
+
     def test_land_role_is_product_in_group(self) -> None:
         from agent.policy.sea_group_ops import is_product
 

inquiry-agent/tests/test_sea_text_inquiry.py

@@ -452,6 +452,39 @@ class SeaSkipAndPullTests(unittest.TestCase):
         )
         self.assertEqual(again, "card_discard")
 
+    def test_pull_prefills_spoken_no_battery(self) -> None:
+        self.ledger.set_sea_staff(
+            [
+                {
+                    "name": "张产品",
+                    "wecomId": "prod1",
+                    "roleCode": "sea",
+                    "routes": "海运·全部线路",
+                    "status": "active",
+                }
+            ]
+        )
+        facts = dict(COMPLETE_SEA)
+        facts["是否含电"] = "否"
+        self.flow.on_text(
+            sender_id="s1",
+            text="海运不含电",
+            reply=self._reply,
+            injected_facts=facts,
+            injected_mode="SEA",
+        )
+        private = "\n".join(self.replies)
+        self.assertNotIn("是否含电", private)
+        self.flow.on_button(
+            sender_id="s1",
+            action=copy.BTN_PULL_COLLAB,
+            reply=self._reply,
+        )
+        group_text = "\n".join(
+            x.get("content") or "" for x in self.groups.sent if x.get("kind") == "text"
+        )
+        self.assertIn("是否含电：否", group_text)
+
     def test_pull_sends_sheet_when_client_not_injected(self) -> None:
         """测服正式路径不注入 group_client，拉群仍要发出填好的模板。"""
         from unittest.mock import patch

inquiry-agent/tests/test_tms_land_query.py

@@ -18,10 +18,9 @@ from agent.schema.tms_land_query import assemble_land_query
 class TmsLandQueryTest(unittest.TestCase):
     def test_outbound_omits_class_and_hk_load(self):
         facts = {
-            "运输类型": "中港",
+            "运输类型": "中港零担/集拼",
             "线路类别": "东南亚",
             "运输分类": "急件",
-            "中港装载方式": "零担/集拼",
             "起运港": "广州",
             "目的港": "香港",
             "品名": "衣服",
@@ -44,9 +43,8 @@ class TmsLandQueryTest(unittest.TestCase):
 
     def test_hk_full_truck_omits_vehicle(self):
         facts = {
-            "运输类型": "中港",
+            "运输类型": "中港整车",
             "线路类别": "东南亚",
-            "中港装载方式": "整车",
             "起运港": "广州",
             "目的港": "香港",
             "品名": "衣服",