落地多段联运私聊：齐了才建单拉群，附件只收格子值，核对改正不再整单重开。

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

inquiry-agent/agent/handlers/air_group_collab.py

@@ -116,6 +116,19 @@ def handle_air_group(
     mark_air_text(chat_id=chat_id, sender_id=message.sender_id, text=text)
     wo = copy.extract_work_order_no(text)
     if wo:
+        from agent.policy.multi_group_ops import handle_multi_inbound, ticket_is_multi
+
+        getter = getattr(engine.ledger, "get_ticket", None)
+        raw_ticket = getter(work_order_no=wo) if callable(getter) else None
+        if raw_ticket is not None and ticket_is_multi(raw_ticket):
+            return handle_multi_inbound(
+                flow=engine,
+                ticket=raw_ticket,
+                chat_id=chat_id,
+                sender_id=message.sender_id,
+                text=text,
+                injected_quote=injected_quote,
+            )
         from agent.policy.handoff_human import try_commit_air_group_handoff
 
         handed = try_commit_air_group_handoff(engine, chat_id=chat_id, work_order_no=wo)

inquiry-agent/agent/handlers/attachment_inquiry.py

@@ -30,6 +30,30 @@ VisionFn = Callable[..., dict[str, Any]]
 DownloadFn = Callable[[str], tuple[bytes, str]]
 PutFn = Callable[..., dict[str, Any]]
 
+
+def prefer_extracted_oral(raw_text: str, organized: str) -> str:
+    """
+    千问把报价单整理成空、几个字或几十个字时，改用抽出的单元格原文。
+
+    测服复现：陆运 997 字被压成 48 字、海运 1325 字被压成 52 字，
+    多段补问就只剩货量/运输分类，起运港目的港整段待补充。
+    整理结果够长、没有被大幅压短时仍用整理后的，方便人读。
+    """
+    raw = (raw_text or "").strip()
+    org = (organized or "").strip()
+    if not raw:
+        return org
+    if not org:
+        return raw
+    # 报价单原文很长、整理后只剩几分之一：用原文。997→48、1325→52 都走这里。
+    if len(raw) >= 200 and len(org) * 8 < len(raw):
+        return raw
+    if len(org) < 8 and len(raw) >= 40:
+        return raw
+    if len(org) < 40 and len(org) * 5 < len(raw):
+        return raw
+    return org
+
 _STASH: dict[str, bytes] = {}
 _STASH_LOCK = threading.Lock()
 _PRECEDE_NOISE = (
@@ -68,10 +92,17 @@ def _stash_get(key: str) -> bytes:
 
 
 def drop_all_private_bookmarks(sender_id: str) -> None:
-    """三线当前书签都丢掉；已出号仍能按工单号找回。"""
+    """
+    海运/空运/陆运/多段当前书签都丢掉；已出号仍能按工单号找回。
+
+    附件发卡或认完必须连多段一起清：否则旧补问草稿会把新附件口播当成补字段，
+    海运只剩货量/运输分类，陆运再问一遍运输类型。
+    """
     from agent.handlers.image_inquiry import drop_all_private_bookmarks as _drop
+    from agent.policy.multi_text_flow import get_multi_text_flow
 
     _drop(sender_id)
+    get_multi_text_flow().drop_current_bookmark(sender_id)
 
 
 def is_active_attachment_wait(sender_id: str, *, now: Optional[float] = None) -> bool:
@@ -404,6 +435,8 @@ def finish_attach_wave(
         media=dict(seed_message.media) if seed_message else {},
         raw=dict(seed_message.raw) if seed_message else {},
     )
+    # 认完附件一律新开：不能粘在发卡前留下的多段/单段补问草稿上。
+    drop_all_private_bookmarks(sender_id)
     phase = handle_text_inquiry(inbound, store=_StoreProxy())
     if reason and not prefix_used["v"]:
         msg_store.enqueue_outbound(
@@ -612,9 +645,18 @@ def _recognize_file(
                 allow_network=bool(settings.llm_allow_network and settings.llm_data_usage_confirmed),
             )
         if packed.get("ok") and str(packed.get("text") or "").strip():
-            organized = str(packed.get("text") or "")
+            organized = prefer_extracted_oral(raw_text, str(packed.get("text") or ""))
         elif not packed.get("ok"):
             org_err = str(packed.get("error") or org_err or "attachment_text_failed")
+    logger.info(
+        "attachment 认完 sender=%s idx=%s name=%s raw_len=%s oral_len=%s err=%s",
+        sender_id,
+        index,
+        filename,
+        len(raw_text),
+        len((organized or "").strip()),
+        org_err or "-",
+    )
     book.complete_material(
         sender_id,
         index,

inquiry-agent/agent/handlers/card_action.py

@@ -20,7 +20,17 @@ logger = logging.getLogger(__name__)
 
 
 def _pick_card_flow(sender_id: str, payload: dict | None):
-    """点卡：海运/陆运书签能找回工单就走对应流程，否则空运。"""
+    """
+    点卡：先认多段联运，再认海运/陆运，最后才空运。
+
+    多段工单主账停点常被收成 tms_miss。若先走空运，「拉产品进群」会被丢掉、销售点了没回声。
+    """
+    from agent.policy.multi_text_flow import get_multi_text_flow
+
+    multi = get_multi_text_flow()
+    multi_sess = multi.session_for_action(sender_id, payload)
+    if multi_sess and (multi_sess.business_line or "").upper() == "MULTI":
+        return multi
     sea = get_sea_text_flow()
     sess = sea.session_for_action(sender_id, payload)
     if sess and (sess.business_line or "").upper() == "SEA":

inquiry-agent/agent/handlers/group_collab.py

@@ -95,6 +95,30 @@ def handle_group_inbound(
             from agent.policy.land_text_flow import get_land_text_flow
 
             engine = get_land_text_flow()
+    view0 = ticket_view(ticket)
+    line = str(
+        view0.get("business_line")
+        or view0.get("businessLine")
+        or getattr(ticket, "business_line", "")
+        or ""
+    ).upper()
+    if line == "MULTI":
+        from agent.policy.multi_group_ops import handle_multi_inbound, sender_mode
+
+        if not (
+            _sender_allowed(ticket, message.sender_id)
+            or sender_mode(engine, message.sender_id)
+        ):
+            logger.info("group_inbound 多段忽略非成员 sender=%s", message.sender_id)
+            return "group_ignore"
+        return handle_multi_inbound(
+            flow=engine,
+            ticket=ticket,
+            chat_id=chat_id,
+            sender_id=message.sender_id,
+            text=message.content or "",
+            injected_quote=injected_quote,
+        )
     ticket = hydrate_ticket(engine, ticket)
     if not _sender_allowed(ticket, message.sender_id):
         logger.info("group_inbound 忽略非成员 sender=%s chat=%s", message.sender_id, chat_id)

inquiry-agent/agent/handlers/text_inquiry.py

@@ -1,5 +1,5 @@
 """
-意图：普通文字询价（空运 / 海运闭环）。
+意图：普通文字询价（空运 / 海运 / 陆运 / 多段联运）。
 
 本文件职责：入站文字 → 流程编排；回复只写入 outbox。
 禁止：在本文件抽字段语义（交给 extract）；禁止直连 TMS；禁止回调线程外的额外阻塞。
@@ -45,7 +45,7 @@ def _drop_other_mode_drafts(*, owner: str, sender_id: str) -> None:
     """
     kind = (owner or "").upper()
     sid = (sender_id or "").strip()
-    if not sid or kind not in {"SEA", "AIR", "LAND"}:
+    if not sid or kind not in {"SEA", "AIR", "LAND", "MULTI"}:
         return
     land = get_land_text_flow()
     air = get_air_text_flow()
@@ -112,6 +112,19 @@ def _pick_text_flow(*, sender_id: str, text: str, injected_mode: str = "", prefe
     from agent.policy.quote_adjust_ops import quoted_adjust_flow
     from agent.policy.handoff_human import is_handoff_frozen
 
+    from agent.policy.multi_segments import classify_multi_text
+    from agent.policy.multi_text_flow import get_multi_text_flow
+
+    multi = get_multi_text_flow()
+    # 多段核对上的「确定」优先于陆运旧核对，避免建出海运/陆运单段工单。
+    if looks_like_confirm(text):
+        multi_hit = multi.session_of(sender_id)
+        if (
+            multi_hit
+            and (multi_hit.business_line or "").upper() == "MULTI"
+            and (multi_hit.phase or "") == "wait_confirm"
+        ):
+            return multi
     land = get_land_text_flow()
     if looks_like_confirm(text):
         restored = land.restore_confirm_session(sender_id)
@@ -123,14 +136,42 @@ def _pick_text_flow(*, sender_id: str, text: str, injected_mode: str = "", prefe
     sea = get_sea_text_flow()
     air = get_air_text_flow()
     spoken = (injected_mode or detect_transport_mode(text) or "").upper()
+    multi_kind = classify_multi_text(text) if not injected_mode else "single"
+    if multi_kind in {"multi", "same_mode"}:
+        _drop_other_mode_drafts(owner="MULTI", sender_id=sender_id)
+        return multi
     # 本句点名运输方式：盖掉其它方式的旧补问，避免海运补字段走陆运样例图。
+    # 多段正在补问时例外：销售会写「段2：运输方式：空运」，不能把多段草稿丢掉改走单段。
+    # 已出号等拉群不算补问：再发陆运附件要走陆运，不能粘在旧工单的手动拉群提示上。
     if spoken in {"SEA", "AIR", "LAND"}:
+        multi_draft = multi.session_of(sender_id)
+        if (
+            multi_draft
+            and (multi_draft.business_line or "").upper() == "MULTI"
+            and (multi_draft.phase or "") == "clarify"
+        ):
+            return multi
         _drop_other_mode_drafts(owner=spoken, sender_id=sender_id)
+        stale_handoff = (multi_draft.phase or "") in {
+            "wait_collab",
+            "wait_manual_group",
+            "collab_group",
+        } if multi_draft else False
+        if _front_bookmark(multi_draft) or stale_handoff:
+            multi.drop_current_bookmark(sender_id)
         if spoken == "SEA":
             return sea
         if spoken == "LAND":
             return land
         return air
+    multi_sess = multi.session_of(sender_id)
+    if (
+        multi_sess
+        and (multi_sess.business_line or "").upper() == "MULTI"
+        and (multi_sess.phase or "") in {"clarify", "wait_confirm"}
+        and not (looks_like_confirm(text) and (multi_sess.phase or "") != "wait_confirm")
+    ):
+        return multi
     sea_sess = sea.session_of(sender_id)
     land_sess = land.session_of(sender_id)
     air_sess = air.session_of(sender_id)

inquiry-agent/agent/ledger/memory_ledger.py

@@ -457,6 +457,9 @@ class MemoryLedger:
                 wait_phase = "wait_collab"
             else:
                 wait_phase = "wait_adopt"
+        elif (ticket.business_line or "").upper() == "MULTI":
+            # 多段建单后仍是询价中；已建群就跟进报价，否则等销售点拉群。
+            wait_phase = "collab_group" if (ticket.collab_chat_id or "").strip() else "wait_collab"
         else:
             wait_phase = "tms_miss"
         return {
@@ -466,6 +469,10 @@ class MemoryLedger:
             "work_order_no": ticket.work_order_no,
             "status": status,
             "business_line": ticket.business_line,
+            "sales_wecom_id": ticket.sales_wecom_id,
+            "salesWecomId": ticket.sales_wecom_id,
+            "sales_name": str(self._staff_by_wecom.get(ticket.sales_wecom_id, {}).get("name") or ticket.sales_wecom_id or ""),
+            "salesName": str(self._staff_by_wecom.get(ticket.sales_wecom_id, {}).get("name") or ticket.sales_wecom_id or ""),
             "facts": dict(ticket.facts),
             "quote": quote,
             "wait_phase": wait_phase,

inquiry-agent/agent/llm/extract_text.py

@@ -23,13 +23,13 @@ _MODE_WORDS = (
 _LABEL_LINE = re.compile(
     r"^(起运地|起运港|目的地|目的港|品名|件数|毛重|体积|重量|重量\(KG\)|体积\(CBM\)|"
     r"包装方式|包装类型|货量|货物数量|整柜或拼柜|箱型箱量|报价日期|运输分类|车型/数量|车型数量|"
-    r"贸易条款|货好时间|货源地|提货地址|收货地址|商品海关编码|海关编码|HS编码|HS|是否含油|是否含电|是否含磁|"
+    r"贸易条款|货好时间|货源地|提货地址|收货地址|送货地址|商品海关编码|海关编码|HS编码|HS|是否含油|是否含电|是否含磁|"
     r"客户名称|货值|是否为危险品|通关口岸（非必填）|通关口岸)\s*[:：]\s*(.+)$"
 )
 _LABEL_INLINE = re.compile(
     r"(起运地|起运港|目的地|目的港|品名|件数|毛重|体积|重量|重量\(KG\)|体积\(CBM\)|"
     r"包装方式|包装类型|货量|货物数量|整柜或拼柜|箱型箱量|报价日期|运输分类|车型/数量|车型数量|"
-    r"贸易条款|货好时间|货源地|提货地址|收货地址|商品海关编码|海关编码|HS编码|HS|是否含油|是否含电|是否含磁|"
+    r"贸易条款|货好时间|货源地|提货地址|收货地址|送货地址|商品海关编码|海关编码|HS编码|HS|是否含油|是否含电|是否含磁|"
     r"客户名称|货值|是否为危险品|通关口岸（非必填）|通关口岸)\s*[:：]\s*(\S+)"
 )
 # 陆运核对卡非必填：销售可能写「通关口岸皇岗」或照抄「通关口岸（非必填）：皇岗」。长词优先。
@@ -46,9 +46,58 @@ _OPTIONAL_INQUIRY_LABELS = (
     "货好时间",
     "提货地址",
     "收货地址",
+    "送货地址",
     "贸易条款",
     "起运港",
 )
+# 图上常见「4. 希望送到…」：编号项不是询价字段，值收到这里为止。
+_NUMBERED_ITEM = re.compile(r"(?m)^\s*\d+[\.、．]\s+\S")
+# 报价单表头和格子在同一行：只洗这些询价键，不碰提货/收货地址。
+_SHEET_CLEAN_KEYS = frozenset(
+    {
+        "起运地",
+        "起运港",
+        "目的地",
+        "目的港",
+        "品名",
+        "件数",
+        "毛重",
+        "体积",
+        "货量",
+        "货物数量",
+        "整柜或拼柜",
+        "箱型箱量",
+        "贸易条款",
+        "运输分类",
+        "报价日期",
+        "包装方式",
+        "包装类型",
+    }
+)
+_SHEET_STOP = (
+    "要求运输时效",
+    "Requested Transit",
+    "费用汇总",
+    "POD Local Charges",
+    "提货地址",
+    "Place of Receipt",
+    "送货地址",
+    "派送地址",
+    "Place of Delivery",
+    "要求货物抵达",
+    "Requested Arrival",
+)
+_SHEET_EN_ALIAS = re.compile(r"^[/／][A-Za-z][A-Za-z ]*")
+_SHEET_REQUIRED_HINT = re.compile(r"（[^）]*必填[^）]*）")
+_SHEET_REQUIRED_HINT_EN = re.compile(
+    r"\([^)]*(?:必填|Packing List|Quantity)[^)]*\)",
+    re.I,
+)
+_SHEET_HEADER_NOISE = re.compile(
+    r"(费用汇总|提货地址|送货地址|派送地址|要求货物抵达|要求运输时效|"
+    r"Local Charges|Place of Receipt|Place of Delivery|Requested Arrival|Requested Transit)",
+    re.I,
+)
 # 群里协同标签，冒号可省略：「客户名称张三」「货值 10万」。长词优先。
 _COLLAB_LABELS = (
     "是否为危险品",
@@ -86,18 +135,25 @@ def detect_transport_mode(text: str) -> str:
 
 
 def parse_labeled_facts(text: str) -> dict[str, str]:
-    """只采集「字段名：值」，整行优先；同一句里的标签也收。不猜港口。"""
+    """
+    只采集「字段名：值」，整行优先；同一句里的标签也收。不猜港口。
+
+    报价单常把表头和格子写在一行（起运港/POL（必填）  纽约港），
+    冒号对不上时再按表头切一刀，值只留格子。
+    """
     facts: dict[str, str] = {}
     raw = text or ""
     for line in raw.splitlines():
         m = _LABEL_LINE.match(line.strip())
         if m:
-            facts[m.group(1)] = m.group(2).strip()
-    if facts:
-        return facts
-    for m in _LABEL_INLINE.finditer(raw):
-        facts[m.group(1)] = m.group(2).strip()
-    return facts
+            facts[m.group(1)] = clip_sheet_field_value(m.group(2).strip())
+    if not facts:
+        for m in _LABEL_INLINE.finditer(raw):
+            facts[m.group(1)] = clip_sheet_field_value(m.group(2).strip())
+    for key, val in harvest_sheet_inquiry_cells(raw).items():
+        if val and not str(facts.get(key) or "").strip():
+            facts[key] = val
+    return {key: val for key, val in facts.items() if val}
 
 
 def _merge_spoken_collab(text: str, facts: dict[str, str]) -> dict[str, str]:
@@ -140,6 +196,10 @@ def extract_inquiry_snapshot(
         facts = harvest_oral_measures(text, normalize_facts(dict(injected_facts)))
         facts = _merge_spoken_collab(text, facts)
         facts = _merge_optional_inquiry_labels(text, facts)
+        for key in ("货好时间", "提货地址", "收货地址", "货源地", "贸易条款"):
+            if str(facts.get(key) or "").strip():
+                facts[key] = clip_inquiry_value(str(facts[key]))
+        facts = _clip_sheet_inquiry_facts(facts)
         subtype = str(facts.get("运输类型") or "").strip()
         return {
             "business_line": mode,
@@ -179,6 +239,10 @@ def extract_inquiry_snapshot(
     facts = harvest_oral_measures(text, normalize_facts(facts))
     facts = _merge_spoken_collab(text, facts)
     facts = _merge_optional_inquiry_labels(text, facts)
+    for key in ("货好时间", "提货地址", "收货地址", "货源地", "贸易条款"):
+        if str(facts.get(key) or "").strip():
+            facts[key] = clip_inquiry_value(str(facts[key]))
+    facts = _clip_sheet_inquiry_facts(facts)
     land_subtype = str((b_payload or {}).get("land_subtype") or facts.get("运输类型") or "").strip()
     if land_subtype and not str(facts.get("运输类型") or "").strip():
         facts["运输类型"] = land_subtype
@@ -213,15 +277,149 @@ def harvest_inquiry_optional_labels(text: str) -> dict[str, str]:
     return _harvest_labels(text, _OPTIONAL_INQUIRY_LABELS)
 
 
+def clip_inquiry_value(raw: str) -> str:
+    """
+    字段值收到下一个编号项为止。
+
+    图上常在货好时间后面接着「4. 希望送到…」「6. 送货地址」。
+    那些不是询价字段，不能进补问。地址里的「602、603」没有点号后空格，会留下。
+    """
+    text = (raw or "").strip()
+    if not text:
+        return ""
+    hit = _NUMBERED_ITEM.search(text)
+    if hit and hit.start() > 0:
+        text = text[: hit.start()]
+    text = re.sub(r"(?:\n|\r)+\s*\d+[\.、．]\s*$", "", text)
+    return text.strip()
+
+
+def clip_sheet_field_value(raw: str) -> str:
+    """
+    报价单格子：去掉 /POL、（必填说明）、Terms 和同一行下一列表头，只留紧挨的值。
+
+    测服手工海运单：起运港/POL（必填）  纽约港  目的港…  巴生西港  费用汇总 → 纽约港 / 巴生西港。
+    取第一格有效值，不要最后一格（最后一格常是费用汇总、提货地址）。
+    地址字段不要走这里。
+    """
+    text = clip_inquiry_value(raw)
+    if not text:
+        return ""
+    text = _SHEET_EN_ALIAS.sub("", text)
+    text = _SHEET_REQUIRED_HINT.sub("", text)
+    text = _SHEET_REQUIRED_HINT_EN.sub("", text)
+    text = re.sub(r"^\((?:KG|CBM)\)\s*[:：]?\s*", "", text, flags=re.I)
+    if "必填" in text:
+        text = re.sub(r"^[^）\n]*必填[^）\n]*）\s*", "", text)
+    text = re.sub(r"^[\s:：]+", "", text)
+    for label in _SHEET_CLEAN_KEYS:
+        idx = 0
+        while True:
+            pos = text.find(label, idx)
+            if pos < 0:
+                break
+            if pos > 0 and _label_is_token(text, pos, label):
+                text = text[:pos]
+                break
+            idx = pos + len(label)
+    for stop in _SHEET_STOP:
+        pos = text.find(stop)
+        if pos >= 0:
+            text = text[:pos]
+    text = text.strip().strip("：:/*")
+    parts = [p.strip().strip("：:/*") for p in re.split(r"[\t]+|\s{2,}", text) if p.strip().strip("：:/*")]
+    parts = [p for p in parts if p and not _SHEET_HEADER_NOISE.search(p)]
+    if not parts:
+        return ""
+    return parts[0].strip()
+
+
+def harvest_sheet_inquiry_cells(text: str) -> dict[str, str]:
+    """
+    按询价字段名切报价单原文。标签必须是独立词，不能切「指定起运港口」的中间。
+
+    一行两个字段时，值收到下一个字段名为止。副作用：无。
+    """
+    raw = text or ""
+    hits: list[tuple[int, str]] = []
+    for label in _SHEET_CLEAN_KEYS:
+        start = 0
+        while True:
+            pos = raw.find(label, start)
+            if pos < 0:
+                break
+            if _label_is_token(raw, pos, label):
+                hits.append((pos, label))
+            start = pos + len(label)
+    if not hits:
+        return {}
+    hits.sort()
+    kept: list[tuple[int, str]] = []
+    for pos, label in hits:
+        if kept and kept[-1][0] == pos and len(label) <= len(kept[-1][1]):
+            continue
+        if kept and kept[-1][0] == pos:
+            kept[-1] = (pos, label)
+            continue
+        kept.append((pos, label))
+    out: dict[str, str] = {}
+    for i, (pos, label) in enumerate(kept):
+        begin = pos + len(label)
+        end = kept[i + 1][0] if i + 1 < len(kept) else len(raw)
+        val = clip_sheet_field_value(raw[begin:end])
+        if val:
+            out[label] = val
+    return out
+
+
+def _label_is_token(raw: str, pos: int, label: str) -> bool:
+    """
+    标签必须是独立词。
+
+    「指定起运港口」中间的起运港不要。重量(KG)/体积(CBM) 留给更长标签。
+    """
+    if pos > 0:
+        prev = raw[pos - 1]
+        if prev.isalnum() or "\u4e00" <= prev <= "\u9fff":
+            return False
+    end = pos + len(label)
+    nxt = raw[end : end + 6]
+    if label in {"重量", "体积"} and nxt.startswith("("):
+        return False
+    return True
+
+
+def _clip_sheet_inquiry_facts(facts: dict[str, str]) -> dict[str, str]:
+    """已抽出的询价键再洗一遍表头。不改地址。"""
+    out = dict(facts or {})
+    for key in list(out):
+        if key not in _SHEET_CLEAN_KEYS:
+            continue
+        val = str(out.get(key) or "").strip()
+        if not val:
+            continue
+        out[key] = clip_sheet_field_value(val)
+    return {key: val for key, val in out.items() if str(val or "").strip()}
+
+
 def _merge_optional_inquiry_labels(text: str, facts: dict[str, str]) -> dict[str, str]:
-    """把原话里的通关口岸写进询价事实。有值才覆盖。"""
+    """把原话里的通关口岸、货好时间、提货/收货地址写进询价事实。有值才覆盖。"""
     from agent.schema.field_validate import normalize_facts
 
     merged = dict(facts or {})
     extra = normalize_facts(harvest_inquiry_optional_labels(text))
     for key, val in extra.items():
-        if val:
-            merged[key] = val
+        cleaned = clip_inquiry_value(val)
+        if key in _SHEET_CLEAN_KEYS:
+            cleaned = clip_sheet_field_value(cleaned)
+        if not cleaned:
+            continue
+        old = str(merged.get(key) or "").strip()
+        if old and cleaned.startswith(old) and len(cleaned) > len(old):
+            continue
+        if old and ("必填" in cleaned or "\n" in cleaned or len(cleaned) > len(old) * 3):
+            continue
+        merged[key] = cleaned
     return merged
 
 
@@ -235,7 +433,8 @@ def _harvest_labels(text: str, labels: tuple[str, ...]) -> dict[str, str]:
             pos = raw.find(label, start)
             if pos < 0:
                 break
-            hits.append((pos, label))
+            if _label_is_token(raw, pos, label):
+                hits.append((pos, label))
             start = pos + len(label)
     if not hits:
         return {}
@@ -429,7 +628,8 @@ def extract_product_quote(text: str, *, injected: Optional[dict[str, Any]] = Non
     raw = (text or "").strip()
     if not raw:
         return {}
-    compact = raw.replace(" ", "").replace("　", "")
+    stripped = raw.replace("@询价小助手", "").replace("询价小助手", "")
+    compact = stripped.replace(" ", "").replace("　", "")
     if re.fullmatch(r"WO\d{12}", compact, re.I):
         return {}
     rows = _parse_fee_rows(raw)

inquiry-agent/agent/policy/air_group_ops.py

@@ -611,7 +611,9 @@ def apply_air_lock(
     # 不能只回「TMS接口异常，IT运维排查中」。
     no = ticket_no(ticket)
     name = operator_name(flow, sender_id)
-    quote = ticket_view(ticket).get("quote") or {}
+    view_quote = dict(ticket_view(ticket).get("quote") or {})
+    air_quote = dict((view_quote.get("segment_quotes") or {}).get("AIR") or {})
+    quote = air_quote or view_quote
     options = _air_options(quote)
     quote_id = _quote_id(quote)
     picked = (option_no or "").strip()

inquiry-agent/agent/policy/air_text_flow.py

@@ -493,6 +493,10 @@ class AirTextInquiryFlow:
                 view.get("owned"),
             )
             return None
+        if (str(view.get("business_line") or "")).upper() == "MULTI":
+            # 多段联运由 MultiTextInquiryFlow 接卡。这里收回会变成 tms_miss，点拉群没回声。
+            logger.info("card 多段工单不走空运收回 wo=%s", work_order_no)
+            return None
         sess = self._session_from_ledger(sender_id=sender_id, view=view)
         self._save(sess)
         logger.info(

inquiry-agent/agent/policy/attachment_wave.py

@@ -23,9 +23,34 @@ from agent.policy.image_wave import WaveKV
 WAIT_TTL_SECONDS = 30 * 60
 # 已开始认的波次短协调：超时后当新开，不当六态。
 _WAVE_TTL_SEC = 2 * 3600
+# 报价单整理后常是空的，文件名仍能标运输方式。
+_LAND_NAME_HINTS = ("陆运", "零担", "拼车", "整车", "中港", "城配")
+_SEA_NAME_HINTS = ("海运",)
+_AIR_NAME_HINTS = ("空运",)
 logger = logging.getLogger(__name__)
 
 
+def infer_attach_mode_word(filename: str) -> str:
+    """
+    从附件文件名读运输方式。只认一种才返回「陆运/海运/空运」。
+
+    陆运报价单被千问整理成空时，仍要靠文件名留下一段，避免口播只剩海运。
+    """
+    name = (filename or "").strip()
+    if not name:
+        return ""
+    hits: list[str] = []
+    if any(word in name for word in _LAND_NAME_HINTS):
+        hits.append("陆运")
+    if any(word in name for word in _SEA_NAME_HINTS):
+        hits.append("海运")
+    if any(word in name for word in _AIR_NAME_HINTS):
+        hits.append("空运")
+    if len(hits) == 1:
+        return hits[0]
+    return ""
+
+
 @dataclass
 class AttachMaterial:
     """一份按到达/上传顺序记下的材料。text 在识别完成后才填。"""
@@ -270,7 +295,12 @@ class AttachWaveBook:
                 self._persist(sid)
 
     def compose_oral(self, sender_id: str) -> str:
-        """开口字 + 按材料顺序的原文 + 后到的字。"""
+        """
+        开口字 + 按材料顺序的原文 + 后到的字。
+
+        文件名能点名运输方式时，即使该份整理结果为空，也写「段N：陆运」这类标，
+        好让多段切段看见陆运+海运，而不是只走海运单聊。
+        """
         sid = (sender_id or "").strip()
         with self._lock:
             self._hydrate(sid)
@@ -280,9 +310,18 @@ class AttachWaveBook:
         parts: list[str] = []
         if wave.precede_text.strip():
             parts.append(wave.precede_text.strip())
+        seg = 0
         for row in wave.materials:
-            if row.text.strip():
-                parts.append(row.text.strip())
+            body = (row.text or "").strip()
+            word = infer_attach_mode_word(row.filename)
+            if word:
+                seg += 1
+                block = f"段{seg}：{word}"
+                if body:
+                    block = f"{block}\n{body}"
+                parts.append(block)
+            elif body:
+                parts.append(body)
         if wave.follow_text.strip():
             parts.append(wave.follow_text.strip())
         return "\n".join(parts)

inquiry-agent/agent/policy/inquiry_copy.py

@@ -219,6 +219,9 @@ _LAND_EXAMPLES = {
     "毛重": "50kg",
     "车型/数量": "9米6/1",
     "通关口岸": "皇岗",
+    "运输类型": "国内运输拼车",
+    "线路类别": "国内长途/零担",
+    "运输分类": "急件",
 }
 
 
@@ -321,13 +324,17 @@ def ask_land_fields(
     报价日期不进待补充。
     """
     src = dict(facts or {})
-    skip = {"报价日期", "运输类型", "线路类别", "运输分类"}
+    skip = {"报价日期"}
     missing = [k for k in missing_keys if k and k not in skip]
     land_type = (land_subtype or src.get("运输类型") or "").strip()
     optional = _land_optional_keys(land_type)
     lines = ["请确认并补充以下信息：", "运输方式：陆运"]
     if land_type:
         lines.append(f"运输类型：{land_type}")
+    for key in ("线路类别", "运输分类"):
+        val = (src.get(key) or "").strip()
+        if val:
+            lines.append(_land_field_line(key, val, required=True))
     for key in _land_cargo_keys(land_type, src):
         val = (src.get(key) or "").strip()
         if val:
@@ -373,7 +380,9 @@ def _air_known_value(facts: dict[str, str], key: str) -> str:
         return (src.get("包装方式") or src.get("包装类型") or "").strip()
     if key == "报价日期":
         return (src.get("报价日期") or "").strip()
-    return (src.get(key) or "").strip()
+    from agent.llm.extract_text import clip_inquiry_value
+
+    return clip_inquiry_value(src.get(key) or "")
 
 
 def _sea_known_value(facts: dict[str, str], key: str) -> str:
@@ -1483,6 +1492,52 @@ def sea_tms_miss_wecom_payload(*, work_order_no: str, action_url: str = "") -> d
     return {"msgtype": "template_card", "template_card": card}
 
 
+MULTI_CONFIRM_TAIL = "请回复「确定」创建工单，内容有误时，请重新发内容。"
+MULTI_SAME_MODE = "同一运输方式出现了两段，请拆开再发。"
+MULTI_CONFIRM_EXPIRED = "刚才的多段核对已过期，请把多段需求再发一次，出核对后再回复确定。"
+
+
+def multi_air_manual_hint(*, work_order_no: str) -> str:
+    """含空运的多段：不自动拉群，请销售把人和智能体拉进原有群并发送工单号。"""
+    no = (work_order_no or "").strip()
+    return f"请手动拉相关人员和智能体到原有群，并在群聊发送工单编号{no}"
+
+
+def multi_pull_text(*, work_order_no: str) -> str:
+    """不含空运：建单后的拉群文字兜底。不写 TMS、不写暂无报价。"""
+    no = (work_order_no or "").strip()
+    return "\n".join(
+        [
+            f"询价工单 {no}",
+            "工单已创建。",
+            "是否拉产品人员进群协同？",
+            BTN_PULL_COLLAB,
+        ]
+    )
+
+
+def multi_pull_wecom_payload(*, work_order_no: str) -> dict[str, Any]:
+    """不含空运：一张卡、一个按钮，点下去才按各段线路建群。"""
+    no = (work_order_no or "").strip()
+    return {
+        "msgtype": "template_card",
+        "template_card": {
+            "card_type": "button_interaction",
+            "source": {"desc": "多段联运", "desc_color": 0},
+            "main_title": {"title": f"询价工单 {no}", "desc": "工单已创建。"},
+            "sub_title_text": "是否拉产品人员进群协同？",
+            "task_id": card_task_id("multi_pull", no or "multi"),
+            "button_list": [
+                {
+                    "text": BTN_PULL_COLLAB,
+                    "style": 1,
+                    "key": f"{BTN_PULL_COLLAB_KEY}:{no}",
+                }
+            ],
+        },
+    }
+
+
 def handoff_air_hint(*, work_order_no: str) -> str:
     """空运转人工：只发文字，不建群。"""
     no = (work_order_no or "").strip()

inquiry-agent/agent/policy/multi_group_ops.py

@@ -0,0 +1,569 @@
+"""
+多段联运的群内报价与成交跟进。
+
+本文件职责：按发言人岗位把报价归到对应段；一段报完就出该段报价单；
+全部段都出了报价单，才在群里发一张整票成交跟进，各段价格不加总。
+含空运时，原有群发出工单号后在这里激活：@人、发各段需求和模板。
+调用：群入站 Handler，在 Worker 线程。禁止查 TMS 运价；禁止按段各建工单。
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+from typing import Any, Callable, Optional
+
+from agent.llm.extract_text import extract_collab_fields, extract_product_quote
+from agent.policy import inquiry_copy as copy
+from agent.policy.multi_segments import Segment, group_brief, includes_air, segments_from_payload
+from agent.policy.sea_group_ops import (
+    apply_group_deal,
+    group_client_of,
+    is_sales,
+    send_filled_sheet,
+    send_group_text,
+    ticket_no,
+    ticket_view,
+)
+from agent.schema.field_validate import collab_fields_for_ticket
+
+logger = logging.getLogger(__name__)
+
+_ROLE_MODE = {"sea": "SEA", "land": "LAND", "air": "AIR"}
+SheetFn = Callable[..., dict[str, Any]]
+
+
+def ticket_is_multi(ticket: object) -> bool:
+    """主账业务线是多段联运。"""
+    view = ticket_view(ticket)
+    line = str(view.get("business_line") or view.get("businessLine") or "").upper()
+    return line == "MULTI"
+
+
+def load_segments(ticket: object) -> list[Segment]:
+    """从工单事实里的 segments_json 读回各段。坏数据当没有段。"""
+    view = ticket_view(ticket)
+    facts = dict(view.get("facts") or {})
+    raw = facts.get("segments_json") or ""
+    if not raw:
+        return []
+    try:
+        payload = json.loads(raw)
+    except json.JSONDecodeError:
+        logger.warning("多段 segments_json 无法解析 wo=%s", ticket_no(ticket))
+        return []
+    return segments_from_payload(payload)
+
+
+def segment_quotes(ticket: object) -> dict[str, dict[str, Any]]:
+    """已落到各段的报价。键是 SEA/LAND/AIR。"""
+    quote = dict(ticket_view(ticket).get("quote") or {})
+    raw = quote.get("segment_quotes") or {}
+    if not isinstance(raw, dict):
+        return {}
+    out: dict[str, dict[str, Any]] = {}
+    for key, val in raw.items():
+        mode = str(key or "").upper()
+        if mode in _ROLE_MODE.values() and isinstance(val, dict):
+            out[mode] = dict(val)
+    return out
+
+
+def sender_mode(flow: Any, sender_id: str) -> str:
+    """
+    发言人负责哪一段。
+
+    海运产品归海运，陆运产品归陆运，航线·空运归空运。对不上返回空。
+    """
+    book = getattr(flow, "ledger", None) or getattr(flow, "_ledger", None)
+    finder = getattr(book, "get_staff_by_wecom_id", None) if book is not None else None
+    if not callable(finder):
+        return ""
+    try:
+        row = finder(wecom_id=sender_id) or {}
+    except Exception:
+        logger.exception("多段查岗位失败 wecom=%s", sender_id)
+        return ""
+    role = str(row.get("roleCode") or row.get("role_code") or "").strip().lower()
+    return _ROLE_MODE.get(role, "")
+
+
+def handle_multi_inbound(
+    *,
+    flow: Any,
+    ticket: object,
+    chat_id: str,
+    sender_id: str,
+    text: str,
+    injected_quote: Optional[dict[str, Any]] = None,
+    sheet_fn: Optional[SheetFn] = None,
+) -> str:
+    """
+    群里一条多段消息。
+
+    只含工单号：绑定本群并发摘要和各段模板。
+    有价格：归到发言人那一段并出该段报价单。全部出齐才成交跟进。
+    副作用：写主账报价、可能发群消息和模板。不查 TMS。
+    """
+    room = (chat_id or "").strip()
+    no = ticket_no(ticket)
+    segments = load_segments(ticket)
+    if not segments:
+        send_group_text(group_client_of(flow), room, f"工单{no}没有可报价的运输段。")
+        return "multi_no_segments"
+    only_no = copy.extract_work_order_no(text or "") == no and _only_work_order(text or "")
+    if only_no or not (ticket_view(ticket).get("collab_chat_id") or ""):
+        _bind_room(flow, ticket, room, sender_id)
+        ticket = _reload(flow, no) or ticket
+    cabin = _cabin_intent(text or "")
+    if cabin and includes_air(segments):
+        return _cabin(flow, ticket, room, sender_id, text or "", cabin)
+    if only_no:
+        _send_activation(flow, ticket, room, segments, sender_id=sender_id, sheet_fn=sheet_fn)
+        return "multi_activated"
+    mode = sender_mode(flow, sender_id)
+    target = next((seg for seg in segments if seg.mode == mode), None)
+    fields = {}
+    if target is not None:
+        keys = collab_fields_for_ticket(target.facts, target.mode)
+        fields = extract_collab_fields(text or "", allowed_keys=keys)
+        if fields:
+            _save_collab(flow, ticket, target, fields)
+            ticket = _reload(flow, no) or ticket
+            segments = load_segments(ticket)
+            target = next((seg for seg in segments if seg.mode == mode), target)
+    quote = extract_product_quote(text or "", injected=injected_quote)
+    if not quote:
+        if fields:
+            return "multi_collab_saved"
+        if is_sales(ticket, sender_id):
+            return _maybe_deal(flow, ticket, sender_id, text)
+        send_group_text(group_client_of(flow), room, copy.group_quote_not_recognized())
+        return "quote_not_recognized"
+    if target is None:
+        send_group_text(group_client_of(flow), room, "没有对上你负责的运输段，请让对应岗位报价。")
+        return "quote_wrong_role"
+    return _save_segment_quote(
+        flow=flow,
+        ticket=ticket,
+        chat_id=room,
+        segment=target,
+        segments=segments,
+        quote=quote,
+        sheet_fn=sheet_fn,
+    )
+
+
+def deal_text(*, work_order_no: str, segments: list[Segment], quotes: dict[str, dict[str, Any]]) -> str:
+    """整票成交跟进文字。每段各自的价格、有效期，海运陆运再加时效。不加总。"""
+    no = (work_order_no or "").strip()
+    lines = [f"询价工单 {no}", copy.DEAL_SUBTITLE]
+    for seg in segments:
+        quote = quotes.get(seg.mode) or {}
+        lines.append(f"段{seg.index}：{seg.word()}")
+        rows = copy.deal_horizontal_rows(quote=quote, facts=seg.facts, business_line=seg.mode)
+        lines.extend(f"{row['keyname']}：{row['value']}" for row in rows)
+    lines.append(copy.DEAL_HINT)
+    return "\n".join(lines)
+
+
+def deal_payload(
+    *,
+    work_order_no: str,
+    segments: list[Segment],
+    quotes: dict[str, dict[str, Any]],
+    card_version: int = 1,
+) -> dict[str, Any]:
+    """成交卡一行一段报价，不出现各段加总。按钮仍是整票一次确认。"""
+    base = copy.deal_wecom_payload(
+        work_order_no=work_order_no,
+        quote={"source_label": "分段报价"},
+        business_line="MULTI",
+        card_version=card_version,
+    )
+    rows = []
+    for seg in segments:
+        quote = quotes.get(seg.mode) or {}
+        total = copy.money_text(
+            str(quote.get("total") or "-").strip() or "-",
+            str(quote.get("currency") or ""),
+        )
+        rows.append({"keyname": f"段{seg.index}{seg.word()}", "value": copy._clip(total, 26)})
+    card = dict(base.get("template_card") or {})
+    card["horizontal_content_list"] = rows
+    card["source"] = {"desc": "分段报价", "desc_color": 0}
+    return {"msgtype": "template_card", "template_card": card}
+
+
+def _save_segment_quote(
+    *,
+    flow: Any,
+    ticket: object,
+    chat_id: str,
+    segment: Segment,
+    segments: list[Segment],
+    quote: dict[str, Any],
+    sheet_fn: Optional[SheetFn],
+) -> str:
+    """写入这一段报价并发该段报价单。其他段未齐时不发成交跟进。"""
+    no = ticket_no(ticket)
+    stored = segment_quotes(ticket)
+    stored[segment.mode] = dict(quote)
+    version = int(dict(ticket_view(ticket).get("quote") or {}).get("deal_version") or 0)
+    all_ready = all(seg.mode in stored and _has_money(stored[seg.mode]) for seg in segments)
+    if all_ready:
+        version += 1
+    payload = {
+        "segment_quotes": stored,
+        "source_label": "分段报价",
+        "deal_version": version,
+    }
+    writer = getattr(flow.ledger, "upsert_quote", None)
+    if callable(writer):
+        writer(
+            work_order_no=no,
+            quote=payload,
+            to_status="已报价" if all_ready else "",
+        )
+    ticket = _reload(flow, no) or ticket
+    _send_sheet(
+        flow=flow,
+        ticket=ticket,
+        chat_id=chat_id,
+        segment=segment,
+        quote=quote,
+        sheet_fn=sheet_fn,
+    )
+    client = group_client_of(flow)
+    if not all_ready:
+        send_group_text(client, chat_id, f"段{segment.index}{segment.word()}报价单已发出。其他段报完后一起跟进成交。")
+        return "segment_quoted"
+    text = deal_text(work_order_no=no, segments=segments, quotes=stored)
+    send_group_text(client, chat_id, text)
+    carder = getattr(client, "send_group_card", None)
+    if callable(carder):
+        carder(
+            chat_id=chat_id,
+            template_card=deal_payload(
+                work_order_no=no,
+                segments=segments,
+                quotes=stored,
+                card_version=version,
+            ).get("template_card")
+            or {},
+        )
+    logger.info("multi.deal wo=%s segments=%s", no, [seg.mode for seg in segments])
+    return "multi_deal_sent"
+
+
+def _cabin(flow: Any, ticket: object, chat_id: str, sender_id: str, text: str, intent: str) -> str:
+    """空运段的锁舱、释放走现有空运规则，不挡住其他段报价。"""
+    from agent.policy.air_group_ops import apply_air_lock, apply_air_release
+
+    if intent == "release":
+        return apply_air_release(
+            flow=flow,
+            ticket=ticket,
+            sender_id=sender_id,
+            chat_id=chat_id,
+            instruction=text,
+        )
+    return apply_air_lock(
+        flow=flow,
+        ticket=ticket,
+        sender_id=sender_id,
+        chat_id=chat_id,
+        instruction=text,
+    )
+
+
+def _cabin_intent(text: str) -> str:
+    raw = text or ""
+    if "释放" in raw and "舱" in raw:
+        return "release"
+    if "锁舱" in raw:
+        return "lock"
+    return ""
+
+
+def _send_activation(
+    flow: Any,
+    ticket: object,
+    chat_id: str,
+    segments: list[Segment],
+    *,
+    sender_id: str = "",
+    sheet_fn: Optional[SheetFn],
+) -> None:
+    """原有群激活：@具体销售和各段产品，发核对同款字段，每段一份模板。不含 TMS。"""
+    no = ticket_no(ticket)
+    sales_name, products, mention_ids = _activation_mentions(
+        flow, ticket, segments, sender_id=sender_id
+    )
+    brief = group_brief(
+        work_order_no=no,
+        segments=segments,
+        sales_name=sales_name,
+        mention_names=products,
+    )
+    send_group_text(group_client_of(flow), chat_id, brief, mention_userids=mention_ids)
+    for seg in segments:
+        _send_sheet(flow=flow, ticket=ticket, chat_id=chat_id, segment=seg, quote={}, sheet_fn=sheet_fn)
+
+
+def _activation_mentions(
+    flow: Any,
+    ticket: object,
+    segments: list[Segment],
+    sender_id: str = "",
+) -> tuple[str, list[str], list[str]]:
+    """激活时要 @ 的人：具体销售 + 各段对得上的产品岗；空运按航线岗名单。"""
+    from agent.policy.sea_group_ops import resolve_group_mentions
+
+    sales_name, products, mention_ids = resolve_group_mentions(
+        flow, ticket, None, fallback_sales_id=sender_id
+    )
+    book = getattr(flow, "ledger", None) or getattr(flow, "_ledger", None)
+    for seg in segments:
+        for row in _staff_rows(book, seg):
+            uid = str(row.get("wecomId") or row.get("wecom_id") or "").strip()
+            name = str(row.get("name") or "").strip() or uid
+            if uid and uid not in mention_ids:
+                mention_ids.append(uid)
+            if name and name not in products and name != sales_name:
+                products.append(name)
+    return sales_name, products, mention_ids
+
+
+def _staff_rows(book: Any, seg: Segment) -> list[dict[str, Any]]:
+    """一段要对上的产品人员。线路没填时，陆运/空运按岗位名单，不猜港口。"""
+    if book is None:
+        return []
+    try:
+        if seg.mode == "SEA":
+            matcher = getattr(book, "match_sea_staff", None)
+            if not callable(matcher):
+                return []
+            rows = matcher(
+                origin=str(seg.facts.get("起运港") or ""),
+                destination=str(seg.facts.get("目的港") or ""),
+            ) or []
+            return [dict(x) for x in rows if isinstance(x, dict)]
+        if seg.mode == "LAND":
+            matcher = getattr(book, "match_land_staff", None)
+            rows = []
+            if callable(matcher):
+                rows = matcher(route_category=str(seg.facts.get("线路类别") or "")) or []
+            hits = [dict(x) for x in rows if isinstance(x, dict)]
+            if hits:
+                return hits
+            lister = getattr(book, "list_staff_by_role", None)
+            if not callable(lister):
+                return []
+            return [dict(x) for x in (lister(role_code="land") or []) if isinstance(x, dict)]
+        if seg.mode == "AIR":
+            lister = getattr(book, "list_staff_by_role", None)
+            if not callable(lister):
+                return []
+            return [dict(x) for x in (lister(role_code="air") or []) if isinstance(x, dict)]
+    except Exception:
+        logger.exception("多段激活匹配产品失败 mode=%s", seg.mode)
+    return []
+
+
+def _send_sheet(
+    *,
+    flow: Any,
+    ticket: object,
+    chat_id: str,
+    segment: Segment,
+    quote: dict[str, Any],
+    sheet_fn: Optional[SheetFn],
+) -> None:
+    """按这一段的运输方式匹配模板。失败只记日志，不挡住报价归段。"""
+    sender = sheet_fn or send_filled_sheet
+    try:
+        sender(
+            flow=flow,
+            ticket=_sheet_ticket(ticket, segment),
+            quote_override=dict(quote or {}),
+            business_line=segment.mode,
+            chat_id=chat_id,
+        )
+    except Exception:
+        logger.exception("多段模板发送失败 wo=%s mode=%s", ticket_no(ticket), segment.mode)
+
+
+def _sheet_ticket(ticket: object, segment: Segment) -> object:
+    """填表时用这一段的询价字段，避免海运模板吃到空运字段。"""
+    view = ticket_view(ticket)
+    facts = _facts_for_segment_sheet(segment)
+    sales_id = str(view.get("sales_wecom_id") or view.get("salesWecomId") or "").strip()
+    if isinstance(ticket, dict):
+        cloned = dict(ticket)
+        cloned["facts"] = facts
+        cloned["business_line"] = segment.mode
+        cloned["businessLine"] = segment.mode
+        if sales_id:
+            cloned["sales_wecom_id"] = sales_id
+            cloned["salesWecomId"] = sales_id
+        return cloned
+    return {
+        "work_order_no": view.get("work_order_no"),
+        "workOrderNo": view.get("work_order_no"),
+        "facts": facts,
+        "business_line": segment.mode,
+        "businessLine": segment.mode,
+        "collab_chat_id": view.get("collab_chat_id"),
+        "sales_wecom_id": sales_id,
+        "salesWecomId": sales_id,
+        "quote": {},
+    }
+
+
+def _facts_for_segment_sheet(segment: Segment) -> dict[str, str]:
+    """
+    一段填表用的字段。
+
+    多段不查 TMS，陆运常缺线路类别。
+    后台陆运拼车模板关键词是「国内拼车」，询价字段是「国内运输拼车」，
+    不补别名就套不上表，只会发出空运模板。
+    """
+    facts = {k: str(v).strip() for k, v in dict(segment.facts).items() if str(v or "").strip()}
+    if segment.mode != "LAND":
+        return facts
+    from agent.schema.land_options import SAMPLE_PAIRS, canonicalize_transport_type
+
+    land_type = canonicalize_transport_type(str(facts.get("运输类型") or ""))
+    if land_type:
+        facts["运输类型"] = land_type
+    if land_type and not str(facts.get("线路类别") or "").strip():
+        for pair_type, pair_route in SAMPLE_PAIRS:
+            if pair_type == land_type:
+                facts["线路类别"] = pair_route
+                break
+    hint = _LAND_SHEET_HINTS.get(land_type, "")
+    if hint:
+        facts["模板匹配"] = hint
+    return facts
+
+
+_LAND_SHEET_HINTS = {
+    "国内运输拼车": "国内拼车 国内零担",
+    "国内运输整车": "国内整车",
+    "跨境整车": "跨境整车",
+    "跨境集拼": "跨境集拼 跨境零担",
+    "中港整车": "中港",
+    "中港零担/集拼": "中港",
+}
+
+
+def _save_collab(flow: Any, ticket: object, segment: Segment, fields: dict[str, str]) -> None:
+    """协同字段写进对应段，不挡报价。"""
+    segments = load_segments(ticket)
+    for seg in segments:
+        if seg.mode != segment.mode:
+            continue
+        seg.facts.update({k: str(v) for k, v in fields.items() if str(v).strip()})
+    _write_segments(flow, ticket_no(ticket), segments)
+    writer = getattr(flow.ledger, "patch_collab_facts", None)
+    if callable(writer):
+        writer(work_order_no=ticket_no(ticket), facts=fields, sender_id="")
+
+
+def _bind_room(flow: Any, ticket: object, chat_id: str, sender_id: str) -> None:
+    """把原有群绑到这张多段工单。已绑同一个群则不重复建。"""
+    view = ticket_view(ticket)
+    if str(view.get("collab_chat_id") or "") == chat_id:
+        return
+    binder = getattr(flow.ledger, "bind_collab_group", None)
+    members = [str(x) for x in (view.get("collab_member_ids") or [])]
+    if sender_id and sender_id not in members:
+        members.append(sender_id)
+    sales = str(view.get("sales_wecom_id") or view.get("salesWecomId") or sender_id or "")
+    if sales and sales not in members:
+        members.append(sales)
+    if callable(binder):
+        binder(
+            work_order_no=ticket_no(ticket),
+            chat_id=chat_id,
+            member_ids=members,
+            product_names=list(view.get("product_names") or []),
+            product_ids=list(view.get("product_ids") or []),
+        )
+
+
+def _maybe_deal(flow: Any, ticket: object, sender_id: str, text: str) -> str:
+    """销售在群里确认成交。还有段没报价单时不确认。"""
+    from agent.routing.group_intent import (
+        INTENT_DEAL,
+        INTENT_LOST,
+        INTENT_NEGOTIATE,
+        classify_group_text,
+    )
+
+    intent = classify_group_text(text or "")
+    outcome = {INTENT_DEAL: "已成交", INTENT_LOST: "未成交", INTENT_NEGOTIATE: "协商中"}.get(intent)
+    if not outcome:
+        return "multi_noop"
+    segments = load_segments(ticket)
+    quotes = segment_quotes(ticket)
+    if not all(seg.mode in quotes and _has_money(quotes[seg.mode]) for seg in segments):
+        send_group_text(
+            group_client_of(flow),
+            str(ticket_view(ticket).get("collab_chat_id") or ""),
+            "还有运输段没有报价单，全部报完后再确认成交。",
+        )
+        return "deal_wait_segments"
+    return apply_group_deal(flow=flow, ticket=ticket, sender_id=sender_id, outcome=outcome)
+
+
+def _write_segments(flow: Any, work_order_no: str, segments: list[Segment]) -> None:
+    from agent.policy.multi_segments import segments_to_payload
+
+    writer = getattr(flow.ledger, "update_facts", None)
+    if not callable(writer):
+        return
+    current = {}
+    getter = getattr(flow.ledger, "get_ticket", None)
+    if callable(getter):
+        ticket = getter(work_order_no=work_order_no)
+        current = dict(ticket_view(ticket).get("facts") or {})
+    current["segments_json"] = json.dumps(segments_to_payload(segments), ensure_ascii=False)
+    writer(work_order_no=work_order_no, facts=current)
+
+
+def _reload(flow: Any, work_order_no: str) -> object:
+    getter = getattr(flow.ledger, "get_ticket", None)
+    if callable(getter):
+        return getter(work_order_no=work_order_no)
+    return None
+
+
+def _has_money(quote: dict[str, Any]) -> bool:
+    return bool(str(quote.get("total") or "").strip() or quote.get("fee_rows") or quote.get("fee_lines"))
+
+
+def _only_work_order(text: str) -> bool:
+    """去掉工单号和 @询价小助手 后几乎没字，当作激活。与空运群同一套，避免 WO 数字被当成报价。"""
+    from agent.routing.air_group_intent import _only_work_order as air_only
+
+    return air_only(text)
+
+
+def send_segment_templates(*, flow: Any, ticket: object, chat_id: str, sheet_fn: Optional[SheetFn] = None) -> None:
+    """拉群或激活后，按段各发一份匹配模板。没有段则不发。"""
+    for seg in load_segments(ticket):
+        _send_sheet(
+            flow=flow,
+            ticket=ticket,
+            chat_id=chat_id,
+            segment=seg,
+            quote={},
+            sheet_fn=sheet_fn,
+        )
+
+
+def multi_has_air(ticket: object) -> bool:
+    """这张多段单含空运段，锁舱和含电含磁才适用。"""
+    return includes_air(load_segments(ticket))

inquiry-agent/agent/policy/multi_segments.py

@@ -0,0 +1,527 @@
+"""
+多段联运的分段与文案。
+
+本文件职责：按原文出现顺序把不同运输方式切成段；汇总各段必填缺口；
+拼私聊核对（只有询价字段）和群摘要（询价字段 + 协同字段，不写 TMS 报价）。
+禁止：猜港口、查 TMS、建群、改六态。调用方在 Worker 线程使用，本模块无共享写状态。
+"""
+
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass, field
+from typing import Any
+
+from agent.policy import inquiry_copy as copy
+from agent.schema.field_validate import (
+    harvest_oral_measures,
+    normalize_facts,
+    validate_required_fields,
+)
+from agent.llm.extract_text import parse_labeled_facts
+from agent.schema.land_options import (
+    harvest_land_options,
+    option_prompt,
+    option_prompt_needs_sample,
+    scrub_land_placeholder_ports,
+    validate_land_options,
+)
+from agent.schema.sea_options import harvest_sea_class
+
+# 行首或「段N：」后面的运输方式。排除「空运费」，避免费用词被当成新的一段。
+_MODE_RE = re.compile(r"(?:段\s*(\d+)\s*[:：]\s*)?(空运|海运|陆运)(?!费)")
+# 已经写了「段1：陆运」时，段内再写「运输方式：海运」不能再切一段。
+_EXPLICIT_SEG = re.compile(r"段\s*(\d+)\s*[:：]\s*(空运|海运|陆运)(?!费)")
+# 单段标签行没覆盖的陆运销售叫法。内部键仍交给字段校验去归一。
+_EXTRA_LABEL = re.compile(r"^(始发站|货物品名|运输类型|线路类别)\s*[:：]\s*(.+)$")
+_SEG_MARK = re.compile(r"段\s*(\d+)")
+
+_MODE_BY_WORD = {"空运": "AIR", "海运": "SEA", "陆运": "LAND"}
+_WORD_BY_MODE = {"AIR": "空运", "SEA": "海运", "LAND": "陆运"}
+_LAND_OPTION_KEYS = frozenset({"运输类型", "线路类别", "运输分类"})
+
+
+@dataclass
+class Segment:
+    """一段询价。index 从 1 起，在同一张工单里保持不变。"""
+
+    index: int
+    mode: str
+    facts: dict[str, str] = field(default_factory=dict)
+    source: str = ""
+
+    def word(self) -> str:
+        return _WORD_BY_MODE.get(self.mode, self.mode)
+
+
+@dataclass
+class SplitOutcome:
+    """切段结果。ok 才是可继续的多段；same_mode 表示同一种方式出现了两次。"""
+
+    ok: bool = False
+    same_mode: bool = False
+    segments: list[Segment] = field(default_factory=list)
+
+
+def classify_multi_text(text: str) -> str:
+    """
+    这句话是不是多段。
+
+    返回 multi / same_mode / single。单段仍走原来的海运、陆运、空运流程。
+    """
+    outcome = split_segments(text)
+    if outcome.same_mode:
+        return "same_mode"
+    if outcome.ok:
+        return "multi"
+    return "single"
+
+
+def split_segments(text: str) -> SplitOutcome:
+    """
+    按运输方式出现的先后切段。
+
+    「段1：海运」和同一句里的「陆运+海运」都认。同一方式出现两次则 same_mode，不建单。
+    已经标了两段以上时，只按「段N：」切，表里重复的「海运/陆运」不当新段。
+    每段正文只收「字段名：值」，不猜没写标签的港口。
+    """
+    raw = text or ""
+    explicit = list(_EXPLICIT_SEG.finditer(raw))
+    hits = explicit if len(explicit) >= 2 else list(_MODE_RE.finditer(raw))
+    if len(hits) < 2:
+        return SplitOutcome()
+    modes = [_MODE_BY_WORD[hit.group(2)] for hit in hits]
+    if len(modes) != len(set(modes)):
+        return SplitOutcome(same_mode=True)
+    segments: list[Segment] = []
+    for i, hit in enumerate(hits):
+        end = hits[i + 1].start() if i + 1 < len(hits) else len(raw)
+        body = raw[hit.end() : end]
+        explicit = hit.group(1)
+        index = int(explicit) if explicit else i + 1
+        segments.append(
+            Segment(
+                index=index,
+                mode=modes[i],
+                facts=_facts_from_body(body, modes[i]),
+                source=body,
+            )
+        )
+    segments.sort(key=lambda seg: seg.index)
+    return SplitOutcome(ok=True, segments=segments)
+
+
+def _segment_required_check(seg: Segment) -> tuple[dict[str, Any], dict[str, str]]:
+    """
+    一段是否齐了询价必填。陆运还要三项选项，模板码站点当没填。
+
+    海运/空运仍走字段合同。返回校验结果和写回后的事实，不改入参。
+    """
+    if seg.mode == "LAND":
+        facts = scrub_land_placeholder_ports(normalize_facts(dict(seg.facts)))
+        opt = validate_land_options(facts)
+        facts.update({k: v for k, v in dict(opt.get("normalized") or {}).items() if v})
+        check = dict(
+            validate_required_fields(
+                facts=facts,
+                business_line="LAND",
+                land_subtype=str(facts.get("运输类型") or ""),
+            )
+        )
+        option_missing = [str(x) for x in (opt.get("missing") or []) if str(x).strip()]
+        cargo_missing = [str(x) for x in (check.get("missing") or []) if str(x).strip()]
+        missing = option_missing + [k for k in cargo_missing if k not in option_missing]
+        merged = dict(check.get("facts") or facts)
+        merged.update({k: facts[k] for k in ("运输类型", "线路类别", "运输分类") if facts.get(k)})
+        check["facts"] = merged
+        check["missing"] = missing
+        check["ok"] = not missing
+        check["need_land_subtype"] = False
+        check["need_land_options"] = bool(option_missing)
+        return check, merged
+    check = validate_required_fields(facts=dict(seg.facts), business_line=seg.mode)
+    facts = dict(seg.facts)
+    for key, val in dict(check.get("facts") or {}).items():
+        if val is None:
+            continue
+        text = str(val).strip()
+        if text:
+            facts[key] = text
+    return check, facts
+
+
+def refresh_segments(segments: list[Segment]) -> tuple[bool, list[Segment], str]:
+    """
+    按各段运输方式校验必填，并把默认报价日期写回。
+
+    返回：是否已齐、更新后的段、缺口文案。缺口一条消息说完，不套群摘要。
+    待补充的举例与单聊相同，例如「品名：(如：茶叶）」。
+    """
+    updated: list[Segment] = []
+    blocks: list[str] = []
+    ready = True
+    for seg in segments:
+        check, facts = _segment_required_check(seg)
+        filled = Segment(index=seg.index, mode=seg.mode, facts=facts, source=seg.source)
+        updated.append(filled)
+        body = _segment_clarify(filled, check)
+        if body:
+            ready = False
+            blocks.append(f"段{filled.index}：{filled.word()}\n{body}")
+    if ready:
+        return True, updated, ""
+    return False, updated, "请确认并补充以下信息：\n\n" + "\n\n".join(blocks)
+
+
+def merge_supplement(
+    segments: list[Segment],
+    text: str,
+    *,
+    extra_facts: dict[str, str] | None = None,
+) -> list[Segment]:
+    """
+    把销售补上的字段写回对应段。
+
+    已有草稿时，再写段1/段2只把新字段叠上去，不整单替换。
+    没有旧草稿才按切段结果当新开。写了段号就按段号归。
+    只有一段还缺、且没写段号时，标签行、口语机读和抽取结果都补到那一段。
+    各段已齐或两段都缺时：货物数量/箱型只补海运，件数/体积只补空运。对不上不猜。
+    extra_facts 来自单聊同一套抽取（DeepSeek B / 单测注入），不在这里猜港口。
+    """
+    resent = split_segments(text)
+    if resent.ok:
+        if segments:
+            return _overlay_segments(segments, resent.segments, extra_facts)
+        return resent.segments
+    marked = _split_by_segment_mark(text)
+    if marked:
+        out = [Segment(index=s.index, mode=s.mode, facts=dict(s.facts)) for s in segments]
+        by_index = {s.index: s for s in out}
+        for index, facts in marked.items():
+            target = by_index.get(index)
+            if target is None:
+                continue
+            for key, val in facts.items():
+                if str(val or "").strip():
+                    target.facts[key] = str(val).strip()
+        return out
+    parsed = _unlabeled_supplement_facts(text, extra_facts)
+    if not parsed:
+        return segments
+    lacking = _lacking_indexes(segments)
+    if len(lacking) == 1:
+        targets = lacking
+    else:
+        # 核对后改海运格子、或两段都缺时只补海运字段：按字段归属落到唯一一段。
+        targets = _indexes_for_supplement(segments, parsed)
+    if len(targets) != 1:
+        return segments
+    out = [Segment(index=s.index, mode=s.mode, facts=dict(s.facts)) for s in segments]
+    for seg in out:
+        if seg.index != targets[0]:
+            continue
+        for key, val in parsed.items():
+            if str(val or "").strip():
+                seg.facts[key] = str(val).strip()
+    return out
+
+
+def _overlay_segments(
+    existing: list[Segment],
+    incoming: list[Segment],
+    extra_facts: dict[str, str] | None = None,
+) -> list[Segment]:
+    """
+    把新切出来的段叠到旧草稿上。同号优先，否则同运输方式。
+
+    只覆盖非空新值，旧港口/品名保住。陆运选项从新段正文再收一遍。
+    """
+    _ = extra_facts
+    out = [
+        Segment(index=seg.index, mode=seg.mode, facts=dict(seg.facts), source=seg.source)
+        for seg in existing
+    ]
+    by_index = {seg.index: seg for seg in out}
+    by_mode: dict[str, Segment] = {}
+    for seg in out:
+        by_mode.setdefault(seg.mode, seg)
+    for neu in incoming:
+        target = by_index.get(neu.index) or by_mode.get(neu.mode)
+        if target is None:
+            continue
+        blob = (neu.source or "").strip()
+        if target.mode == "LAND":
+            overlay = harvest_land_options(blob, dict(neu.facts))
+        elif target.mode == "SEA":
+            overlay = harvest_sea_class(blob, dict(neu.facts))
+        else:
+            overlay = dict(neu.facts)
+        overlay = normalize_facts(overlay)
+        for key, val in overlay.items():
+            token = str(val or "").strip()
+            if token:
+                target.facts[key] = token
+        if target.mode == "SEA":
+            target.facts = harvest_sea_class(blob, target.facts)
+        if target.mode == "LAND":
+            target.facts = harvest_land_options(blob, target.facts)
+    return out
+
+
+def confirm_text(segments: list[Segment]) -> str:
+    """
+    私聊核对。每段只列询价字段，最后一句是创建工单。
+
+    不写协同字段，不写 TMS。各段沿用该运输方式现有核对卡的字段，去掉原「查价」尾句。
+    """
+    blocks: list[str] = []
+    for seg in segments:
+        body = _segment_inquiry_block(seg)
+        blocks.append(f"段{seg.index}：{seg.word()}\n{body}".rstrip())
+    blocks.append(copy.MULTI_CONFIRM_TAIL)
+    return "\n\n".join(blocks)
+
+
+def group_brief(
+    *,
+    work_order_no: str,
+    segments: list[Segment],
+    sales_name: str = "",
+    mention_names: list[str] | None = None,
+) -> str:
+    """
+    拉群后发到群里的摘要。
+
+    结构对齐私聊核对：@销售和各段产品，下一行直接接段1询价字段。
+    不写 TMS，不写协同段。工单号已在私聊提示过，群里不再重复一行。
+    """
+    _ = work_order_no
+    names: list[str] = []
+    sales = (sales_name or "").strip()
+    if sales:
+        names.append(sales)
+    for name in mention_names or []:
+        token = (name or "").strip()
+        if token and token not in names:
+            names.append(token)
+    lines: list[str] = []
+    if names:
+        lines.append("".join(f"@{n}" for n in names))
+    for i, seg in enumerate(segments):
+        # @人下面不要空行；段与段之间仍空一行，方便扫。
+        if i > 0:
+            lines.append("")
+        lines.append(f"段{seg.index}：{seg.word()}")
+        body = _segment_inquiry_block(seg).strip()
+        if body:
+            lines.extend(body.splitlines())
+    return "\n".join(lines)
+
+
+def segments_to_payload(segments: list[Segment]) -> list[dict[str, Any]]:
+    """书签和工单里保存的段。只保留字符串，避免书签序列化掉类型。"""
+    return [
+        {
+            "index": seg.index,
+            "mode": seg.mode,
+            "facts": {str(k): str(v) for k, v in (seg.facts or {}).items() if str(v).strip()},
+        }
+        for seg in segments
+    ]
+
+
+def segments_from_payload(raw: Any) -> list[Segment]:
+    """从书签读回段。坏数据当空，调用方不得据此建单。"""
+    if not isinstance(raw, list):
+        return []
+    out: list[Segment] = []
+    for item in raw:
+        if not isinstance(item, dict):
+            continue
+        mode = str(item.get("mode") or "").strip().upper()
+        if mode not in _WORD_BY_MODE:
+            continue
+        facts = {
+            str(k): str(v)
+            for k, v in dict(item.get("facts") or {}).items()
+            if str(v).strip()
+        }
+        try:
+            index = int(item.get("index") or len(out) + 1)
+        except (TypeError, ValueError):
+            index = len(out) + 1
+        out.append(Segment(index=index, mode=mode, facts=facts))
+    return out
+
+
+def includes_air(segments: list[Segment]) -> bool:
+    """任一段是空运，就走手动拉群，不点「拉产品进群」。"""
+    return any(seg.mode == "AIR" for seg in segments)
+
+
+def _segment_clarify(seg: Segment, check: dict[str, Any]) -> str:
+    """
+    一段的补问正文，格式对齐单聊：已识别 + 待补充带举例。
+
+    不要再写一遍「请确认并补充」和运输方式，外层已经有段号。
+    """
+    if check.get("ok") and not check.get("need_land_subtype"):
+        return ""
+    missing = [str(x) for x in (check.get("missing") or []) if str(x).strip()]
+    if check.get("need_land_subtype"):
+        return "待补充：\n运输类型（必填）：(如：国内运输拼车）"
+    if seg.mode == "LAND":
+        if check.get("need_land_options"):
+            # 和单聊一样先列出线路类别、运输分类，选完再问询价字段。
+            return option_prompt(current=seg.facts)
+        cargo_missing = [k for k in missing if k not in _LAND_OPTION_KEYS]
+        raw = copy.ask_land_fields(
+            facts=seg.facts,
+            missing_keys=cargo_missing,
+            land_subtype=str(seg.facts.get("运输类型") or ""),
+        )
+    else:
+        raw = copy.ask_clarify(
+            facts=seg.facts,
+            missing_keys=missing,
+            transport_mode=seg.word(),
+        )
+    if not raw:
+        return ""
+    skip = {"请确认并补充以下信息：", f"运输方式：{seg.word()}"}
+    kept = [row for row in raw.splitlines() if row.strip() not in skip]
+    return "\n".join(kept).strip()
+
+
+def _segment_inquiry_block(seg: Segment) -> str:
+    """一段的询价字段正文，去掉运输方式行和单段「去查价」尾句。"""
+    if seg.mode == "AIR":
+        text = copy.air_confirm_card(seg.facts)
+        tail = copy.AIR_CONFIRM_TAIL
+    elif seg.mode == "LAND":
+        text = copy.land_confirm_card(seg.facts)
+        tail = copy.LAND_CONFIRM_TAIL
+    else:
+        rows = copy.inquiry_field_lines(
+            work_order_no="",
+            facts=seg.facts,
+            transport_mode="海运",
+        )
+        kept = [
+            row
+            for row in rows
+            if row and not row.startswith("工单号") and not row.startswith("运输方式")
+        ]
+        return "\n".join(kept)
+    lines = []
+    for row in text.splitlines():
+        if not row.strip() or row.strip() == tail:
+            continue
+        if row.startswith("运输方式"):
+            continue
+        lines.append(row)
+    return "\n".join(lines)
+
+
+def _unlabeled_supplement_facts(
+    text: str,
+    extra_facts: dict[str, str] | None,
+) -> dict[str, str]:
+    """
+    没写段号时的补字段：标签行 + 体积重量包装 + 抽取结果。
+
+    运输类型等选项只认原话枚举，抽取结果不得把「陆运」盖掉已选的国内运输拼车。
+    海运运输分类与单聊相同：原话 Port to Port / 港到港 / 其他 就要收下。
+    """
+    parsed = parse_labeled_facts(text)
+    parsed = harvest_oral_measures(text, parsed)
+    parsed = harvest_land_options(text, parsed)
+    parsed = harvest_sea_class(text, parsed)
+    extra = normalize_facts(dict(extra_facts or {}))
+    for key, val in extra.items():
+        token = str(val or "").strip()
+        if not token or key in _LAND_OPTION_KEYS:
+            continue
+        if not str(parsed.get(key) or "").strip():
+            parsed[key] = token
+    return {key: str(val).strip() for key, val in parsed.items() if str(val or "").strip()}
+
+
+# 核对后只改正文、没写段号时，用这些键判断补到哪一段。
+_SEA_ONLY_KEYS = frozenset({"货量", "货物数量", "整柜或拼柜", "箱型箱量"})
+_LAND_ONLY_KEYS = frozenset({"运输类型", "线路类别", "车型/数量", "通关口岸"})
+_AIR_ONLY_KEYS = frozenset({"件数", "毛重", "体积", "包装方式", "包装类型"})
+
+
+def _indexes_for_supplement(segments: list[Segment], parsed: dict[str, str]) -> list[int]:
+    """
+    没写段号时，按字段判断补到哪一段。
+
+    货物数量/箱型只属于海运；件数/体积只属于空运。对不上或对上两段则空，不猜。
+    """
+    keys = {str(k).strip() for k in parsed if str(k).strip()}
+    if keys & _SEA_ONLY_KEYS:
+        return [seg.index for seg in segments if seg.mode == "SEA"]
+    if keys & _LAND_ONLY_KEYS:
+        return [seg.index for seg in segments if seg.mode == "LAND"]
+    if keys & _AIR_ONLY_KEYS:
+        return [seg.index for seg in segments if seg.mode == "AIR"]
+    if keys & {"起运港", "目的港", "品名", "贸易条款"}:
+        sea_air = [seg.index for seg in segments if seg.mode in {"SEA", "AIR"}]
+        if len(sea_air) == 1:
+            return sea_air
+        land = [seg.index for seg in segments if seg.mode == "LAND"]
+        if not sea_air and len(land) == 1:
+            return land
+    return []
+
+
+def _lacking_indexes(segments: list[Segment]) -> list[int]:
+    """还缺必填的段号。只用来决定口语补句写进哪一段。"""
+    lacking: list[int] = []
+    for seg in segments:
+        ok, _, _ = refresh_segments([seg])
+        if not ok:
+            lacking.append(seg.index)
+    return lacking
+
+
+def _facts_from_body(body: str, mode: str) -> dict[str, str]:
+    """一段正文里的「字段名：值」。陆运再收运输类型、线路类别和销售侧叫法。"""
+    facts = parse_labeled_facts(body)
+    for line in (body or "").splitlines():
+        matched = _EXTRA_LABEL.match(line.strip())
+        if not matched:
+            continue
+        key, val = matched.group(1), matched.group(2).strip()
+        if val and not str(facts.get(key) or "").strip():
+            facts[key] = val
+    if mode == "LAND":
+        facts = harvest_land_options(body, facts)
+    if mode == "SEA":
+        facts = harvest_sea_class(body, facts)
+    return facts
+
+
+def _split_by_segment_mark(text: str) -> dict[int, dict[str, str]]:
+    """
+    补充消息里「段1」「段2」后面的内容。
+
+    除「字段名：值」外，还收陆运选项（如「国内运输拼车」）和包装词（如「托盘」）。
+    没有段号则返回空。
+    """
+    raw = text or ""
+    marks = list(_SEG_MARK.finditer(raw))
+    if not marks:
+        return {}
+    out: dict[int, dict[str, str]] = {}
+    for i, mark in enumerate(marks):
+        end = marks[i + 1].start() if i + 1 < len(marks) else len(raw)
+        body = raw[mark.end() : end]
+        facts = parse_labeled_facts(body)
+        facts = harvest_land_options(body, facts)
+        facts = harvest_sea_class(body, facts)
+        facts = harvest_oral_measures(body, facts)
+        if facts:
+            out[int(mark.group(1))] = facts
+    return out

inquiry-agent/agent/policy/multi_text_flow.py

@@ -0,0 +1,778 @@
+"""
+多段联运私聊：补问、核对、建单、拉群。
+
+本文件职责：一段一段收齐必填后才建一张工单；不去 TMS 查价。
+不含空运时，销售点「拉产品进群」后按各段线路把人拉进同一个群，并发群摘要。
+含空运时只提示手动拉人到原有群。报价和成交跟进不在本文件。
+调用：文字询价 Handler 在 Worker 线程；禁止回调线程同步跑。
+禁止：直连 TMS；用两段同一种运输方式建单；把协同字段当成必填。
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import threading
+from typing import Any, Optional
+
+from agent.policy import inquiry_copy as copy
+from agent.policy.air_text_flow import (
+    FlowSession,
+    ReplyFn,
+    _session_from_payload,
+    _session_payload,
+    build_ledger,
+)
+from agent.policy.multi_segments import (
+    Segment,
+    classify_multi_text,
+    confirm_text,
+    group_brief,
+    includes_air,
+    merge_supplement,
+    refresh_segments,
+    segments_from_payload,
+    segments_to_payload,
+    split_segments,
+)
+from agent.schema.land_options import (
+    looks_like_confirm,
+    option_prompt_needs_sample,
+    sample_pair_allowed,
+)
+
+logger = logging.getLogger(__name__)
+
+# 只有补问/核对还能续同一张多段草稿。已出号的拉群停点不能拦住新询价。
+_FRONT = frozenset({"clarify", "wait_confirm"})
+
+
+class MultiTextInquiryFlow:
+    """多段联运私聊 Owner。一张工单、一个群，段存在书签 quote.multi_segments。"""
+
+    bookmark_kind = "multi"
+
+    def __init__(self, ledger: Any = None, group_client: Any = None, *, bookmark_store: Any = None) -> None:
+        self._ledger = ledger or build_ledger()
+        self._lock = threading.Lock()
+        self._by_sender: dict[str, FlowSession] = {}
+        self._by_ticket: dict[str, FlowSession] = {}
+        self._group_client = group_client
+        self._bookmark_store = bookmark_store
+
+    @property
+    def ledger(self) -> Any:
+        return self._ledger
+
+    def session_of(self, sender_id: str) -> Optional[FlowSession]:
+        """当前销售的多段书签。进程内没有就读 Redis 书签。"""
+        sid = (sender_id or "").strip()
+        with self._lock:
+            hit = self._by_sender.get(sid)
+        if hit is not None:
+            return hit
+        loaded = self._load_bookmark(sid)
+        if loaded is None:
+            return None
+        self._save(loaded)
+        return loaded
+
+    def session_for_action(
+        self,
+        sender_id: str,
+        card_meta: Optional[dict[str, Any]] = None,
+        *,
+        text: str = "",
+    ) -> Optional[FlowSession]:
+        """
+        点卡：用工单号找回这一单。HTTP 进程往往没有 Worker 内存书签。
+
+        卡上有工单号时，不对上就不拿当前草稿，避免串到别人/最新单。
+        进程刚重启再从主账收回 MULTI 工单。
+        """
+        sid = (sender_id or "").strip()
+        if not sid:
+            return None
+        no = copy.work_order_from_card_meta(card_meta, text=text)
+        if no:
+            with self._lock:
+                hit = self._by_ticket.get(no)
+                if hit and hit.sender_id == sid:
+                    return hit
+        sess = self.session_of(sid)
+        if sess and (sess.business_line or "").upper() == "MULTI":
+            if not no:
+                return sess
+            if (sess.work_order_no or "").strip() == no:
+                return sess
+        if no:
+            return self._load_ticket_session(sid, no)
+        return None
+
+    def on_button(
+        self,
+        *,
+        sender_id: str,
+        action: str,
+        reply: ReplyFn,
+        card_meta: Optional[dict[str, Any]] = None,
+    ) -> str:
+        """
+        处理多段拉群卡。错工单/未允许直接丢弃，不改主账。
+
+        副作用：点「拉产品进群协同」时建一个群并发摘要、各段模板。
+        调用：卡片 Handler，HTTP 回调线程只入队后由 Worker/本流程执行；本函数会打企微建群。
+        """
+        sess = self.session_for_action(sender_id, card_meta)
+        card_wo = copy.work_order_from_card_meta(card_meta)
+        if not sess or action not in (sess.allowed or ()):
+            logger.info(
+                "multi card 丢弃 sender=%s action=%s card_wo=%s sess_wo=%s phase=%s",
+                sender_id,
+                action,
+                card_wo or "-",
+                (sess.work_order_no if sess else "-"),
+                (sess.phase if sess else "-"),
+            )
+            return "card_discard"
+        if card_wo and sess.work_order_no and card_wo != sess.work_order_no:
+            logger.warning(
+                "multi card 工单不一致 card_wo=%s sess_wo=%s，拒绝串单",
+                card_wo,
+                sess.work_order_no,
+            )
+            return "card_discard"
+        if action == copy.BTN_PULL_COLLAB:
+            return self._pull_group(sess, reply)
+        return "card_discard"
+
+    def drop_current_bookmark(self, sender_id: str) -> Optional[FlowSession]:
+        """销售改发单段时丢掉未完成的多段草稿。已出号的仍留在工单号索引。"""
+        sid = (sender_id or "").strip()
+        if not sid:
+            return None
+        with self._lock:
+            old = self._by_sender.pop(sid, None)
+        self._delete_bookmark(sid)
+        return old
+
+    def match_button(self, text: str, sess: Optional[FlowSession]) -> str:
+        """拉群按钮。没到可点阶段就不当按钮。"""
+        if sess is None:
+            return ""
+        action = copy.canonical_button(text)
+        if action == copy.BTN_PULL_COLLAB and action in (sess.allowed or ()):
+            return action
+        return ""
+
+    def offer_handoff(self, *, sender_id: str, text: str, reply: ReplyFn) -> str:
+        """多段还在收字段时不转人工，请销售先把各段发齐。"""
+        _ = sender_id, text
+        reply("多段询价请先把各段需求发齐并回复确定。需要人来接管时，请在工单创建后再说。")
+        return "multi_handoff_later"
+
+    def on_text(
+        self,
+        *,
+        sender_id: str,
+        text: str,
+        reply: ReplyFn,
+        injected_facts: Optional[dict[str, Any]] = None,
+        injected_mode: str = "",
+        b_result: Optional[dict[str, Any]] = None,
+        allow_b: bool = False,
+        chat_fn: Any = None,
+        inbound_raw: Optional[dict[str, Any]] = None,
+    ) -> str:
+        """
+        处理一条多段私聊。
+
+        副作用：写书签、可能建主账工单、点拉群时建企微群。不查 TMS。
+        新开多段仍按原文切段；补问中的口语句走 injected_* / DeepSeek B，与单聊同一套抽取。
+        已出号等拉群时，再发一段陆运+海运视为新询价，不能回旧工单的手动拉群提示。
+        """
+        _ = inbound_raw
+        sid = (sender_id or "").strip()
+        raw = text or ""
+        sess = self.session_of(sid)
+        if sess and self.match_button(raw, sess):
+            return self._pull_group(sess, reply)
+        verdict = classify_multi_text(raw)
+        if verdict == "same_mode":
+            reply(copy.MULTI_SAME_MODE)
+            return "reject_same_mode"
+        if verdict == "multi":
+            # 补问里再写段1/段2是补字段，不能当新开把已填冲掉。
+            if sess and (sess.phase or "") == "clarify":
+                return self._continue_clarify(
+                    sess,
+                    raw,
+                    reply,
+                    injected_facts=injected_facts,
+                    injected_mode=injected_mode,
+                    b_result=b_result,
+                    allow_b=allow_b,
+                    chat_fn=chat_fn,
+                )
+            return self._open(
+                sid,
+                split_segments(raw).segments,
+                reply,
+                immutable=raw,
+                allow_b=allow_b,
+                chat_fn=chat_fn,
+            )
+        if sess and sess.phase == "wait_confirm" and looks_like_confirm(raw):
+            return self._create_ticket(sess, reply)
+        if sess and sess.phase in {"wait_collab", "collab_group"}:
+            if (sess.collab_chat_id or "").strip():
+                reply(copy.sea_group_exists(work_order_no=sess.work_order_no))
+                return "collab_group"
+            reply(copy.multi_pull_text(work_order_no=sess.work_order_no))
+            return "wait_collab"
+        if sess and sess.phase == "wait_manual_group":
+            reply(copy.multi_air_manual_hint(work_order_no=sess.work_order_no))
+            return "wait_manual_group"
+        if sess and sess.phase in _FRONT and not looks_like_confirm(raw):
+            if sess.phase == "clarify":
+                return self._continue_clarify(
+                    sess,
+                    raw,
+                    reply,
+                    injected_facts=injected_facts,
+                    injected_mode=injected_mode,
+                    b_result=b_result,
+                    allow_b=allow_b,
+                    chat_fn=chat_fn,
+                )
+            if sess.phase == "wait_confirm":
+                # 核对卡写了「内容有误请重新发」：销售会只改正字段，按补问叠上去。
+                return self._continue_clarify(
+                    sess,
+                    raw,
+                    reply,
+                    injected_facts=injected_facts,
+                    injected_mode=injected_mode,
+                    b_result=b_result,
+                    allow_b=allow_b,
+                    chat_fn=chat_fn,
+                )
+        if sess and sess.phase == "wait_confirm":
+            reply(copy.MULTI_CONFIRM_EXPIRED)
+            return "multi_confirm_expired"
+        reply(copy.MULTI_SAME_MODE if verdict == "same_mode" else "请按段发多段需求，例如段1海运、段2空运。")
+        return "multi_need_text"
+
+    def _open(
+        self,
+        sender_id: str,
+        segments: list[Segment],
+        reply: ReplyFn,
+        *,
+        immutable: str,
+        allow_b: bool = False,
+        chat_fn: Any = None,
+    ) -> str:
+        """
+        新的多段原文：缺必填就一次说完，齐了出核对。不建单。
+
+        附件口播经常没有「起运港：」这种标签，要按段走和单聊一样的抽取，
+        不能只认标签行，否则会整段待补充。
+        """
+        filled = self._enrich_segments(segments, allow_b=allow_b, chat_fn=chat_fn)
+        ready, updated, gaps = refresh_segments(filled)
+        if not ready:
+            self._reply_gaps(updated, gaps, reply)
+            self._save(self._draft(sender_id, updated, phase="clarify", immutable=immutable))
+            return "clarify"
+        reply(confirm_text(updated))
+        self._save(self._draft(sender_id, updated, phase="wait_confirm", immutable=immutable))
+        return "wait_confirm"
+
+    def _reply_gaps(self, segments: list[Segment], gaps: str, reply: ReplyFn) -> None:
+        """
+        补问正文。陆运还缺类型或线路时，再发一张和单聊相同的填写样例图。
+
+        副作用：出站文字，可能再出一张图片。只在 Worker 里调。
+        """
+        reply(gaps)
+        for seg in segments:
+            if seg.mode != "LAND":
+                continue
+            if option_prompt_needs_sample(current=seg.facts):
+                reply(copy.land_sample_image_fallback(), copy.land_sample_image_payload())
+                return
+
+    def _enrich_segments(
+        self,
+        segments: list[Segment],
+        *,
+        allow_b: bool,
+        chat_fn: Any,
+    ) -> list[Segment]:
+        """
+        各段正文再抽一遍字段。标签行已经在切段时收过，这里只补口语/附件里没写冒号的值。
+
+        没有正文就不调模型，避免补问合并后空跑。
+        """
+        if not segments:
+            return segments
+        if not allow_b and chat_fn is None:
+            return segments
+        from agent.llm.extract_text import extract_inquiry_snapshot
+
+        out: list[Segment] = []
+        for seg in segments:
+            body = (seg.source or "").strip()
+            if not body:
+                out.append(seg)
+                continue
+            snap = extract_inquiry_snapshot(
+                body,
+                injected_mode=seg.mode,
+                allow_b=allow_b,
+                chat_fn=chat_fn,
+            )
+            extra = {
+                str(key): str(val).strip()
+                for key, val in dict(snap.get("facts") or {}).items()
+                if str(val or "").strip()
+            }
+            subtype = str(snap.get("land_subtype") or "").strip()
+            if seg.mode == "LAND" and subtype and not str(extra.get("运输类型") or "").strip():
+                extra["运输类型"] = subtype
+            facts = dict(seg.facts)
+            for key, val in extra.items():
+                if not str(facts.get(key) or "").strip():
+                    facts[key] = val
+            out.append(
+                Segment(index=seg.index, mode=seg.mode, facts=facts, source=seg.source)
+            )
+        return out
+
+    def _continue_clarify(
+        self,
+        sess: FlowSession,
+        text: str,
+        reply: ReplyFn,
+        *,
+        injected_facts: Optional[dict[str, Any]] = None,
+        injected_mode: str = "",
+        b_result: Optional[dict[str, Any]] = None,
+        allow_b: bool = False,
+        chat_fn: Any = None,
+    ) -> str:
+        """
+        补问中的下一条：能整段重发就替换，否则按段号或唯一缺口段补上。
+
+        口语补句走单聊同一套抽取，不在这里猜「广州-深圳」。
+        """
+        from agent.llm.extract_text import extract_inquiry_snapshot
+        from agent.policy.system_exception import handle_extract_tech_fail
+
+        snap = extract_inquiry_snapshot(
+            text,
+            injected_facts=injected_facts,
+            injected_mode=injected_mode,
+            b_result=b_result,
+            allow_b=allow_b,
+            chat_fn=chat_fn,
+        )
+        llm_phase = handle_extract_tech_fail(
+            snap=snap,
+            ledger=self._ledger,
+            reply=reply,
+            sender_id=sess.sender_id,
+            sess=sess,
+        )
+        if llm_phase:
+            return llm_phase
+        extra = {
+            str(key): str(val).strip()
+            for key, val in dict(snap.get("facts") or {}).items()
+            if str(val or "").strip()
+        }
+        merged = merge_supplement(self._segments(sess), text, extra_facts=extra)
+        return self._open(
+            sess.sender_id,
+            merged,
+            reply,
+            immutable=sess.immutable_text or text,
+            allow_b=allow_b,
+            chat_fn=chat_fn,
+        )
+
+    def _restart(self, sender_id: str, text: str, reply: ReplyFn) -> str:
+        """核对之后没有回复确定，而是重发内容：整段重来，不在旧核对上改一个字。"""
+        verdict = classify_multi_text(text)
+        if verdict == "same_mode":
+            reply(copy.MULTI_SAME_MODE)
+            return "reject_same_mode"
+        if verdict != "multi":
+            reply("请把多段需求再发一次。")
+            return "wait_confirm"
+        return self._open(sender_id, split_segments(text).segments, reply, immutable=text)
+
+    def _create_ticket(self, sess: FlowSession, reply: ReplyFn) -> str:
+        """回复确定后建一张 MULTI 工单。含空运只给手动拉群提示，不含空运出拉群卡。"""
+        segments = self._segments(sess)
+        ready, updated, gaps = refresh_segments(segments)
+        if not ready:
+            self._reply_gaps(updated, gaps, reply)
+            sess.phase = "clarify"
+            self._put_segments(sess, updated)
+            self._save(sess)
+            return "clarify"
+        payload = segments_to_payload(updated)
+        created = self._ledger.create_ticket(
+            sender_id=sess.sender_id,
+            business_line="MULTI",
+            facts={"segments_json": json.dumps(payload, ensure_ascii=False)},
+            idempotency_key=f"multi:{sess.sender_id}:{sess.wait_version}",
+        )
+        if not created.get("ok"):
+            reply("工单创建失败，请稍后再回复确定。")
+            return "wait_confirm"
+        no = str(created.get("work_order_no") or "").strip()
+        sess.work_order_no = no
+        sess.status = "询价中"
+        self._put_segments(sess, updated)
+        if includes_air(updated):
+            sess.phase = "wait_manual_group"
+            sess.allowed = ()
+            self._save(sess)
+            reply(copy.multi_air_manual_hint(work_order_no=no))
+            logger.info("multi.created air wo=%s segments=%s", no, len(updated))
+            return "wait_manual_group"
+        sess.phase = "wait_collab"
+        sess.allowed = (copy.BTN_PULL_COLLAB,)
+        self._save(sess)
+        reply(copy.multi_pull_text(work_order_no=no), copy.multi_pull_wecom_payload(work_order_no=no))
+        logger.info("multi.created pull wo=%s segments=%s", no, len(updated))
+        return "wait_collab"
+
+    def _pull_group(self, sess: FlowSession, reply: ReplyFn) -> str:
+        """
+        点拉群：海运段按港口拉海运产品，陆运段按线路拉陆运产品，进同一个群。
+
+        任一段没有人，或陆运类型+线路不在填写样例里，不建群，按钮仍可点。
+        """
+        no = (sess.work_order_no or "").strip()
+        if (sess.collab_chat_id or "").strip():
+            reply(copy.sea_group_exists(work_order_no=no))
+            return "collab_group"
+        segments = self._segments(sess)
+        if includes_air(segments):
+            reply(copy.multi_air_manual_hint(work_order_no=no))
+            return "wait_manual_group"
+        for seg in segments:
+            if seg.mode != "LAND":
+                continue
+            land_type = str(seg.facts.get("运输类型") or "").strip()
+            route = str(seg.facts.get("线路类别") or "").strip()
+            if not sample_pair_allowed(land_type, route):
+                reply(copy.land_group_pair_rejected())
+                return sess.phase or "wait_collab"
+        userids = [sess.sender_id]
+        product_names: list[str] = []
+        product_ids: list[str] = []
+        for seg in segments:
+            staff = self._staff_for(seg)
+            if not staff:
+                reply(copy.land_group_no_staff() if seg.mode == "LAND" else copy.sea_group_no_staff())
+                return sess.phase or "wait_collab"
+            for row in staff:
+                uid = str(row.get("wecomId") or row.get("wecom_id") or "").strip()
+                name = str(row.get("name") or "").strip() or uid
+                if not uid:
+                    continue
+                if uid not in userids:
+                    userids.append(uid)
+                if uid not in product_ids:
+                    product_ids.append(uid)
+                    product_names.append(name)
+        if not product_ids:
+            reply(copy.sea_group_no_staff())
+            return sess.phase or "wait_collab"
+        for uid in self._archive_seats():
+            if uid not in userids:
+                userids.append(uid)
+        client = self._client()
+        from agent.policy.system_exception import call_with_retries
+
+        created = call_with_retries(
+            lambda: client.create_group(name=no, userids=userids) or {},
+            is_ok=lambda x: bool((x or {}).get("ok") and (x or {}).get("chat_id")),
+        )
+        if not created.get("ok") or not created.get("chat_id"):
+            reply(copy.sea_group_create_fail(reason=str(created.get("error") or "")))
+            return sess.phase or "wait_collab"
+        chat_id = str(created.get("chat_id") or "").strip()
+        sales_name = self._sales_name(sess.sender_id)
+        binder = getattr(self._ledger, "bind_collab_group", None)
+        if callable(binder):
+            binder(
+                work_order_no=no,
+                chat_id=chat_id,
+                member_ids=userids,
+                product_names=product_names,
+                product_ids=product_ids,
+            )
+        sess.collab_chat_id = chat_id
+        sess.phase = "collab_group"
+        sess.allowed = ()
+        sess.invited_product_names = tuple(product_names)
+        sess.invited_sales_name = sales_name
+        self._save(sess)
+        from agent.policy.sea_group_ops import send_group_text
+
+        brief = group_brief(
+            work_order_no=no,
+            segments=segments,
+            sales_name=sales_name,
+            mention_names=product_names,
+        )
+        send_group_text(client, chat_id, brief, mention_userids=product_ids)
+        ticket = self._ledger.get_ticket(work_order_no=no) if hasattr(self._ledger, "get_ticket") else None
+        if ticket is not None:
+            from agent.policy.multi_group_ops import send_segment_templates
+
+            send_segment_templates(flow=self, ticket=ticket, chat_id=chat_id)
+        reply(
+            copy.sea_group_created(
+                work_order_no=no,
+                sales_name=sales_name,
+                product_names=product_names,
+            )
+        )
+        logger.info("multi.group wo=%s chat=%s products=%s", no, chat_id, product_names)
+        return "collab_group"
+
+    def _staff_for(self, seg: Segment) -> list[dict[str, Any]]:
+        """按这一段自己的线路找产品。空运段不在这里拉人。"""
+        if seg.mode == "SEA":
+            matcher = getattr(self._ledger, "match_sea_staff", None)
+            if not callable(matcher):
+                return []
+            try:
+                rows = matcher(
+                    origin=str(seg.facts.get("起运港") or ""),
+                    destination=str(seg.facts.get("目的港") or ""),
+                ) or []
+            except Exception:
+                logger.exception("多段匹配海运产品失败")
+                return []
+            return [dict(x) for x in rows if isinstance(x, dict)]
+        if seg.mode == "LAND":
+            matcher = getattr(self._ledger, "match_land_staff", None)
+            if not callable(matcher):
+                return []
+            try:
+                rows = matcher(route_category=str(seg.facts.get("线路类别") or "")) or []
+            except Exception:
+                logger.exception("多段匹配陆运产品失败")
+                return []
+            return [dict(x) for x in rows if isinstance(x, dict)]
+        return []
+
+    def _client(self) -> Any:
+        """建群客户端。单测注入；生产与海运共用企微应用客户端。"""
+        if self._group_client is not None:
+            return self._group_client
+        from agent.policy.sea_text_flow import _default_group_client
+
+        self._group_client = _default_group_client()
+        return self._group_client
+
+    def _archive_seats(self) -> list[str]:
+        """会话存档账号。查本账本，没有也不挡建群。"""
+        from agent.config import get_settings
+        from agent.policy.sea_text_flow import ARCHIVE_SEAT_NAMES
+
+        ids: list[str] = []
+        seat = str(getattr(get_settings(), "wecom_archive_seat_user_ids", "") or "").strip()
+        for part in seat.replace(";", ",").split(","):
+            uid = part.strip()
+            if uid and uid not in ids:
+                ids.append(uid)
+        finder = getattr(self._ledger, "find_staff_by_name", None)
+        if not callable(finder):
+            return ids
+        for name in ARCHIVE_SEAT_NAMES:
+            try:
+                row = finder(name=name) or {}
+            except Exception:
+                logger.exception("多段查会话存档账号失败 name=%s", name)
+                continue
+            uid = str(row.get("wecomId") or row.get("wecom_id") or "").strip()
+            if uid and uid not in ids:
+                ids.append(uid)
+        return ids
+
+    def _sales_name(self, sender_id: str) -> str:
+        """群里 @ 的销售名。主账没有姓名时退回 userid。"""
+        finder = getattr(self._ledger, "get_staff_by_wecom_id", None)
+        if callable(finder):
+            try:
+                row = finder(wecom_id=sender_id) or {}
+            except Exception:
+                row = {}
+            name = str(row.get("name") or "").strip()
+            if name:
+                return name
+        return (sender_id or "").strip() or "销售"
+
+    def _draft(self, sender_id: str, segments: list[Segment], *, phase: str, immutable: str) -> FlowSession:
+        sess = FlowSession(
+            sender_id=sender_id,
+            thread_id=sender_id,
+            phase=phase,
+            business_line="MULTI",
+            immutable_text=immutable,
+            wait_version=1,
+        )
+        self._put_segments(sess, segments)
+        return sess
+
+    def _segments(self, sess: FlowSession) -> list[Segment]:
+        return segments_from_payload((sess.quote or {}).get("multi_segments"))
+
+    def _put_segments(self, sess: FlowSession, segments: list[Segment]) -> None:
+        quote = dict(sess.quote or {})
+        quote["multi_segments"] = segments_to_payload(segments)
+        sess.quote = quote
+
+    def _load_ticket_session(self, sender_id: str, work_order_no: str) -> Optional[FlowSession]:
+        """
+        点卡时本进程没有书签：只读主账收回这一张 MULTI 工单。
+
+        找不到、不是本人、不是多段：返回 None，绝不改成当前最新单或空运停点。
+        """
+        getter = getattr(self._ledger, "get_for_agent", None)
+        if not callable(getter):
+            logger.warning("multi card 点到 %s 但无主账可读", work_order_no)
+            return None
+        view = getter(work_order_no=work_order_no, sender_id=sender_id) or {}
+        if not view.get("found") or view.get("owned") is False:
+            logger.warning(
+                "multi card 点到 %s 主账未收回 found=%s owned=%s",
+                work_order_no,
+                view.get("found"),
+                view.get("owned"),
+            )
+            return None
+        if (str(view.get("business_line") or "")).upper() != "MULTI":
+            return None
+        sess = self._session_from_ledger(sender_id=sender_id, view=view)
+        self._save(sess)
+        logger.info("multi card 从主账收回 wo=%s phase=%s", sess.work_order_no, sess.phase)
+        return sess
+
+    def _session_from_ledger(self, *, sender_id: str, view: dict[str, Any]) -> FlowSession:
+        """
+        主账快照收成多段书签。询价中/被空运误标成 tms_miss 都按等拉群处理。
+
+        段列表优先 quote.multi_segments，没有就解 facts.segments_json。
+        """
+        no = str(view.get("work_order_no") or "").strip()
+        facts = dict(view.get("facts") or {})
+        quote = dict(view.get("quote") or {})
+        raw_segs = quote.get("multi_segments")
+        if not raw_segs:
+            blob = facts.get("segments_json") or facts.get("segmentsJson") or ""
+            if isinstance(blob, str) and blob.strip():
+                try:
+                    parsed = json.loads(blob)
+                except json.JSONDecodeError:
+                    parsed = None
+                if isinstance(parsed, list):
+                    quote["multi_segments"] = parsed
+                    raw_segs = parsed
+        segments = segments_from_payload(raw_segs)
+        status = str(view.get("status") or "").strip()
+        chat = str(view.get("collab_chat_id") or view.get("collabChatId") or "").strip()
+        phase = str(view.get("wait_phase") or view.get("waitPhase") or "").strip()
+        allowed: tuple[str, ...] = ()
+        if chat:
+            phase = "collab_group"
+        elif includes_air(segments):
+            phase = "wait_manual_group"
+        elif status in {"已成交", "已关闭"}:
+            phase = "done"
+        elif status == "未成交":
+            phase = "wait_lost_reason"
+        elif status == "协商中" or phase == "wait_deal":
+            phase = "wait_deal"
+            allowed = (copy.BTN_DEAL, copy.BTN_LOST, copy.BTN_NEGOTIATE)
+        else:
+            # 询价中、tms_miss、wait_collab：销售还没拉群
+            phase = "wait_collab"
+            allowed = (copy.BTN_PULL_COLLAB,)
+        return FlowSession(
+            sender_id=sender_id,
+            thread_id=sender_id,
+            phase=phase,
+            business_line="MULTI",
+            facts={str(k): str(v) for k, v in facts.items()},
+            work_order_no=no,
+            quote=quote,
+            status=status,
+            allowed=allowed,
+            wait_version=1,
+            invited_sales_name=str(view.get("sales_name") or view.get("salesName") or ""),
+            collab_chat_id=chat,
+        )
+
+    def _save(self, sess: FlowSession) -> None:
+        with self._lock:
+            self._by_sender[sess.sender_id] = sess
+            no = (sess.work_order_no or "").strip()
+            if no:
+                self._by_ticket[no] = sess
+        self._persist_bookmark(sess)
+
+    def _persist_bookmark(self, sess: FlowSession) -> None:
+        from agent.redis_coord.flow_bookmark import save_flow_bookmark
+
+        save_flow_bookmark(
+            self.bookmark_kind,
+            sess.sender_id,
+            _session_payload(sess),
+            store=self._bookmark_store,
+        )
+
+    def _load_bookmark(self, sender_id: str) -> Optional[FlowSession]:
+        from agent.redis_coord.flow_bookmark import load_flow_bookmark
+
+        raw = load_flow_bookmark(self.bookmark_kind, sender_id, store=self._bookmark_store)
+        if not raw:
+            return None
+        return _session_from_payload(raw)
+
+    def _delete_bookmark(self, sender_id: str) -> None:
+        from agent.redis_coord.flow_bookmark import delete_flow_bookmark
+
+        delete_flow_bookmark(self.bookmark_kind, sender_id, store=self._bookmark_store)
+
+
+_FLOW: Optional[MultiTextInquiryFlow] = None
+_FLOW_GUARD = threading.Lock()
+
+
+def get_multi_text_flow() -> MultiTextInquiryFlow:
+    """进程内一份多段流程。单测请自己 new，不要碰这份单例。"""
+    global _FLOW
+    with _FLOW_GUARD:
+        if _FLOW is None:
+            _FLOW = MultiTextInquiryFlow()
+        return _FLOW
+
+
+def reset_multi_text_flow_for_test() -> MultiTextInquiryFlow:
+    """单测重置全局实例（强制内存账本）。"""
+    global _FLOW
+    from agent.ledger.memory_ledger import MemoryLedger
+
+    with _FLOW_GUARD:
+        _FLOW = MultiTextInquiryFlow(MemoryLedger())
+        return _FLOW

inquiry-agent/agent/policy/sea_group_ops.py

@@ -286,18 +286,21 @@ def resolve_group_mentions(
     flow: Any,
     ticket: object,
     sess: Any = None,
+    fallback_sales_id: str = "",
 ) -> tuple[str, list[str], list[str]]:
     """
     群摘要必须 @ 到具体销售和每位产品，禁止只写岗位词「销售」。
 
-    进程重启后内存会话没了，主账 byCollabChat 也可能不带姓名：
-    用销售 userid + 群成员（去掉存档账号）反查员工管理。
+    进程重启后内存会话没了，主账 byCollabChat / getForAgent 也可能不带姓名或企微号：
+    用销售 userid（没有则用发言人）+ 群成员（去掉存档账号）反查员工管理。
+    查不到姓名就用 userid，仍然是具体人。
     返回：(销售展示名, 产品展示名列表, 要点到的 userid)。
     """
     view = ticket_view(ticket)
     sales_id = str(view.get("sales_wecom_id") or view.get("salesWecomId") or "").strip()
     if sess is not None:
         sales_id = sales_id or str(getattr(sess, "sales_wecom_id", "") or getattr(sess, "sender_id", "") or "").strip()
+    sales_id = sales_id or (fallback_sales_id or "").strip()
     members = [
         str(x).strip()
         for x in (view.get("collab_member_ids") or view.get("memberIds") or [])
@@ -323,10 +326,10 @@ def resolve_group_mentions(
     sales_name = sales_name or str(view.get("sales_name") or view.get("salesName") or "").strip()
     if not sales_name or sales_name == "销售":
         looked = _staff_display_name(flow, sales_id) if sales_id else ""
-        if looked:
+        if looked and looked != "销售":
             sales_name = looked
-    if not sales_name:
-        sales_name = sales_id or "销售"
+    if not sales_name or sales_name == "销售":
+        sales_name = sales_id
 
     products: list[str] = []
     if sess is not None:
@@ -383,7 +386,19 @@ def hydrate_ticket(flow: Any, ticket: object) -> object:
     facts = dict(view.get("facts") or {})
     has_facts = any(
         str(facts.get(k) or "").strip()
-        for k in ("起运港", "目的港", "品名", "货量", "起运地", "目的地", "件数", "毛重")
+        for k in (
+            "起运港",
+            "目的港",
+            "品名",
+            "货量",
+            "起运地",
+            "目的地",
+            "件数",
+            "毛重",
+            "运输类型",
+            "始发站",
+            "目的站",
+        )
     )
     quote = dict(view.get("quote") or {})
     has_quote = bool(quote.get("total") or _quote_fee_rows(quote))
@@ -515,6 +530,28 @@ def send_brief_and_sheet(
     return "brief_and_sheet"
 
 
+def _has_sheet_inquiry_facts(facts: dict[str, Any]) -> bool:
+    """段票/填表票是否已带可匹配模板的询价字段。"""
+    src = dict(facts or {})
+    return any(
+        str(src.get(k) or "").strip()
+        for k in (
+            "起运港",
+            "目的港",
+            "品名",
+            "货量",
+            "起运地",
+            "目的地",
+            "件数",
+            "毛重",
+            "运输类型",
+            "线路类别",
+            "始发站",
+            "目的站",
+        )
+    )
+
+
 def send_filled_sheet(
     *,
     flow: Any,
@@ -529,6 +566,8 @@ def send_filled_sheet(
 
     business_line 决定匹配哪套后台模板；空运必须传 AIR，禁止落到海运默认模板。
     """
+    incoming_view = ticket_view(ticket) if ticket is not None else {}
+    incoming_facts = dict(incoming_view.get("facts") or {})
     ticket = hydrate_ticket(flow, ticket)
     view = ticket_view(ticket) if ticket is not None else {}
     no = (sess.work_order_no if sess else "") or ticket_no(ticket)
@@ -539,6 +578,11 @@ def send_filled_sheet(
         or ""
     ).strip()
     facts = dict(view.get("facts") or {})
+    # 段票已带询价字段时，禁止被主账 MULTI 快照（往往只有 segments_json）盖掉。
+    if _has_sheet_inquiry_facts(incoming_facts):
+        merged = dict(facts)
+        merged.update({k: v for k, v in incoming_facts.items() if str(v or "").strip()})
+        facts = merged
     if sess is not None and getattr(sess, "facts", None):
         facts.update(dict(sess.facts or {}))
     collab = dict(view.get("collab_facts") or {})

inquiry-agent/agent/quote_templates/match.py

@@ -174,11 +174,35 @@ _GENERIC_KEYWORDS = frozenset(
     {"海运", "空运", "陆运", "sea", "air", "land", "国内"}
 )
 
+# 后台陆运模板常用「国内拼车」，询价字段是「国内运输拼车」，不能当两回事。
+_LAND_KEYWORD_ALIASES = {
+    "国内拼车": ("国内运输拼车", "拼车"),
+    "国内零担": ("国内运输拼车", "国内长途/零担", "零担"),
+    "国内整车": ("国内运输整车", "整车"),
+    "跨境整车": ("跨境整车",),
+    "跨境集拼": ("跨境集拼",),
+    "跨境零担": ("跨境集拼",),
+}
+
+
+def _keyword_needles(kw: str) -> list[str]:
+    """关键词本身 + 陆运别名，都小写。"""
+    raw = (kw or "").strip()
+    if not raw:
+        return []
+    out = [raw.lower()]
+    for alias in _LAND_KEYWORD_ALIASES.get(raw, ()):
+        token = alias.lower()
+        if token not in out:
+            out.append(token)
+    return out
+
 
 def _keyword_score(raw: Any, haystack: str, type_val: str) -> tuple[int, int]:
     """
     运输类型上的命中优先，其次全文词长。
     「陆运/国内」跳过，避免中港、国内整车抢走拼车单。
+    「国内拼车」要对上「国内运输拼车」。
     """
     type_hits = 0
     length = 0
@@ -187,11 +211,12 @@ def _keyword_score(raw: Any, haystack: str, type_val: str) -> tuple[int, int]:
         k = kw.lower()
         if k in _GENERIC_KEYWORDS:
             continue
-        if type_val and k in type_val:
+        needles = _keyword_needles(kw)
+        if type_val and any(n in type_val for n in needles):
             type_hits += 1
             length += len(k)
             any_hit = True
-        elif k in haystack:
+        elif any(n in haystack for n in needles):
             length += len(k)
             any_hit = True
     if not any_hit:

inquiry-agent/agent/routing/dispatch.py

@@ -162,7 +162,8 @@ HANDLER_REGISTRY: dict[str, HandlerFn] = {
     "handoff_human": _wrap_text_inquiry,
     "attachment_inquiry": _wrap_attachment,
     "image_inquiry": _wrap_image,
-    "multi_segment_transport": _wrap_new,
+    # 多段联运与普通文字询价同一条闭环；禁止再走 new_inquiry 空壳（销售补字段会没回音）。
+    "multi_segment_transport": _wrap_text_inquiry,
     "tms_read_only_quote": _wrap_continue,
     "new_inquiry": _wrap_new,
     "continue_thread": _wrap_continue,

inquiry-agent/agent/schema/field_validate.py

@@ -59,6 +59,9 @@ SEA_DISPLAY = {key: label for key, label in SEA_REQUIRED_FIELDS}
 
 # 陆运销售侧名称：卡片写始发站/货物品名；内部仍用合同键起运港/品名。
 LAND_DISPLAY = {
+    "运输类型": "运输类型",
+    "线路类别": "线路类别",
+    "运输分类": "运输分类",
     "起运港": "始发站",
     "目的港": "目的地",
     "品名": "货物品名",
@@ -208,6 +211,7 @@ _ALIASES = {
     "货好时间": "货好时间",
     "提货地址": "提货地址",
     "收货地址": "收货地址",
+    "送货地址": "收货地址",
     "贸易条款": "贸易条款",
     "货源地（非必填）": "货源地",
     "货好时间（非必填）": "货好时间",
@@ -234,6 +238,7 @@ def normalize_facts(facts: dict[str, Any] | None) -> dict[str, str]:
     out: dict[str, str] = {}
     for raw_key, raw_val in (facts or {}).items():
         key = str(raw_key).strip()
+        key = re.sub(r"（必填）|（非必填）|\(必填\)|\(非必填\)", "", key).strip()
         key = _ALIASES.get(key, key)
         if raw_val is None:
             continue
@@ -528,6 +533,9 @@ def validate_required_fields(
             "business_line": line,
         }
     if line == "LAND":
+        from agent.schema.land_options import scrub_land_placeholder_ports
+
+        normalized = scrub_land_placeholder_ports(normalized)
         land_subtype = (land_subtype or "").strip() or str(
             normalized.get("运输类型") or ""
         ).strip()

inquiry-agent/agent/schema/land_options.py

@@ -33,6 +33,26 @@ LAND_CLASSES = (
     "急件",
 )
 
+# 报价单模板码，不能当始发站/目的地。
+_TEMPLATE_PORT_CODES = frozenset({"DOM", "FTL", "LTL", "INTL"})
+
+
+def scrub_land_placeholder_ports(facts: dict[str, str] | None) -> dict[str, str]:
+    """
+    丢掉报价模板里的 DOM/FTL 这类站点码。
+
+    这些不是城市或港口，当已填会让多段误出确定卡。返回新 dict，不改入参。
+    """
+    out: dict[str, str] = {}
+    for key, raw in dict(facts or {}).items():
+        val = str(raw).strip()
+        if not val:
+            continue
+        if key in {"起运港", "目的港", "始发站", "目的地"} and val.upper() in _TEMPLATE_PORT_CODES:
+            continue
+        out[str(key)] = val
+    return out
+
 # 填写样例，同时作为拉群门禁：对不上的类型+线路可以查价，但不能拉群。
 SAMPLE_PAIRS = (
     ("国内运输拼车", "国内长途/零担"),
@@ -66,7 +86,9 @@ _ROUTE_ALIASES = (
     ("国内长途/零担", "国内长途/零担"),
     ("中港/中亚/中欧", "中港/中亚/中欧"),
     ("国内城配", "国内城配/拖车/打包"),
+    ("国内长途/零担", "国内长途/零担"),
     ("国内长途", "国内长途/零担"),
+    ("国内零担", "国内长途/零担"),
     ("东南亚", "东南亚"),
 )
 

inquiry-agent/agent/schema/sea_options.py

@@ -54,8 +54,15 @@ def harvest_sea_class(text: str, current: dict[str, str] | None = None) -> dict[
     kept = canonicalize_sea_class(str(merged.get("运输分类") or ""))
     if kept:
         merged["运输分类"] = kept
-    else:
-        merged.pop("运输分类", None)
+        return merged
+    from agent.schema.land_options import canonicalize_land_class
+
+    # 多段陆运补「急件」时，不能按海运名单把已收的陆运分类清掉。
+    land_kept = canonicalize_land_class(str(merged.get("运输分类") or ""))
+    if land_kept:
+        merged["运输分类"] = land_kept
+        return merged
+    merged.pop("运输分类", None)
     return merged
 
 

inquiry-agent/tests/test_attachment_inquiry.py

@@ -35,6 +35,7 @@ from agent.policy.air_text_flow import reset_air_text_flow_for_test
 from agent.policy.attachment_wave import get_attach_wave_book, reset_attach_wave_book_for_test
 from agent.policy.image_wave import reset_image_wave_book_for_test
 from agent.policy.land_text_flow import reset_land_text_flow_for_test
+from agent.policy.multi_text_flow import get_multi_text_flow, reset_multi_text_flow_for_test
 from agent.policy.sea_text_flow import get_sea_text_flow, reset_sea_text_flow_for_test
 from agent.routing.decision import RouteDecision
 from agent.routing.dispatch import dispatch_inbound
@@ -51,6 +52,23 @@ SEA_ORAL = """海运
 运输分类：Port to Port
 """
 
+LAND_ORAL = """运输类型：国内运输整车
+线路类别：国内长途/零担
+运输分类：普件(东南亚、中港常用线路/常规车型)
+始发站：东莞
+目的地：上海
+货物品名：普货
+重量：1000
+体积：10
+"""
+
+# 测服陆运报价单抽出约 997 字、整理后只剩 48 字。
+_SEA_SHEET_RAW = (
+    "起运港：上海\n目的港：洛杉矶\n品名：普货\n整柜或拼柜：整柜\n"
+    "箱型箱量：1x40HQ\n贸易条款：FOB\n货物数量：20吨\n运输分类：Port to Port\n"
+) * 16
+_SEA_SHEET_TINY = "货物数量：100\n运输分类：其他\n报价日期：2026-09-23\n备注：报价单整理"
+
 
 def _extract(text: str):
     def _fn(_name: str, _raw: bytes):
@@ -83,6 +101,7 @@ class AttachmentInquiryTests(unittest.TestCase):
         reset_sea_text_flow_for_test()
         reset_land_text_flow_for_test()
         reset_air_text_flow_for_test()
+        reset_multi_text_flow_for_test()
         reset_attach_wave_book_for_test()
         reset_image_wave_book_for_test()
         self.store = MemoryMessageStore()
@@ -93,6 +112,69 @@ class AttachmentInquiryTests(unittest.TestCase):
     def _extras(self) -> list[dict]:
         return [i.payload for i in self.store._outbox.values()]
 
+    def test_tiny_organize_keeps_extracted_text(self) -> None:
+        """报价单被整理成一个字时，改用抽出的原文，否则多段开单看不到附件内容。"""
+        from agent.handlers.attachment_inquiry import prefer_extracted_oral
+
+        raw = "起运港\t纽约港\n目的港\t巴生西港\n品名\t机器设备\n" * 8
+        kept = "运输方式：海运\n起运港：纽约港\n目的港：巴生西港\n品名：机器设备\n箱型箱量：40HQ*1"
+        self.assertEqual(prefer_extracted_oral(raw, "。"), raw.strip())
+        self.assertEqual(prefer_extracted_oral(raw, kept), kept)
+        self.assertEqual(prefer_extracted_oral("", ""), "")
+        self.assertGreaterEqual(len(_SEA_SHEET_RAW), 200)
+        self.assertLess(len(_SEA_SHEET_TINY), 80)
+        self.assertEqual(prefer_extracted_oral(_SEA_SHEET_RAW, _SEA_SHEET_TINY), _SEA_SHEET_RAW.strip())
+
+    def test_leftover_multi_plus_shrunk_sheet_still_keeps_sea_ports(self) -> None:
+        """
+        旧多段补问草稿 + 千问把报价单压成几十个字时，
+        新附件必须当新询价，并把单元格原文里的起运港/目的港留下。
+        """
+        leftover = get_multi_text_flow()
+        leftover.on_text(
+            sender_id="s1",
+            text="段1：陆运\n段2：海运\n货物数量：100\n运输分类：其他",
+            reply=lambda t, extra=None: None,
+        )
+        self.assertEqual((leftover.session_of("s1").phase or ""), "clarify")
+
+        handle_attachment_inquiry(
+            InboundMessage(sender_id="s1", message_id="m-attach", content="附件询价"),
+            store=self.store,
+        )
+        self.assertIsNone(get_multi_text_flow().session_of("s1"))
+
+        def _extract_two(name: str, _raw: bytes):
+            if "陆运" in name:
+                return {"ok": True, "text": LAND_ORAL, "empty": False, "error": ""}
+            return {"ok": True, "text": _SEA_SHEET_RAW, "empty": False, "error": ""}
+
+        def _shrink_qianwen(**kwargs):
+            raw = str((kwargs.get("payload") or {}).get("raw_text") or "")
+            if "起运港" in raw:
+                return {"ok": True, "text": _SEA_SHEET_TINY, "empty": False, "error": ""}
+            return {"ok": True, "text": "运输类型：国内运输整车", "empty": False, "error": ""}
+
+        accept_h5_uploads(
+            sender_id="s1",
+            files=[("WO陆运报价单.xlsx", b"land"), ("手工报价海运.xlsx", b"sea")],
+            store=self.store,
+            extract_fn=_extract_two,
+            attach_text_fn=_shrink_qianwen,
+        )
+        bodies = self._bodies()
+        joined = "\n".join(bodies)
+        self.assertNotIn("起运港：(如：上海）", joined)
+        sess = get_multi_text_flow().session_of("s1")
+        self.assertIsNotNone(sess)
+        segs = (sess.quote or {}).get("multi_segments") or []
+        sea = next((row for row in segs if str(row.get("mode") or "") == "SEA"), None)
+        self.assertIsNotNone(sea)
+        sea_facts = dict((sea or {}).get("facts") or {})
+        self.assertEqual(str(sea_facts.get("起运港") or ""), "上海")
+        self.assertEqual(str(sea_facts.get("目的港") or ""), "洛杉矶")
+        self.assertEqual(str(sea_facts.get("品名") or ""), "普货")
+
     def test_intent_sends_upload_card_not_ticket(self) -> None:
         phase = handle_attachment_inquiry(
             InboundMessage(sender_id="s1", message_id="m1", content="帮我传个Excel询价，海运上海到纽约"),

inquiry-agent/tests/test_attachment_wave.py

@@ -68,6 +68,45 @@ class AttachWaveTests(unittest.TestCase):
         oral = self.book.compose_oral("s1")
         self.assertEqual(oral, "开口字\n海运 宁波到汉堡\n空运 上海到纽约\n只要40尺")
 
+    def test_empty_land_quote_sheet_still_makes_multi_with_sea(self) -> None:
+        """陆运报价单被整理成空时，仍要按文件名留陆运段，不能只剩海运。"""
+        from agent.policy.multi_segments import classify_multi_text
+
+        self.book.open_wait("s1", precede_text="", token="tok1", now=1.0)
+        self.book.add_files(
+            "s1",
+            ["WO202608240017_陆运报价单(1).xlsx", "手工报价海运.xlsx"],
+            now=2.0,
+        )
+        self.book.complete_material("s1", 0, text="")
+        self.book.complete_material(
+            "s1",
+            1,
+            text="运输方式：海运\n起运港：纽约港\n目的港：巴生西港",
+        )
+        oral = self.book.compose_oral("s1")
+        self.assertIn("陆运", oral)
+        self.assertIn("海运", oral)
+        self.assertIn("纽约港", oral)
+        self.assertEqual(classify_multi_text(oral), "multi")
+
+    def test_ltl_filename_counts_as_land_segment(self) -> None:
+        """零担询价表文件名没有「陆运」二字，也要和海运表切成两段。"""
+        from agent.policy.multi_segments import classify_multi_text
+
+        self.book.open_wait("s1", precede_text="", token="tok1", now=1.0)
+        self.book.add_files(
+            "s1",
+            ["东莞-上海  整车及零担 询价.xlsx", "手工海运报价单.xlsx"],
+            now=2.0,
+        )
+        self.book.complete_material("s1", 0, text="始发站：东莞\n目的站：上海")
+        self.book.complete_material("s1", 1, text="起运港：宁波港")
+        oral = self.book.compose_oral("s1")
+        self.assertIn("陆运", oral)
+        self.assertIn("海运", oral)
+        self.assertEqual(classify_multi_text(oral), "multi")
+
     def test_after_business_done_not_active(self) -> None:
         self.book.open_wait("s1", precede_text="", token="tok1", now=1.0)
         self.book.add_files("s1", ["a.xlsx"], now=2.0)

inquiry-agent/tests/test_extract_oral.py

@@ -327,5 +327,114 @@ class OralExtractTests(unittest.TestCase):
         self.assertNotIn("【询价卡片】", joined)
 
 
+class ImageOptionalClipTests(unittest.TestCase):
+    def test_numbered_extra_lines_do_not_enter_fields(self) -> None:
+        oral = """
+运输方式：空运
+目的地：印尼巴淡
+件数：1PLT
+重量：55.8KGS
+体积：1.23CBM
+包装类型：托盘
+报价日期：2026-09-23
+起运地：深圳坪山保税仓
+货好时间：今天货好
+4. 希望送到目的地工厂时间: 尽快
+5.
+提货地址：深圳市坪山区龙田街道大工业区绿荫北路7号601、
+602、603 (保税仓库)
+6. 送货地址: PT SAT NUSAPERSADA TBK----JL. PELITA VI NO.99
+"""
+        snap = extract_inquiry_snapshot(oral, allow_b=False)
+        facts = snap["facts"]
+        self.assertEqual(facts.get("货好时间"), "今天货好")
+        self.assertIn("坪山区", facts.get("提货地址") or "")
+        self.assertNotIn("希望送到", facts.get("货好时间") or "")
+        self.assertNotIn("希望送到", facts.get("提货地址") or "")
+        self.assertNotIn("送货地址", facts.get("提货地址") or "")
+        self.assertIn("PT SAT", facts.get("收货地址") or "")
+        from agent.policy import inquiry_copy as copy
+
+        card = copy.ask_clarify(
+            facts=facts,
+            missing_keys=["品名"],
+            transport_mode="空运",
+        )
+        self.assertNotIn("希望送到", card)
+        self.assertNotIn("4.", card)
+        self.assertIn("收货地址", card)
+
+
+# 测服手工海运报价单：表头和格子在同一行，还带 POL/POD/Terms。
+SEA_SHEET_ORAL = """
+起运港/POL（有指定起运港口的必填）        纽约港
+*品名/Commodity：        机器设备        目的港/POD（有指定目的港口的必填）        巴生西港
+*货物数量/Quantity (或提供装箱单/Packing List)        18
+整柜或拼柜：FCL
+箱型箱量：40HQ*1
+贸易条款/Terms：        CIF        要求运输时效/Requested Transit Time
+运输分类：其他
+报价日期：2026-09-07
+"""
+
+
+class SeaSheetHeaderClipTests(unittest.TestCase):
+    def test_quote_sheet_headers_do_not_stick_to_values(self) -> None:
+        """报价单表头不能进补问：起运港只要纽约港，贸易条款只要 CIF。"""
+        snap = extract_inquiry_snapshot(SEA_SHEET_ORAL, allow_b=False)
+        facts = snap["facts"]
+        self.assertEqual(facts.get("起运港"), "纽约港")
+        self.assertEqual(facts.get("目的港"), "巴生西港")
+        self.assertEqual(facts.get("品名"), "机器设备")
+        self.assertEqual(str(facts.get("货量") or facts.get("货物数量") or ""), "18")
+        self.assertEqual(facts.get("整柜或拼柜"), "FCL")
+        self.assertEqual(facts.get("箱型箱量"), "40HQ*1")
+        self.assertEqual(facts.get("贸易条款"), "CIF")
+        self.assertNotIn("必填", facts.get("起运港") or "")
+        self.assertNotIn("Terms", facts.get("贸易条款") or "")
+        self.assertNotIn("Commodity", facts.get("品名") or "")
+        self.assertNotIn("时效", facts.get("贸易条款") or "")
+
+        from agent.policy import inquiry_copy as copy
+
+        card = "\n".join(
+            copy.inquiry_field_lines(
+                work_order_no="",
+                facts=facts,
+                transport_mode="海运",
+            )
+        )
+        self.assertIn("纽约港", card)
+        self.assertNotIn("口的必填", card)
+        self.assertNotIn("Commodity", card)
+        self.assertNotIn("Requested Transit", card)
+
+    def test_next_column_header_is_not_the_value(self) -> None:
+        """
+        手工海运单同一行是「字段、格子、下一列表头」。
+        目的港只要巴生西港，不能收成费用汇总/POD Local Charges。
+        """
+        oral = """
+起运港/POL（有指定起运港口的必填）        纽约港        目的港/POD（有指定目的港口的必填）        巴生西港        费用汇总/POD Local Charges
+*品名/Commodity：        机器设备        *货物数量/Quantity (或提供装箱单/Packing List)        18        提货地址/Place of Receipt
+整柜或拼柜：FCL        箱型箱量：40HQ*1        送货地址/Place of Delivery
+贸易条款/Terms：        CIF        要求货物抵达时间/Requested Arrival Time
+运输分类：其他
+报价日期：2026-09-07
+"""
+        facts = extract_inquiry_snapshot(oral, allow_b=False)["facts"]
+        self.assertEqual(facts.get("起运港"), "纽约港")
+        self.assertEqual(facts.get("目的港"), "巴生西港")
+        self.assertEqual(facts.get("品名"), "机器设备")
+        self.assertEqual(str(facts.get("货量") or facts.get("货物数量") or ""), "18")
+        self.assertEqual(facts.get("整柜或拼柜"), "FCL")
+        self.assertEqual(facts.get("箱型箱量"), "40HQ*1")
+        self.assertEqual(facts.get("贸易条款"), "CIF")
+        self.assertNotIn("费用汇总", facts.get("目的港") or "")
+        self.assertNotIn("提货地址", facts.get("货量") or facts.get("货物数量") or "")
+        self.assertNotIn("送货地址", facts.get("整柜或拼柜") or "")
+        self.assertNotIn("抵达", facts.get("箱型箱量") or "")
+
+
 if __name__ == "__main__":
     unittest.main()

inquiry-agent/tests/test_land_options.py

@@ -169,6 +169,16 @@ class LandOptionsTest(unittest.TestCase):
         self.assertFalse(sample_pair_allowed("国内运输拼车", "东南亚"))
         self.assertFalse(sample_pair_allowed("中港整车", "东南亚"))
 
+    def test_scrub_template_ports(self):
+        from agent.schema.land_options import scrub_land_placeholder_ports
+
+        out = scrub_land_placeholder_ports(
+            {"起运港": "DOM", "目的港": "FTL", "品名": "衣服", "运输类型": "国内运输整车"}
+        )
+        self.assertNotIn("起运港", out)
+        self.assertNotIn("目的港", out)
+        self.assertEqual(out.get("品名"), "衣服")
+
     def test_confirm_intent(self):
         self.assertTrue(looks_like_confirm("确认"))
         self.assertTrue(looks_like_confirm("可以查了"))

inquiry-agent/tests/test_multi_group.py

@@ -0,0 +1,420 @@
+"""多段群内：按岗位归段报价，全部出单才整票成交跟进。"""
+
+from __future__ import annotations
+
+import json
+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.ledger.memory_ledger import MemoryLedger
+from agent.policy.multi_group_ops import deal_text, handle_multi_inbound
+from agent.policy.multi_segments import Segment
+
+
+class _Client:
+    def __init__(self) -> None:
+        self.texts: list[str] = []
+        self.cards: list[dict] = []
+        self.files: list[str] = []
+
+    def send_group(self, chat_id: str, content: str, mention_userids: list[str] | None = None) -> dict:
+        _ = chat_id, mention_userids
+        self.texts.append(content)
+        return {"ok": True}
+
+    def send_group_card(self, chat_id: str, template_card: dict) -> dict:
+        _ = chat_id
+        self.cards.append(template_card)
+        return {"ok": True}
+
+    def send_group_file(self, *, chat_id: str, filename: str, filepath: str = "", file_b64: str = "") -> dict:
+        _ = chat_id, filepath, file_b64
+        self.files.append(filename)
+        return {"ok": True}
+
+
+class _Flow:
+    def __init__(self, ledger: MemoryLedger, client: _Client) -> None:
+        self.ledger = ledger
+        self._ledger = ledger
+        self._group_client = client
+
+    def group_client(self) -> _Client:
+        return self._group_client
+
+
+def _ticket(ledger: MemoryLedger) -> object:
+    created = ledger.create_ticket(
+        sender_id="sales",
+        business_line="MULTI",
+        facts={
+            "segments_json": json.dumps(
+                [
+                    {"index": 1, "mode": "SEA", "facts": {"起运港": "广州港", "目的港": "巴生港"}},
+                    {"index": 2, "mode": "LAND", "facts": {"线路类别": "国内长途/零担"}},
+                ],
+                ensure_ascii=False,
+            )
+        },
+    )
+    ledger.set_sea_staff(
+        [{"wecomId": "sea1", "name": "海产", "roleCode": "sea", "status": "active", "routes": "海运·全部线路"}]
+    )
+    ledger.set_land_staff(
+        [{"wecomId": "land1", "name": "陆产", "roleCode": "land", "status": "active", "routes": "国内长途/零担"}]
+    )
+    ledger.upsert_staff({"wecomId": "sales", "name": "销售甲", "roleCode": "sales", "status": "active"})
+    no = created["work_order_no"]
+    ledger.bind_collab_group(
+        work_order_no=no,
+        chat_id="room-1",
+        member_ids=["sales", "sea1", "land1"],
+        product_names=["海产", "陆产"],
+        product_ids=["sea1", "land1"],
+    )
+    return ledger.get_ticket(work_order_no=no)
+
+
+class QuoteTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.ledger = MemoryLedger()
+        self.client = _Client()
+        self.flow = _Flow(self.ledger, self.client)
+        self.ticket = _ticket(self.ledger)
+        self.sheets: list[str] = []
+
+    def sheet(self, **kwargs: object) -> dict:
+        self.sheets.append(str(kwargs.get("business_line") or ""))
+        return {"ok": True}
+
+    def test_one_segment_quote_does_not_open_deal(self) -> None:
+        phase = handle_multi_inbound(
+            flow=self.flow,
+            ticket=self.ticket,
+            chat_id="room-1",
+            sender_id="sea1",
+            text="海运费 100CNY",
+            sheet_fn=self.sheet,
+        )
+        self.assertEqual(phase, "segment_quoted")
+        self.assertEqual(self.sheets, ["SEA"])
+        self.assertFalse(self.client.cards)
+        self.assertTrue(any("其他段报完" in text for text in self.client.texts))
+
+    def test_both_quotes_one_deal_without_sum(self) -> None:
+        handle_multi_inbound(
+            flow=self.flow,
+            ticket=self.ledger.get_ticket(work_order_no=self.ticket.work_order_no),
+            chat_id="room-1",
+            sender_id="sea1",
+            text="海运费 100CNY",
+            sheet_fn=self.sheet,
+        )
+        phase = handle_multi_inbound(
+            flow=self.flow,
+            ticket=self.ledger.get_ticket(work_order_no=self.ticket.work_order_no),
+            chat_id="room-1",
+            sender_id="land1",
+            text="陆运费 80CNY",
+            sheet_fn=self.sheet,
+        )
+        self.assertEqual(phase, "multi_deal_sent")
+        self.assertEqual(self.sheets, ["SEA", "LAND"])
+        body = self.client.texts[-1]
+        self.assertIn("段1：海运", body)
+        self.assertIn("段2：陆运", body)
+        self.assertNotIn("180", body)
+        rows = self.client.cards[-1]["horizontal_content_list"]
+        self.assertEqual([row["keyname"] for row in rows], ["段1海运", "段2陆运"])
+        joined = " ".join(row["value"] for row in rows)
+        self.assertNotIn("180", joined)
+
+    def test_land_person_cannot_fill_sea_segment(self) -> None:
+        phase = handle_multi_inbound(
+            flow=self.flow,
+            ticket=self.ticket,
+            chat_id="room-1",
+            sender_id="land1",
+            text="海运费 100CNY",
+            sheet_fn=self.sheet,
+        )
+        self.assertEqual(phase, "segment_quoted")
+        self.assertEqual(self.sheets, ["LAND"])
+
+    def test_requote_refreshes_deal(self) -> None:
+        handle_multi_inbound(
+            flow=self.flow,
+            ticket=self.ticket,
+            chat_id="room-1",
+            sender_id="sea1",
+            text="海运费 100CNY",
+            sheet_fn=self.sheet,
+        )
+        handle_multi_inbound(
+            flow=self.flow,
+            ticket=self.ledger.get_ticket(work_order_no=self.ticket.work_order_no),
+            chat_id="room-1",
+            sender_id="land1",
+            text="陆运费 80CNY",
+            sheet_fn=self.sheet,
+        )
+        phase = handle_multi_inbound(
+            flow=self.flow,
+            ticket=self.ledger.get_ticket(work_order_no=self.ticket.work_order_no),
+            chat_id="room-1",
+            sender_id="sea1",
+            text="海运费 120CNY",
+            sheet_fn=self.sheet,
+        )
+        self.assertEqual(phase, "multi_deal_sent")
+        self.assertIn("120", self.client.texts[-1])
+        self.assertNotIn("100", deal_text(
+            work_order_no=self.ticket.work_order_no,
+            segments=[Segment(index=1, mode="SEA"), Segment(index=2, mode="LAND")],
+            quotes={"SEA": {"total": "120", "currency": "CNY"}, "LAND": {"total": "80", "currency": "CNY"}},
+        ))
+
+    def test_work_order_in_group_sends_brief_and_templates(self) -> None:
+        no = self.ticket.work_order_no
+        phase = handle_multi_inbound(
+            flow=self.flow,
+            ticket=self.ticket,
+            chat_id="room-9",
+            sender_id="sales",
+            text=no,
+            sheet_fn=self.sheet,
+        )
+        self.assertEqual(phase, "multi_activated")
+        self.assertEqual(self.sheets, ["SEA", "LAND"])
+        brief = self.client.texts[-1]
+        self.assertIn("段1：海运", brief)
+        self.assertIn("段2：陆运", brief)
+        self.assertNotIn("TMS报价", brief)
+        self.assertNotIn("其他段报完", brief)
+        self.assertEqual(self.ledger.get_ticket(work_order_no=no).collab_chat_id, "room-9")
+
+    def test_at_bot_work_order_activates_not_quote(self) -> None:
+        self.ledger.upsert_staff(
+            {"wecomId": "air1", "name": "航线甲", "roleCode": "air", "status": "active"}
+        )
+        created = self.ledger.create_ticket(
+            sender_id="sales",
+            business_line="MULTI",
+            facts={
+                "segments_json": json.dumps(
+                    [
+                        {
+                            "index": 1,
+                            "mode": "LAND",
+                            "facts": {
+                                "运输类型": "国内运输拼车",
+                                "起运港": "广州",
+                                "目的港": "深圳",
+                                "品名": "LMM振动测试定位治具",
+                                "体积": "100CBM",
+                                "毛重": "1.5kg",
+                            },
+                        },
+                        {
+                            "index": 2,
+                            "mode": "AIR",
+                            "facts": {
+                                "目的港": "CRK",
+                                "品名": "LMM振动测试定位治具",
+                                "件数": "1件",
+                                "毛重": "1.5",
+                                "体积": "100CBM",
+                                "包装方式": "托盘",
+                                "起运港": "ZUH",
+                                "贸易条款": "DAP",
+                            },
+                        },
+                    ],
+                    ensure_ascii=False,
+                )
+            },
+        )
+        no = created["work_order_no"]
+        ticket = self.ledger.get_ticket(work_order_no=no)
+        phase = handle_multi_inbound(
+            flow=self.flow,
+            ticket=ticket,
+            chat_id="air-room",
+            sender_id="sales",
+            text=f"@询价小助手 {no}",
+            sheet_fn=self.sheet,
+        )
+        self.assertEqual(phase, "multi_activated")
+        self.assertEqual(self.sheets, ["LAND", "AIR"])
+        brief = self.client.texts[0]
+        self.assertIn("@销售甲", brief)
+        self.assertIn("@陆产", brief)
+        self.assertIn("@航线甲", brief)
+        self.assertIn("段1：陆运", brief)
+        self.assertIn("运输类型（必填）：国内运输拼车", brief)
+        self.assertIn("始发站（必填）：广州", brief)
+        self.assertIn("段2：空运", brief)
+        self.assertIn("目的地（必填）：CRK", brief)
+        self.assertIn("贸易条款（非必填）：DAP", brief)
+        self.assertNotIn("其他段报完", "\n".join(self.client.texts))
+        self.assertNotIn("询价字段：", brief)
+
+    def test_http_snapshot_activation_mentions_real_sales_not_job_word(self) -> None:
+        """主账快照没带回销售企微号时，群里仍要 @ 具体人，不能写岗位词「销售」。"""
+        self.ledger.upsert_staff(
+            {"wecomId": "WuJiLin", "name": "吴吉林", "roleCode": "sales", "status": "active"}
+        )
+        self.ledger.upsert_staff(
+            {"wecomId": "air1", "name": "航线甲", "roleCode": "air", "status": "active"}
+        )
+        created = self.ledger.create_ticket(
+            sender_id="WuJiLin",
+            business_line="MULTI",
+            facts={
+                "segments_json": json.dumps(
+                    [
+                        {
+                            "index": 1,
+                            "mode": "LAND",
+                            "facts": {
+                                "运输类型": "国内运输拼车",
+                                "起运港": "广州",
+                                "目的港": "深圳",
+                                "品名": "治具",
+                            },
+                        },
+                        {"index": 2, "mode": "AIR", "facts": {"起运港": "ZUH", "目的港": "CRK", "品名": "治具"}},
+                    ],
+                    ensure_ascii=False,
+                )
+            },
+        )
+        no = created["work_order_no"]
+        real_get = self.ledger.get_ticket
+
+        def http_snap(*, work_order_no: str):
+            ticket = real_get(work_order_no=work_order_no)
+            if ticket is None:
+                return None
+            return {
+                "ok": True,
+                "found": True,
+                "work_order_no": ticket.work_order_no,
+                "business_line": "MULTI",
+                "facts": dict(ticket.facts),
+                "sales_name": "销售",
+                "salesName": "销售",
+                "collab_chat_id": ticket.collab_chat_id,
+            }
+
+        self.ledger.get_ticket = http_snap  # type: ignore[method-assign]
+        phase = handle_multi_inbound(
+            flow=self.flow,
+            ticket=http_snap(work_order_no=no),
+            chat_id="live-room",
+            sender_id="WuJiLin",
+            text=f"@询价小助手 {no}",
+            sheet_fn=self.sheet,
+        )
+        self.assertEqual(phase, "multi_activated")
+        brief = self.client.texts[0]
+        self.assertIn("@吴吉林", brief)
+        self.assertNotRegex(brief, r"@销售(?:@|\n|$)")
+
+    def test_land_sheet_matches_when_route_missing_and_no_default(self) -> None:
+        """多段陆运常缺线路类别；后台陆运模板只按线路关键词时，仍要发出陆运表。"""
+        self.ledger.set_quote_templates(
+            [
+                {
+                    "templateId": "air-default",
+                    "name": "空运默认",
+                    "bizType": "空运",
+                    "keywords": [],
+                    "isDefault": True,
+                },
+                {
+                    "templateId": "land-lcl",
+                    "name": "陆运国内运输拼车报价单",
+                    "bizType": "陆运",
+                    "keywords": ["国内拼车", "国内零担"],
+                    "isDefault": False,
+                },
+            ]
+        )
+        created = self.ledger.create_ticket(
+            sender_id="sales",
+            business_line="MULTI",
+            facts={
+                "segments_json": json.dumps(
+                    [
+                        {
+                            "index": 1,
+                            "mode": "LAND",
+                            "facts": {
+                                "运输类型": "国内运输拼车",
+                                "起运港": "广州",
+                                "目的港": "深圳",
+                                "品名": "治具",
+                                "体积": "100CBM",
+                                "毛重": "1.5kg",
+                            },
+                        },
+                        {
+                            "index": 2,
+                            "mode": "AIR",
+                            "facts": {"起运港": "ZUH", "目的港": "CRK", "品名": "治具"},
+                        },
+                    ],
+                    ensure_ascii=False,
+                )
+            },
+        )
+        no = created["work_order_no"]
+        ticket = self.ledger.get_ticket(work_order_no=no)
+        phase = handle_multi_inbound(
+            flow=self.flow,
+            ticket=ticket,
+            chat_id="sheet-room",
+            sender_id="sales",
+            text=f"@询价小助手 {no}",
+        )
+        self.assertEqual(phase, "multi_activated")
+        self.assertEqual(len(self.client.files), 2)
+        self.assertTrue(self.ledger.last_render.get("ok"))
+
+    def test_lock_on_air_segment_does_not_wait_for_other_quotes(self) -> None:
+        created = self.ledger.create_ticket(
+            sender_id="sales",
+            business_line="MULTI",
+            facts={
+                "segments_json": json.dumps(
+                    [
+                        {"index": 1, "mode": "SEA", "facts": {"起运港": "广州港"}},
+                        {"index": 2, "mode": "AIR", "facts": {"目的港": "CRK"}},
+                    ],
+                    ensure_ascii=False,
+                )
+            },
+        )
+        self.ledger.upsert_staff({"wecomId": "air1", "name": "航线甲", "roleCode": "air", "status": "active"})
+        no = created["work_order_no"]
+        ticket = self.ledger.get_ticket(work_order_no=no)
+        phase = handle_multi_inbound(
+            flow=self.flow,
+            ticket=ticket,
+            chat_id="air-room",
+            sender_id="air1",
+            text=f"{no}锁舱",
+            sheet_fn=self.sheet,
+        )
+        self.assertEqual(phase, "lock_ok")
+        self.assertFalse(any("其他段报完" in text for text in self.client.texts))
+
+
+if __name__ == "__main__":
+    unittest.main()

inquiry-agent/tests/test_multi_segments.py

@@ -0,0 +1,818 @@
+"""多段联运：分段、补问、核对建单、按线路拉进同一个群。"""
+
+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.ledger.memory_ledger import MemoryLedger
+from agent.policy.multi_segments import (
+    Segment,
+    classify_multi_text,
+    group_brief,
+    merge_supplement,
+    refresh_segments,
+    split_segments,
+)
+from agent.policy.multi_text_flow import MultiTextInquiryFlow
+
+CLARIFY_REPLY = """段1：国内运输拼车
+段2：运输方式：空运
+起运地：ZUH
+目的地：CRK
+品名：LMM振动测试定位治具
+件数：1件
+重量：1.5
+体积：100CBM
+报价日期：2026-09-25
+贸易条款：DAP
+托盘
+"""
+
+SEA_AIR = """段1：海运
+起运港：广州港
+目的港：巴生港
+品名：机器设备
+货物数量：15件
+整柜或拼柜：FCL
+箱型箱量：40HQ*1
+贸易条款：DOOR-DOOR
+运输分类：Port to Port
+报价日期：2026-09-23
+
+段2：空运
+目的地：CRK
+品名：LMM振动测试定位治具
+件数：1件
+重量：1.5
+体积：100CBM
+报价日期：2026-09-25
+起运地：ZUH
+贸易条款：DAP
+"""
+
+SEA_AIR_READY = SEA_AIR + "\n包装类型：纸箱\n"
+
+
+class SplitTests(unittest.TestCase):
+    def test_labeled_segments_keep_order_and_fields(self) -> None:
+        outcome = split_segments(SEA_AIR)
+        self.assertTrue(outcome.ok)
+        self.assertEqual([seg.mode for seg in outcome.segments], ["SEA", "AIR"])
+        self.assertEqual(outcome.segments[0].facts.get("起运港"), "广州港")
+        self.assertEqual(outcome.segments[1].facts.get("目的地"), "CRK")
+
+    def test_same_mode_twice_is_rejected(self) -> None:
+        self.assertEqual(classify_multi_text("段1：海运\n起运港：A\n段2：海运\n目的港：B"), "same_mode")
+
+    def test_explicit_markers_ignore_mode_words_in_body(self) -> None:
+        """文件名先标了段，表里再写「运输方式：海运」不能当成又一段海运。"""
+        oral = (
+            "段1：陆运\n"
+            "运输类型：国内运输拼车\n"
+            "段2：海运\n"
+            "运输方式：海运\n"
+            "起运港：纽约港\n"
+            "目的港：巴生西港"
+        )
+        self.assertEqual(classify_multi_text(oral), "multi")
+        outcome = split_segments(oral)
+        self.assertEqual([seg.mode for seg in outcome.segments], ["LAND", "SEA"])
+        self.assertEqual(outcome.segments[1].facts.get("起运港"), "纽约港")
+
+    def test_sea_sheet_headers_do_not_stick_to_ports(self) -> None:
+        """海运报价单表头不能写进补问，只要格子里的纽约港/CIF。"""
+        oral = (
+            "段1：陆运\n运输类型：国内运输整车\n"
+            "段2：海运\n"
+            "起运港/POL（有指定起运港口的必填）        纽约港\n"
+            "*品名/Commodity：        机器设备        "
+            "目的港/POD（有指定目的港口的必填）        巴生西港\n"
+            "*货物数量/Quantity (或提供装箱单/Packing List)        18\n"
+            "整柜或拼柜：FCL\n箱型箱量：40HQ*1\n"
+            "贸易条款/Terms：        CIF        要求运输时效/Requested Transit Time\n"
+            "运输分类：其他\n"
+        )
+        outcome = split_segments(oral)
+        sea = outcome.segments[1]
+        self.assertEqual(sea.facts.get("起运港"), "纽约港")
+        self.assertEqual(sea.facts.get("目的港"), "巴生西港")
+        self.assertEqual(sea.facts.get("品名"), "机器设备")
+        self.assertEqual(sea.facts.get("贸易条款"), "CIF")
+        _, _, gaps = refresh_segments(outcome.segments)
+        self.assertNotIn("口的必填", gaps)
+        self.assertNotIn("Commodity", gaps)
+
+    def test_inline_two_modes(self) -> None:
+        outcome = split_segments("陆运+海运")
+        self.assertTrue(outcome.ok)
+        self.assertEqual([seg.mode for seg in outcome.segments], ["LAND", "SEA"])
+
+    def test_group_brief_has_no_blank_under_mentions(self) -> None:
+        """拉群摘要 @人后直接接段1，中间不要空行。"""
+        segs = split_segments(SEA_AIR).segments
+        text = group_brief(
+            work_order_no="WO1",
+            segments=segs,
+            sales_name="吴季林",
+            mention_names=["林绵助"],
+        )
+        self.assertIn("@吴季林@林绵助\n段1：", text)
+        self.assertNotIn("@林绵助\n\n段1", text)
+
+    def test_missing_fields_use_single_chat_examples(self) -> None:
+        outcome = split_segments("段1：海运\n起运港：广州港\n段2：空运\n目的地：CRK")
+        ready, _, text = refresh_segments(outcome.segments)
+        self.assertFalse(ready)
+        self.assertIn("请确认并补充以下信息：", text)
+        self.assertIn("待补充：", text)
+        self.assertIn("品名：(如：茶叶）", text)
+        self.assertIn("货源地（非必填）：(如：东莞）", text)
+        self.assertIn("贸易条款（非必填）：(如：FOB）", text)
+
+    def test_port_to_port_fills_sea_class(self) -> None:
+        """海运只缺运输分类时，回复 Port to Port 要写进该段，和单聊一样。"""
+        sea = Segment(
+            index=2,
+            mode="SEA",
+            facts={
+                "起运港": "纽约港",
+                "目的港": "巴生西港",
+                "品名": "机器设备",
+                "货量": "20吨",
+                "整柜或拼柜": "FCL",
+                "箱型箱量": "40HQ*1",
+                "贸易条款": "CIF",
+            },
+        )
+        sea_ok, _, _ = refresh_segments([sea])
+        self.assertFalse(sea_ok)
+        merged = merge_supplement([sea], "Port to Port")
+        self.assertEqual(merged[0].facts.get("运输分类"), "Port to Port")
+
+    def test_segment_marked_supplement_keeps_old_facts(self) -> None:
+        """补问里再写段1/段2是补字段，不能整单重开把已填港口冲掉。"""
+        draft = [
+            Segment(
+                index=1,
+                mode="LAND",
+                facts={"运输类型": "国内运输整车", "品名": "衣服", "车型/数量": "13米×3"},
+            ),
+            Segment(
+                index=2,
+                mode="SEA",
+                facts={
+                    "起运港": "纽约港",
+                    "目的港": "巴生西港",
+                    "品名": "机器设备",
+                    "整柜或拼柜": "FCL",
+                    "箱型箱量": "40HQ*1",
+                    "贸易条款": "CIF",
+                },
+            ),
+        ]
+        text = (
+            "段1：陆运：\n国内长途/零担、普件(东南亚、中港常用线路/常规车型)\n\n"
+            "段2：海运\n货物数量：100\n运输分类（必填）：其他。"
+        )
+        merged = merge_supplement(draft, text)
+        land = next(seg for seg in merged if seg.mode == "LAND")
+        sea = next(seg for seg in merged if seg.mode == "SEA")
+        self.assertEqual(land.facts.get("运输类型"), "国内运输整车")
+        self.assertEqual(land.facts.get("品名"), "衣服")
+        self.assertEqual(land.facts.get("线路类别"), "国内长途/零担")
+        self.assertIn("普件", str(land.facts.get("运输分类") or ""))
+        self.assertEqual(sea.facts.get("起运港"), "纽约港")
+        self.assertEqual(sea.facts.get("目的港"), "巴生西港")
+        self.assertEqual(sea.facts.get("品名"), "机器设备")
+        self.assertEqual(sea.facts.get("货量") or sea.facts.get("货物数量"), "100")
+        self.assertEqual(sea.facts.get("运输分类"), "其他")
+
+    def test_incomplete_land_does_not_confirm_when_sea_ready(self) -> None:
+        """海运齐了、陆运还缺线路/分类、站点是模板码时，不能出确定创建工单。"""
+        segs = [
+            Segment(
+                index=1,
+                mode="LAND",
+                facts={
+                    "运输类型": "国内运输整车",
+                    "起运港": "DOM",
+                    "目的港": "FTL",
+                    "品名": "衣服；车型/数量：13米×3；体积：100CBM；重量：10KG",
+                    "车型/数量": "13米×3",
+                    "报价日期": "2026-08-24",
+                },
+            ),
+            Segment(
+                index=2,
+                mode="SEA",
+                facts={
+                    "起运港": "纽约港",
+                    "目的港": "巴生西港",
+                    "品名": "机器设备",
+                    "货量": "100",
+                    "整柜或拼柜": "FCL",
+                    "箱型箱量": "40HQ*1",
+                    "贸易条款": "CIF",
+                    "运输分类": "其他",
+                    "报价日期": "2026-09-23",
+                },
+            ),
+        ]
+        ready, updated, text = refresh_segments(segs)
+        self.assertFalse(ready)
+        self.assertIn("线路类别（必填）：可选择回复", text)
+        self.assertIn("国内长途/零担", text)
+        self.assertIn("国内城配/拖车/打包", text)
+        self.assertIn("运输分类（必填）：可选择回复", text)
+        self.assertIn("1.普件", text)
+        self.assertNotIn("始发站", text)
+        self.assertNotIn("请回复「确定」创建工单", text)
+        filled = merge_supplement(
+            updated, "线路类别：国内长途/零担\n运输分类：急件"
+        )
+        ready2, _, cargo = refresh_segments(filled)
+        self.assertFalse(ready2)
+        self.assertIn("始发站", cargo)
+        self.assertIn("目的地", cargo)
+        self.assertNotIn("可选择回复", cargo)
+        self.assertNotIn("请回复「确定」创建工单", cargo)
+
+    def test_clarify_reply_fills_land_type_and_air_pack(self) -> None:
+        draft = split_segments("段1：陆运\n段2：空运\n报价日期：2026-09-23")
+        self.assertTrue(draft.ok)
+        merged = merge_supplement(draft.segments, CLARIFY_REPLY)
+        by_mode = {seg.mode: seg for seg in merged}
+        self.assertEqual(by_mode["LAND"].facts.get("运输类型"), "国内运输拼车")
+        self.assertEqual(by_mode["AIR"].facts.get("目的港") or by_mode["AIR"].facts.get("目的地"), "CRK")
+        self.assertEqual(by_mode["AIR"].facts.get("包装方式") or by_mode["AIR"].facts.get("包装类型"), "托盘")
+
+    def test_oral_land_line_fills_only_lacking_segment(self) -> None:
+        draft = split_segments(
+            "段1：陆运\n运输类型：国内运输拼车\n段2：空运\n目的地：CRK\n"
+            "品名：治具\n件数：1件\n重量：1.5\n体积：100CBM\n包装类型：托盘"
+        )
+        oral = "广州-深圳，LMM振动测试定位治具，100CBM，1.5"
+        extra = {
+            "起运港": "广州",
+            "目的港": "深圳",
+            "品名": "LMM振动测试定位治具",
+            "体积": "100CBM",
+            "毛重": "1.5",
+        }
+        merged = merge_supplement(draft.segments, oral, extra_facts=extra)
+        land = next(seg for seg in merged if seg.mode == "LAND")
+        air = next(seg for seg in merged if seg.mode == "AIR")
+        self.assertEqual(land.facts.get("起运港"), "广州")
+        self.assertEqual(land.facts.get("目的港"), "深圳")
+        self.assertEqual(land.facts.get("品名"), "LMM振动测试定位治具")
+        self.assertIn("100", str(land.facts.get("体积") or ""))
+        self.assertEqual(air.facts.get("品名"), "治具")
+
+    def test_confirm_sea_fix_updates_ready_sea_segment(self) -> None:
+        """核对后只改正海运格子，要写回海运段，不能当重开。"""
+        segs = split_segments(SEA_LAND_READY).segments
+        merged = merge_supplement(
+            segs,
+            "目的港：巴生西港\n货物数量：18\n整柜或拼柜：FCL\n箱型箱量：40HQ*1",
+        )
+        sea = next(seg for seg in merged if seg.mode == "SEA")
+        land = next(seg for seg in merged if seg.mode == "LAND")
+        self.assertEqual(sea.facts.get("目的港"), "巴生西港")
+        self.assertEqual(str(sea.facts.get("货量") or sea.facts.get("货物数量") or ""), "18")
+        self.assertEqual(sea.facts.get("箱型箱量"), "40HQ*1")
+        self.assertEqual(land.facts.get("品名") or land.facts.get("货物品名"), "零件")
+
+
+class FlowTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.ledger = MemoryLedger()
+        self.replies: list[str] = []
+        self.flow = MultiTextInquiryFlow(self.ledger, group_client=_Group())
+
+    def reply(self, text: str, extra: dict | None = None) -> str:
+        _ = extra
+        self.replies.append(text)
+        return "ok"
+
+    def last_prompt(self) -> str:
+        """样例图兜底会后发，取带段号或可选择回复的那条补问。"""
+        for text in reversed(self.replies):
+            if "段1" in text or "可选择回复" in text or "请回复「确定」" in text or "待补充" in text:
+                return text
+        return self.replies[-1] if self.replies else ""
+
+    def test_missing_air_pack_is_asked_once(self) -> None:
+        phase = self.flow.on_text(sender_id="sales", text=SEA_AIR, reply=self.reply)
+        self.assertEqual(phase, "clarify")
+        card = self.replies[-1]
+        self.assertIn("请确认并补充以下信息：", card)
+        self.assertIn("段2：空运", card)
+        self.assertIn("待补充：", card)
+        self.assertIn("包装类型：(如：", card)
+        self.assertNotIn("TMS", card)
+
+    def test_clarify_reply_gets_a_followup(self) -> None:
+        phase = self.flow.on_text(sender_id="sales", text="段1：陆运\n段2：空运", reply=self.reply)
+        self.assertEqual(phase, "clarify")
+        phase = self.flow.on_text(sender_id="sales", text=CLARIFY_REPLY, reply=self.reply)
+        self.assertIn(phase, {"clarify", "wait_confirm"})
+        self.assertTrue(self.replies[-1].strip())
+        self.assertIn("请确认", self.last_prompt())
+
+    def test_open_extracts_unlabeled_attachment_oral(self) -> None:
+        """附件口播常没有「起运港：」，开多段时要走和单聊一样的抽取，不能整段待补充。"""
+        import json
+        from types import SimpleNamespace
+
+        def chat_fn(**kwargs):
+            msgs = kwargs.get("messages") or []
+            blob = str((msgs[-1] or {}).get("content") if msgs else "")
+            if "东莞" in blob:
+                payload = {
+                    "transport_mode": "LAND",
+                    "land_subtype": "国内运输拼车",
+                    "fields": {"起运港": "东莞", "目的港": "上海", "品名": "茶叶"},
+                }
+            else:
+                payload = {
+                    "transport_mode": "SEA",
+                    "land_subtype": "",
+                    "fields": {
+                        "起运港": "纽约港",
+                        "目的港": "巴生西港",
+                        "品名": "机器设备",
+                        "整柜或拼柜": "FCL",
+                        "箱型箱量": "40HQ*1",
+                    },
+                }
+            return SimpleNamespace(
+                ok=True,
+                content="",
+                error="",
+                tool_calls=[
+                    {
+                        "function": {
+                            "name": "extract_inquiry_fields",
+                            "arguments": json.dumps(payload, ensure_ascii=False),
+                        }
+                    }
+                ],
+            )
+
+        oral = (
+            "段1：陆运\n东莞到上海 茶叶 2吨\n"
+            "段2：海运\n纽约到巴生 机器设备 40HQ"
+        )
+        phase = self.flow.on_text(
+            sender_id="sales",
+            text=oral,
+            reply=self.reply,
+            chat_fn=chat_fn,
+        )
+        card = self.last_prompt()
+        self.assertEqual(phase, "clarify")
+        self.assertIn("线路类别（必填）：可选择回复", card)
+        self.assertIn("纽约港", card)
+        self.assertIn("巴生西港", card)
+        self.assertIn("机器设备", card)
+        self.assertNotIn("东莞", card)
+        self.assertNotIn("起运港：(如：上海）", card)
+
+    def test_oral_land_supplement_is_recognized(self) -> None:
+        start = (
+            "段1：陆运\n运输类型：国内运输拼车\n"
+            "段2：空运\n目的地：CRK\n品名：治具\n件数：1件\n重量：1.5\n"
+            "体积：100CBM\n包装类型：托盘\n报价日期：2026-09-23"
+        )
+        phase = self.flow.on_text(sender_id="sales", text=start, reply=self.reply)
+        self.assertEqual(phase, "clarify")
+        first = self.last_prompt()
+        self.assertIn("线路类别（必填）：可选择回复", first)
+        self.assertIn("运输分类（必填）：可选择回复", first)
+        self.assertNotIn("始发站", first)
+        phase = self.flow.on_text(
+            sender_id="sales",
+            text="广州-深圳，LMM振动测试定位治具，100CBM，1.5",
+            reply=self.reply,
+            injected_facts={
+                "起运港": "广州",
+                "目的港": "深圳",
+                "品名": "LMM振动测试定位治具",
+                "体积": "100CBM",
+                "毛重": "1.5",
+            },
+            injected_mode="LAND",
+        )
+        card = self.last_prompt()
+        self.assertEqual(phase, "clarify")
+        self.assertIn("线路类别（必填）：可选择回复", card)
+        self.assertNotIn("始发站", card)
+        self.assertNotIn("请回复「确定」创建工单", card)
+
+    def test_ready_confirm_has_no_collab_and_no_tms(self) -> None:
+        phase = self.flow.on_text(sender_id="sales", text=SEA_AIR_READY, reply=self.reply)
+        self.assertEqual(phase, "wait_confirm")
+        card = self.replies[-1]
+        self.assertIn("段1：海运", card)
+        self.assertIn("段2：空运", card)
+        self.assertIn("请回复「确定」创建工单", card)
+        self.assertNotIn("协同补充", card)
+        self.assertNotIn("TMS", card)
+
+    def test_wait_confirm_field_fix_stays_on_draft(self) -> None:
+        """核对卡上改海运目的港/箱量，不能回「请把多段需求再发一次」。"""
+        phase = self.flow.on_text(sender_id="sales", text=SEA_LAND_READY, reply=self.reply)
+        self.assertEqual(phase, "wait_confirm")
+        phase = self.flow.on_text(
+            sender_id="sales",
+            text="目的港：巴生西港\n货物数量：18\n整柜或拼柜：FCL\n箱型箱量：40HQ*1",
+            reply=self.reply,
+        )
+        card = self.last_prompt()
+        self.assertEqual(phase, "wait_confirm")
+        self.assertNotIn("请把多段需求再发一次", card)
+        self.assertIn("巴生西港", card)
+        self.assertIn("18", card)
+        sea = next(seg for seg in self.flow._segments(self.flow.session_of("sales")) if seg.mode == "SEA")
+        self.assertEqual(sea.facts.get("目的港"), "巴生西港")
+        self.assertEqual(str(sea.facts.get("货量") or ""), "18")
+
+    def test_incomplete_land_stays_clarify_when_sea_ready(self) -> None:
+        """销售看到的那张卡：海运齐、陆运空着线路和模板码站点，必须继续补问。"""
+        text = """段1：陆运
+运输类型：国内运输整车
+始发站：DOM
+目的地：FTL
+货物品名：衣服
+车型/数量：13米×3
+报价日期：2026-08-24
+段2：海运
+起运港：纽约港
+目的港：巴生西港
+品名：机器设备
+货物数量：100
+整柜或拼柜：FCL
+箱型箱量：40HQ*1
+贸易条款：CIF
+运输分类：其他
+报价日期：2026-09-23
+"""
+        phase = self.flow.on_text(sender_id="sales", text=text, reply=self.reply)
+        self.assertEqual(phase, "clarify")
+        card = self.last_prompt()
+        self.assertIn("线路类别（必填）：可选择回复", card)
+        self.assertIn("运输分类（必填）：可选择回复", card)
+        self.assertNotIn("始发站", card)
+        self.assertNotIn("请回复「确定」创建工单", card)
+
+    def test_clarify_segment_reply_is_not_a_new_inquiry(self) -> None:
+        """补问阶段按段回复线路和货量，要写回原单，不能再要一遍起运港。"""
+        start = """段1：陆运
+运输类型：国内运输整车
+段2：海运
+起运港：纽约港
+目的港：巴生西港
+品名：机器设备
+整柜或拼柜：FCL
+箱型箱量：40HQ*1
+贸易条款：CIF
+"""
+        phase = self.flow.on_text(sender_id="sales", text=start, reply=self.reply)
+        self.assertEqual(phase, "clarify")
+        self.replies.clear()
+        phase = self.flow.on_text(
+            sender_id="sales",
+            text=(
+                "段1：陆运：\n国内长途/零担、普件(东南亚、中港常用线路/常规车型)\n\n"
+                "段2：海运\n货物数量：100\n运输分类（必填）：其他。"
+            ),
+            reply=self.reply,
+        )
+        self.assertEqual(phase, "clarify")
+        card = self.last_prompt()
+        self.assertNotIn("起运港：(如：上海）", card)
+        self.assertIn("始发站", card)
+        sess = self.flow.session_of("sales")
+        segs = self.flow._segments(sess)
+        sea = next(seg for seg in segs if seg.mode == "SEA")
+        land = next(seg for seg in segs if seg.mode == "LAND")
+        self.assertEqual(sea.facts.get("起运港"), "纽约港")
+        self.assertEqual(land.facts.get("线路类别"), "国内长途/零担")
+
+    def test_confirm_creates_one_ticket_and_air_hint(self) -> None:
+        self.flow.on_text(sender_id="sales", text=SEA_AIR_READY, reply=self.reply)
+        phase = self.flow.on_text(sender_id="sales", text="确定", reply=self.reply)
+        self.assertEqual(phase, "wait_manual_group")
+        hint = self.replies[-1]
+        self.assertIn("请手动拉相关人员和智能体到原有群", hint)
+        self.assertIn("WO", hint)
+        tickets = list(self.ledger._tickets.values())
+        self.assertEqual(len(tickets), 1)
+        self.assertEqual(tickets[0].business_line, "MULTI")
+
+    def test_new_multi_after_manual_group_starts_fresh(self) -> None:
+        """含空运工单已出号后，再发陆运+海运是新询价，不能回旧工单拉群提示。"""
+        self.flow.on_text(sender_id="sales", text=SEA_AIR_READY, reply=self.reply)
+        self.flow.on_text(sender_id="sales", text="确定", reply=self.reply)
+        old_wo = self.flow.session_of("sales").work_order_no
+        self.assertTrue(old_wo)
+        self.replies.clear()
+        phase = self.flow.on_text(
+            sender_id="sales",
+            text="段1：陆运\n运输类型：国内运输拼车\n段2：海运\n起运港：上海港\n目的港：汉堡",
+            reply=self.reply,
+        )
+        self.assertNotEqual(phase, "wait_manual_group")
+        self.assertNotIn(old_wo, self.last_prompt())
+        self.assertNotIn("请手动拉相关人员和智能体", self.last_prompt())
+        self.assertIn("陆运", self.last_prompt())
+        self.assertIn("海运", self.last_prompt())
+
+    def test_plain_text_after_manual_group_still_hints(self) -> None:
+        """不是新询价时，含空运工单仍提示去原有群发工单号。"""
+        self.flow.on_text(sender_id="sales", text=SEA_AIR_READY, reply=self.reply)
+        self.flow.on_text(sender_id="sales", text="确定", reply=self.reply)
+        old_wo = self.flow.session_of("sales").work_order_no
+        self.replies.clear()
+        phase = self.flow.on_text(sender_id="sales", text="嗯", reply=self.reply)
+        self.assertEqual(phase, "wait_manual_group")
+        self.assertIn("请手动拉相关人员和智能体到原有群", self.replies[-1])
+        self.assertIn(old_wo, self.replies[-1])
+
+    def test_same_mode_does_not_create(self) -> None:
+        phase = self.flow.on_text(
+            sender_id="sales",
+            text="段1：海运\n起运港：广州港\n段2：海运\n目的港：新加坡",
+            reply=self.reply,
+        )
+        self.assertEqual(phase, "reject_same_mode")
+        self.assertEqual(self.ledger._tickets, {})
+
+
+SEA_LAND_READY = """段1：海运
+起运港：广州港
+目的港：巴生港
+品名：机器
+货物数量：1
+整柜或拼柜：FCL
+箱型箱量：40HQ*1
+贸易条款：FOB
+运输分类：Port to Port
+报价日期：2026-09-23
+段2：陆运
+运输类型：国内运输拼车
+线路类别：国内长途/零担
+运输分类：急件
+始发站：上海
+目的地：苏州
+货物品名：零件
+重量：10
+体积：1
+车型/数量：4.2米/1
+报价日期：2026-09-23
+"""
+
+
+def _staff_ledger() -> MemoryLedger:
+    """海运+陆运各一个人，点拉群才能建得成。"""
+    ledger = MemoryLedger()
+    ledger.set_sea_staff(
+        [{"wecomId": "sea1", "name": "海产", "roleCode": "sea", "status": "active", "routes": "海运·全部线路"}]
+    )
+    ledger.set_land_staff(
+        [
+            {
+                "wecomId": "land1",
+                "name": "陆产",
+                "roleCode": "land",
+                "status": "active",
+                "routes": "国内长途/零担",
+            }
+        ]
+    )
+    return ledger
+
+
+class _Group:
+    def __init__(self) -> None:
+        self.created: list[tuple[str, list[str]]] = []
+        self.messages: list[str] = []
+
+    def create_group(self, name: str, userids: list[str]) -> dict:
+        self.created.append((name, list(userids)))
+        return {"ok": True, "chat_id": "room-1"}
+
+    def send_group(self, chat_id: str, content: str, mention_userids: list[str] | None = None) -> dict:
+        _ = chat_id, mention_userids
+        self.messages.append(content)
+        return {"ok": True}
+
+
+class PullTests(unittest.TestCase):
+    def test_sea_and_land_share_one_group_without_tms(self) -> None:
+        ledger = _staff_ledger()
+        group = _Group()
+        flow = MultiTextInquiryFlow(ledger, group_client=group)
+        replies: list[str] = []
+
+        def reply(body: str, extra: dict | None = None) -> str:
+            _ = extra
+            replies.append(body)
+            return "ok"
+
+        phase = flow.on_text(sender_id="sales", text=SEA_LAND_READY, reply=reply)
+        if phase == "clarify":
+            self.fail(replies[-1])
+        self.assertEqual(phase, "wait_confirm")
+        phase = flow.on_text(sender_id="sales", text="确定", reply=reply)
+        self.assertEqual(phase, "wait_collab")
+        self.assertIn("拉产品进群协同", replies[-1])
+        self.assertNotIn("TMS", replies[-1])
+        phase = flow.on_text(sender_id="sales", text="拉产品进群协同", reply=reply)
+        self.assertEqual(phase, "collab_group")
+        self.assertEqual(len(group.created), 1)
+        _name, userids = group.created[0]
+        self.assertIn("sea1", userids)
+        self.assertIn("land1", userids)
+        brief = group.messages[-1]
+        self.assertIn("段1：海运", brief)
+        self.assertIn("段2：陆运", brief)
+        self.assertIn("运输类型（必填）", brief)
+        self.assertNotIn("TMS报价", brief)
+        brief_text = group_brief(
+            work_order_no="WO1",
+            segments=flow._segments(flow.session_of("sales")),
+        )
+        self.assertNotIn("TMS", brief_text)
+
+
+class CardPullTests(unittest.TestCase):
+    """点卡走 HTTP 进程：内存没有 Worker 书签，不能当空运查价失败丢掉。"""
+
+    def _open_wait_collab(self, flow: MultiTextInquiryFlow, sender_id: str) -> str:
+        replies: list[str] = []
+
+        def reply(body: str, extra: dict | None = None) -> str:
+            _ = extra
+            replies.append(body)
+            return "ok"
+
+        phase = flow.on_text(sender_id=sender_id, text=SEA_LAND_READY, reply=reply)
+        self.assertEqual(phase, "wait_confirm", replies[-1] if replies else "")
+        phase = flow.on_text(sender_id=sender_id, text="确定", reply=reply)
+        self.assertEqual(phase, "wait_collab")
+        return flow.session_of(sender_id).work_order_no
+
+    def test_card_click_without_memory_creates_one_group(self) -> None:
+        """模拟 HTTP 刚起来：进程内没有书签，只靠卡上工单号从主账收回。"""
+        from agent.channel.queue import MemoryMessageStore
+        from agent.handlers.card_action import handle_card_action
+        from agent.policy import inquiry_copy as copy
+
+        ledger = _staff_ledger()
+        group = _Group()
+        worker = MultiTextInquiryFlow(ledger, group_client=group)
+        wo = self._open_wait_collab(worker, "sales-card")
+        http = MultiTextInquiryFlow(ledger, group_client=group)
+        self.assertIsNone(http.session_of("sales-card"))
+        result = handle_card_action(
+            sender_id="sales-card",
+            thread_id="",
+            wait_version=0,
+            action=copy.BTN_PULL_COLLAB,
+            payload={"task_id": f"multi_pull-{wo}-x", "event_key": f"pull_collab:{wo}"},
+            flow=http,
+            store=MemoryMessageStore(),
+        )
+        self.assertEqual(result, "collab_group")
+        self.assertEqual(len(group.created), 1)
+        self.assertIn("sea1", group.created[0][1])
+        self.assertIn("land1", group.created[0][1])
+
+    def test_pick_card_flow_uses_multi_not_air(self) -> None:
+        """点卡默认空运会把 MULTI 收回成 tms_miss；必须先认多段。"""
+        from agent.handlers.card_action import _pick_card_flow
+        from agent.policy.multi_text_flow import get_multi_text_flow, reset_multi_text_flow_for_test
+
+        flow = reset_multi_text_flow_for_test()
+        ledger = _staff_ledger()
+        flow._ledger = ledger
+        group = _Group()
+        flow._group_client = group
+        wo = self._open_wait_collab(flow, "sales-pick")
+        flow._by_sender.clear()
+        flow._by_ticket.clear()
+        picked = _pick_card_flow(
+            "sales-pick",
+            {"task_id": f"multi_pull-{wo}-y", "event_key": f"pull_collab:{wo}"},
+        )
+        self.assertIs(picked, get_multi_text_flow())
+        sess = picked.session_for_action(
+            "sales-pick",
+            {"task_id": f"multi_pull-{wo}-y", "event_key": f"pull_collab:{wo}"},
+        )
+        self.assertIsNotNone(sess)
+        self.assertEqual((sess.business_line or "").upper(), "MULTI")
+        self.assertEqual(sess.phase, "wait_collab")
+
+
+class DispatchClarifyTests(unittest.TestCase):
+    def test_multi_segment_intent_is_text_inquiry(self) -> None:
+        from agent.routing.dispatch import HANDLER_REGISTRY, _wrap_text_inquiry
+
+        self.assertIs(HANDLER_REGISTRY["multi_segment_transport"], _wrap_text_inquiry)
+
+    def test_clarify_reply_stays_on_multi_even_if_mentions_air(self) -> None:
+        from agent.handlers.text_inquiry import _pick_text_flow
+        from agent.policy.multi_text_flow import get_multi_text_flow
+
+        flow = get_multi_text_flow()
+        replies: list[str] = []
+        phase = flow.on_text(
+            sender_id="wu-clarify",
+            text="段1：陆运\n段2：空运",
+            reply=lambda t, extra=None: replies.append(t) or "ok",
+        )
+        self.assertEqual(phase, "clarify")
+        picked = _pick_text_flow(sender_id="wu-clarify", text=CLARIFY_REPLY)
+        self.assertIs(picked, flow)
+        self.assertEqual((flow.session_of("wu-clarify").phase or ""), "clarify")
+
+    def test_manual_group_bookmark_does_not_steal_new_land_sea(self) -> None:
+        from agent.handlers.text_inquiry import _pick_text_flow, handle_text_inquiry
+        from agent.channel.wecom.models import InboundMessage
+        from agent.policy.multi_text_flow import reset_multi_text_flow_for_test
+
+        flow = reset_multi_text_flow_for_test()
+        replies: list[str] = []
+        flow.on_text(
+            sender_id="wu-attach",
+            text=SEA_AIR_READY,
+            reply=lambda t, extra=None: replies.append(t) or "ok",
+        )
+        flow.on_text(
+            sender_id="wu-attach",
+            text="确定",
+            reply=lambda t, extra=None: replies.append(t) or "ok",
+        )
+        self.assertEqual((flow.session_of("wu-attach").phase or ""), "wait_manual_group")
+        old_wo = flow.session_of("wu-attach").work_order_no
+        picked = _pick_text_flow(
+            sender_id="wu-attach",
+            text="段1：陆运\n运输类型：国内运输拼车\n段2：海运\n起运港：宁波",
+        )
+        self.assertIs(picked, flow)
+
+        class _Store:
+            def __init__(self) -> None:
+                self.texts: list[str] = []
+
+            def enqueue_outbound(self, **kwargs: object) -> tuple[bool, str]:
+                self.texts.append(str(kwargs.get("content") or ""))
+                return True, "ok"
+
+        store = _Store()
+        phase = handle_text_inquiry(
+            InboundMessage(
+                sender_id="wu-attach",
+                message_id="attach-land-sea-1",
+                content="段1：陆运\n运输类型：国内运输拼车\n段2：海运\n起运港：宁波港\n目的港：鹿特丹",
+            ),
+            store=store,
+        )
+        self.assertNotEqual(phase, "wait_manual_group")
+        self.assertTrue(store.texts)
+        self.assertNotIn(old_wo, store.texts[-1])
+        self.assertNotIn("请手动拉相关人员和智能体", store.texts[-1])
+
+    def test_spoken_land_after_manual_group_leaves_multi(self) -> None:
+        from agent.handlers.text_inquiry import _pick_text_flow
+        from agent.policy.land_text_flow import get_land_text_flow
+        from agent.policy.multi_text_flow import reset_multi_text_flow_for_test
+
+        flow = reset_multi_text_flow_for_test()
+        replies: list[str] = []
+        flow.on_text(
+            sender_id="wu-land",
+            text=SEA_AIR_READY,
+            reply=lambda t, extra=None: replies.append(t) or "ok",
+        )
+        flow.on_text(
+            sender_id="wu-land",
+            text="确定",
+            reply=lambda t, extra=None: replies.append(t) or "ok",
+        )
+        self.assertEqual((flow.session_of("wu-land").phase or ""), "wait_manual_group")
+        picked = _pick_text_flow(sender_id="wu-land", text="陆运 国内运输拼车")
+        self.assertIs(picked, get_land_text_flow())
+        self.assertIsNone(flow.session_of("wu-land"))
+
+
+if __name__ == "__main__":
+    unittest.main()

inquiry-agent/tests/test_quote_template_match.py

@@ -168,6 +168,32 @@ class QuoteTemplateMatchTests(unittest.TestCase):
         self.assertEqual(hit.match_type, "default")
         self.assertEqual(hit.name, "陆运默认")
 
+    def test_land_admin_keyword_domestic_lcl_matches_transport_type(self) -> None:
+        """后台陆运拼车模板关键词是「国内拼车」，对得上字段「国内运输拼车」。"""
+        hit = pick_quote_template(
+            templates=[
+                {
+                    "templateId": "hk",
+                    "name": "中港",
+                    "bizType": "陆运",
+                    "keywords": ["中港", "陆运"],
+                    "isDefault": False,
+                },
+                {
+                    "templateId": "ltl",
+                    "name": "陆运国内运输拼车报价单",
+                    "bizType": "陆运",
+                    "keywords": ["国内拼车", "国内零担"],
+                    "isDefault": False,
+                },
+            ],
+            business_line="LAND",
+            facts={"运输类型": "国内运输拼车", "起运港": "广州", "目的港": "深圳"},
+        )
+        self.assertIsNotNone(hit)
+        self.assertEqual(hit.match_type, "keyword")
+        self.assertEqual(hit.name, "陆运国内运输拼车报价单")
+
     def test_land_none_without_default(self) -> None:
         hit = pick_quote_template(
             templates=[

inquiry-api/jeecg-module-inquiry/src/main/java/org/jeecg/modules/inquiry/service/impl/InquiryWorkOrderServiceImpl.java

@@ -224,6 +224,22 @@ public class InquiryWorkOrderServiceImpl extends ServiceImpl<InquiryTicketMapper
         return out;
     }
 
+    /**
+     * 按企微 userid 取员工管理姓名。建单和智能体读单都走这里，禁止把岗位词「销售」当人名。
+     * 查不到返回空，调用方再用 userid。副作用：只读员工表。
+     */
+    private String resolveStaffDisplayName(String salesWecomId) {
+        String uid = StringUtils.trimToEmpty(salesWecomId);
+        if (uid.isEmpty()) {
+            return "";
+        }
+        InquiryStaffRole row = staffRoleMapper.selectOne(new LambdaQueryWrapper<InquiryStaffRole>()
+                .eq(InquiryStaffRole::getWecomId, uid)
+                .eq(InquiryStaffRole::getStatus, "active")
+                .last("LIMIT 1"));
+        return row == null ? "" : StringUtils.trimToEmpty(row.getName());
+    }
+
     /** 批量按企微 userid 取员工姓名，避免列表 N+1。 */
     private Map<String, String> loadStaffNames(List<InquiryTicket> tickets) {
         Set<String> wecomIds = new HashSet<>();
@@ -479,6 +495,7 @@ public class InquiryWorkOrderServiceImpl extends ServiceImpl<InquiryTicketMapper
         InquiryTicket ticket = new InquiryTicket();
         ticket.setWorkOrderNo(nextWorkOrderNo());
         ticket.setSalesWecomId(req.getSalesWecomId());
+        ticket.setSalesName(resolveStaffDisplayName(req.getSalesWecomId()));
         ticket.setBusinessLine(req.getBusinessLine().trim().toUpperCase());
         ticket.setBizTypeLabel(bizLabel(ticket.getBusinessLine()));
         ticket.setOrigin(facts.get("起运港"));
@@ -805,8 +822,13 @@ public class InquiryWorkOrderServiceImpl extends ServiceImpl<InquiryTicketMapper
         out.put("ok", true);
         out.put("status", status);
         out.put("business_line", StringUtils.defaultString(ticket.getBusinessLine()));
-        out.put("sales_name", StringUtils.defaultString(ticket.getSalesName()));
-        out.put("salesName", StringUtils.defaultString(ticket.getSalesName()));
+        String salesWecomId = StringUtils.defaultString(ticket.getSalesWecomId());
+        out.put("salesWecomId", salesWecomId);
+        out.put("sales_wecom_id", salesWecomId);
+        String salesName = InquiryTicketDisplaySupport.resolveSalesDisplayName(
+                ticket.getSalesName(), salesWecomId, resolveStaffDisplayName(salesWecomId));
+        out.put("sales_name", salesName);
+        out.put("salesName", salesName);
         out.put("facts", facts);
         out.put("quote", quote);
         out.put("tms_quote", tmsQuote);

inquiry-api/jeecg-module-inquiry/src/main/java/org/jeecg/modules/inquiry/template/QuoteTemplateMatcher.java

@@ -137,6 +137,40 @@ public final class QuoteTemplateMatcher {
     /**
      * 业务线名、过宽的「国内」不当关键词。
      */
+    /**
+     * 后台陆运模板写「国内拼车」，询价字段是「国内运输拼车」。
+     */
+    private static final Map<String, String[]> LAND_KEYWORD_ALIASES = new LinkedHashMap<>();
+
+    static {
+        LAND_KEYWORD_ALIASES.put("国内拼车", new String[] {"国内运输拼车", "拼车"});
+        LAND_KEYWORD_ALIASES.put("国内零担", new String[] {"国内运输拼车", "国内长途/零担", "零担"});
+        LAND_KEYWORD_ALIASES.put("国内整车", new String[] {"国内运输整车", "整车"});
+        LAND_KEYWORD_ALIASES.put("跨境整车", new String[] {"跨境整车"});
+        LAND_KEYWORD_ALIASES.put("跨境集拼", new String[] {"跨境集拼"});
+        LAND_KEYWORD_ALIASES.put("跨境零担", new String[] {"跨境集拼"});
+    }
+
+    static boolean hitNeedle(String text, String kw) {
+        if (text == null || text.isEmpty() || kw == null || kw.isEmpty()) {
+            return false;
+        }
+        String hay = text.toLowerCase(Locale.ROOT);
+        if (hay.contains(kw.toLowerCase(Locale.ROOT))) {
+            return true;
+        }
+        String[] aliases = LAND_KEYWORD_ALIASES.get(kw);
+        if (aliases == null) {
+            return false;
+        }
+        for (String alias : aliases) {
+            if (hay.contains(alias.toLowerCase(Locale.ROOT))) {
+                return true;
+            }
+        }
+        return false;
+    }
+
     static boolean skipKeyword(String kw) {
         if (kw == null || kw.isEmpty()) {
             return true;
@@ -161,11 +195,11 @@ public final class QuoteTemplateMatcher {
                 continue;
             }
             String k = kw.toLowerCase(Locale.ROOT);
-            if (!type.isEmpty() && type.contains(k)) {
+            if (hitNeedle(type, kw) ) {
                 typeHits++;
                 length += k.length();
                 any = true;
-            } else if (hay.contains(k)) {
+            } else if (hitNeedle(hay, kw)) {
                 length += k.length();
                 any = true;
             }

inquiry-api/jeecg-module-inquiry/src/test/java/org/jeecg/modules/inquiry/template/QuoteTemplateMatcherTest.java

@@ -133,6 +133,21 @@ public class QuoteTemplateMatcherTest {
         assertEquals("跨境整车", hit.get("name"));
     }
 
+    @Test
+    public void landAdminKeywordDomesticLclMatchesTransportType() {
+        List<Map<String, Object>> list = new ArrayList<>();
+        list.add(tpl("中港", "陆运", false, "中港", "陆运"));
+        list.add(tpl("陆运国内运输拼车报价单", "陆运", false, "国内拼车", "国内零担"));
+        Map<String, Object> facts = new HashMap<>();
+        facts.put("运输类型", "国内运输拼车");
+        facts.put("起运港", "广州");
+        facts.put("目的港", "深圳");
+        Map<String, Object> hit = QuoteTemplateMatcher.pick(list, "LAND", facts, null);
+        assertNotNull(hit);
+        assertEquals("keyword", hit.get("matchType"));
+        assertEquals("陆运国内运输拼车报价单", hit.get("name"));
+    }
+
     @Test
     public void landFallsBackToDefault() {
         List<Map<String, Object>> list = new ArrayList<>();