落地私聊图片询价：识图当口播复用三线文字流程，图上航线三字码保持原样，避免被改写成中文地名。

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

inquiry-agent/agent/channel/outbox/sender.py

@@ -67,6 +67,21 @@ class WeComAppClient:
         副作用：出站 HTTP 调企微；失败返回 ok=False，由 outbox 重试。
         """
         extra = dict(payload or {})
+        if extra.get("msgtype") == "image":
+            image_result = self.send_image(
+                touser=touser,
+                filename=str(extra.get("filename") or "image.png"),
+                filepath=str(extra.get("filepath") or ""),
+                file_b64=str(extra.get("file_b64") or extra.get("fileBase64") or ""),
+            )
+            if image_result.ok:
+                return image_result
+            logger.warning(
+                "图片发送失败，回退文本 err=%s:%s",
+                image_result.errcode,
+                image_result.errmsg,
+            )
+            return self.send_text(touser=touser, content=content)
         if extra.get("msgtype") == "file":
             file_result = self.send_file(
                 touser=touser,
@@ -213,7 +228,9 @@ class WeComAppClient:
                 return WeComSendResult(ok=False, errcode=-1, errmsg=f"b64:{exc}")
         if not raw:
             return WeComSendResult(ok=False, errcode=-1, errmsg="file_empty")
-        media_id, up = self._upload_media(filename=filename or "quote.xlsx", content=raw)
+        media_id, up = self._upload_media(
+            filename=filename or "quote.xlsx", content=raw, media_type="file"
+        )
         if not media_id:
             return up
         return self._post_message(
@@ -226,8 +243,64 @@ class WeComAppClient:
             preview=f"[file]{filename}",
         )
 
-    def _upload_media(self, *, filename: str, content: bytes) -> tuple[str, WeComSendResult]:
-        """cgi-bin/media/upload；密钥不写日志。"""
+    def send_image(
+        self,
+        *,
+        touser: str,
+        filename: str,
+        filepath: str = "",
+        file_b64: str = "",
+    ) -> WeComSendResult:
+        """
+        先上传 image 素材再发图片消息，聊天里直接看图。
+
+        字节来自本机路径或 Base64。禁止在回调线程调用。
+        """
+        if not touser:
+            return WeComSendResult(ok=False, errcode=-1, errmsg="touser 为空")
+        raw = b""
+        if filepath:
+            try:
+                from pathlib import Path
+
+                raw = Path(filepath).read_bytes()
+            except OSError as exc:
+                return WeComSendResult(ok=False, errcode=-1, errmsg=f"read_file:{exc}")
+        elif file_b64:
+            import base64
+
+            try:
+                raw = base64.b64decode(file_b64)
+            except Exception as exc:  # noqa: BLE001
+                return WeComSendResult(ok=False, errcode=-1, errmsg=f"b64:{exc}")
+        if not raw:
+            return WeComSendResult(ok=False, errcode=-1, errmsg="image_empty")
+        name = filename or "image.png"
+        media_id, up = self._upload_media(
+            filename=name, content=raw, media_type="image"
+        )
+        if not media_id:
+            return up
+        return self._post_message(
+            {
+                "touser": touser,
+                "msgtype": "image",
+                "agentid": self._agent_id,
+                "image": {"media_id": media_id},
+            },
+            preview=f"[image]{name}",
+        )
+
+    def _upload_media(
+        self,
+        *,
+        filename: str,
+        content: bytes,
+        media_type: str = "file",
+    ) -> tuple[str, WeComSendResult]:
+        """cgi-bin/media/upload；type=file 或 image。密钥不写日志。"""
+        kind = (media_type or "file").strip() or "file"
+        mime = "image/png" if kind == "image" else "application/octet-stream"
         if not self._corp_id or not self._secret:
             return "", WeComSendResult(ok=False, errcode=-1, errmsg="WECOM_CORP_ID/SECRET 未配置")
         try:
@@ -240,8 +313,8 @@ class WeComAppClient:
             with httpx.Client(timeout=max(self._timeout, 30.0)) as client:
                 resp = client.post(
                     url,
-                    params={"access_token": token, "type": "file"},
-                    files={"media": (filename, content, "application/octet-stream")},
+                    params={"access_token": token, "type": kind},
+                    files={"media": (filename, content, mime)},
                 )
                 data = resp.json()
         except Exception as exc:  # noqa: BLE001
@@ -250,7 +323,7 @@ class WeComAppClient:
         errcode = int(data.get("errcode") or 0)
         media_id = str(data.get("media_id") or "")
         if errcode != 0 or not media_id:
-            logger.warning("企微上传文件失败 errcode=%s", errcode)
+            logger.warning("企微上传素材失败 type=%s errcode=%s", kind, errcode)
             return "", WeComSendResult(
                 ok=False, errcode=errcode, errmsg=str(data.get("errmsg") or "upload_fail"), raw=data
             )
@@ -555,6 +628,42 @@ class WeComAppClient:
             self._token_expire_at = time.time() + expires_in
         return token
 
+    def download_media(self, media_id: str) -> tuple[bytes, str]:
+        """
+        下载企微临时素材（cgi-bin/media/get）。
+
+        返回 (字节, 错误码)。失败时字节为空。禁止把 token 写入日志。
+        调用：Worker 识别槽，禁止回调线程同步下载。
+        """
+        mid = (media_id or "").strip()
+        if not mid:
+            return b"", "media_empty"
+        try:
+            token = self._get_token()
+        except Exception:
+            logger.exception("下载素材取 token 失败")
+            return b"", "token_failed"
+        url = f"{self._api_base}/cgi-bin/media/get"
+        try:
+            with httpx.Client(timeout=self._timeout) as client:
+                resp = client.get(url, params={"access_token": token, "media_id": mid})
+            ctype = (resp.headers.get("content-type") or "").lower()
+            raw = resp.content or b""
+            if "application/json" in ctype or raw[:1] in (b"{", b"["):
+                try:
+                    data = resp.json()
+                except Exception:
+                    data = {}
+                if isinstance(data, dict) and int(data.get("errcode") or 0) != 0:
+                    return b"", str(data.get("errmsg") or "media_get_fail")
+                return b"", "media_json_body"
+            if len(raw) < 32:
+                return b"", "media_too_small"
+            return raw, ""
+        except Exception:
+            logger.exception("下载素材失败")
+            return b"", "media_download_failed"
+
 
 # 企微侧明确不可重试的错误（IP 白名单、非法 userid 等）
 PERMANENT_SEND_ERRORS = frozenset(

inquiry-agent/agent/channel/runtime.py

@@ -75,6 +75,11 @@ class ChannelRuntime:
                     from agent.routing.decision import RouteDecision
 
                     decision = RouteDecision(intent="ordinary_text_other")
+                elif (message.msg_type or "text").lower() == "image":
+                    from agent.routing.decision import RouteDecision
+
+                    # 私聊图片不进 DeepSeek A（正文为空会被判闲聊）。
+                    decision = RouteDecision(intent="image_inquiry", confidence=1.0)
                 else:
                     decision = self._router.route_inbound(
                         text=message.content or "",

inquiry-agent/agent/handlers/__init__.py

@@ -2,10 +2,16 @@
 Handler 包：一类意图 / 一类卡片一个模块。
 
 禁止：单文件处理全部卡片；禁止把 Schema 校验复制粘贴进每个 handler（放 schema/policy）。
-当前：文字询价空运/海运闭环 + 海运应用群协同 + 空运 BOT 群文字/附件 + echo 联调残留。
+当前：文字询价空运/海运/陆运闭环 + 私聊图片询价 + 海运应用群协同 + 空运 BOT 群文字/附件 + echo 联调残留。
 """
 
 from agent.handlers.echo_text import handle_echo_text
+from agent.handlers.image_inquiry import handle_image_inquiry
 from agent.handlers.text_inquiry import handle_text_inquiry, handle_text_other
 
-__all__ = ["handle_echo_text", "handle_text_inquiry", "handle_text_other"]
+__all__ = [
+    "handle_echo_text",
+    "handle_image_inquiry",
+    "handle_text_inquiry",
+    "handle_text_other",
+]

inquiry-agent/agent/handlers/image_inquiry.py

@@ -0,0 +1,409 @@
+"""
+意图：私聊图片询价。
+
+本文件职责：收图 → 回「正在识别」→ 千问抄可见原文 → 拼口播 → 新开后复用文字询价。
+禁止：千问填最终缺项；另写海运/空运/陆运图片专线；回调线程同步等千问；群图走本 Handler。
+"""
+
+from __future__ import annotations
+
+import logging
+import time
+from typing import Any, Callable, Optional
+
+from agent.channel.queue import get_message_store
+from agent.channel.wecom.models import InboundMessage
+from agent.policy import inquiry_copy as copy
+from agent.policy.image_wave import get_image_wave_book
+
+logger = logging.getLogger(__name__)
+
+VisionFn = Callable[..., dict[str, Any]]
+DownloadFn = Callable[[str], tuple[bytes, str]]
+PutFn = Callable[..., dict[str, Any]]
+
+
+def drop_all_private_bookmarks(sender_id: str) -> None:
+    """三线当前书签都丢掉；已出号仍能按工单号找回。"""
+    from agent.policy.air_text_flow import get_air_text_flow
+    from agent.policy.land_text_flow import get_land_text_flow
+    from agent.policy.sea_text_flow import get_sea_text_flow
+
+    get_air_text_flow().drop_current_bookmark(sender_id)
+    get_sea_text_flow().drop_current_bookmark(sender_id)
+    get_land_text_flow().drop_current_bookmark(sender_id)
+
+
+def is_active_image_wave(sender_id: str) -> bool:
+    """业务结果尚未发出的图片波次。"""
+    return get_image_wave_book().active(sender_id) is not None
+
+
+def append_wave_text(sender_id: str, text: str) -> bool:
+    """认图完成前的字并进本波，不走文字询价。"""
+    return get_image_wave_book().append_follow_text(sender_id, text)
+
+
+def note_private_text(sender_id: str, text: str, *, now: Optional[float] = None) -> None:
+    """文字询价记下最近一句，供 10 秒内发图带走。"""
+    get_image_wave_book().note_text(sender_id, text, now=now if now is not None else time.time())
+
+
+def handle_image_inquiry(
+    message: InboundMessage,
+    *,
+    store=None,
+    vision_fn: Optional[VisionFn] = None,
+    download_fn: Optional[DownloadFn] = None,
+    put_fn: Optional[PutFn] = None,
+    process_now: bool = True,
+) -> str:
+    """
+    私聊一张图。inbox 线程只开波、回识别中；认图在 Worker 或单测 vision_fn。
+
+    须已有 sender_id。群消息不要调用本函数。
+    """
+    if not message.has_sender():
+        logger.warning("image_inquiry 拒绝：无 sender_id")
+        return "reject_no_sender"
+    if (message.chat_type or "single") == "group":
+        logger.info("image_inquiry 忽略群图 chat=%s", message.chat_id)
+        return "ignore_group_image"
+
+    book = get_image_wave_book()
+    msg_store = store or get_message_store()
+    media_id = str((message.media or {}).get("media_id") or "")
+    filename = str((message.media or {}).get("filename") or "image.jpg")
+    wave, how = book.open_or_join(message.sender_id, media_id, now=time.time())
+    if how in {"opened", "new_after_done"}:
+        drop_all_private_bookmarks(message.sender_id)
+        msg_store.enqueue_outbound(
+            touser=message.sender_id,
+            content=copy.IMAGE_RECOGNIZING,
+            dedupe_key=f"img-ack:{wave.wave_id}",
+        )
+        book.mark_ack(message.sender_id)
+        logger.info("image_inquiry 开波 how=%s wave=%s", how, wave.wave_id)
+    else:
+        logger.info(
+            "image_inquiry 并入 how=%s wave=%s pending=%s",
+            how,
+            wave.wave_id,
+            wave.pending,
+        )
+
+    if process_now and vision_fn is not None:
+        _recognize_one(
+            sender_id=message.sender_id,
+            media_id=media_id,
+            filename=filename,
+            vision_fn=vision_fn,
+            download_fn=download_fn,
+            put_fn=put_fn,
+        )
+        wave = book.active(message.sender_id) or wave
+        if wave.pending <= 0:
+            return finish_wave(message.sender_id, store=msg_store, seed_message=message)
+        return "recognizing"
+
+    if vision_fn is None:
+        try:
+            from agent.jobs.recognition import enqueue_recognition
+
+            enqueue_recognition(
+                kind="vision",
+                payload={
+                    "sender_id": message.sender_id,
+                    "media_id": media_id,
+                    "filename": filename,
+                    "message_id": message.message_id,
+                    "wave_id": wave.wave_id,
+                },
+                idempotency_key=message.message_id or media_id,
+            )
+        except Exception:
+            logger.exception("image_inquiry 入识别队失败 sender=%s", message.sender_id)
+            _recognize_one(
+                sender_id=message.sender_id,
+                media_id=media_id,
+                filename=filename,
+                vision_fn=None,
+                download_fn=download_fn,
+                put_fn=put_fn,
+            )
+            return finish_wave(message.sender_id, store=msg_store, seed_message=message)
+        return "recognizing"
+
+    return "recognizing"
+
+
+def flush_active_wave(
+    sender_id: str,
+    *,
+    store=None,
+    vision_fn: Optional[VisionFn] = None,
+    download_fn: Optional[DownloadFn] = None,
+    put_fn: Optional[PutFn] = None,
+    seed_message: Optional[InboundMessage] = None,
+) -> str:
+    """单测：两张图都 join 后再认，只出一次业务结果。"""
+    book = get_image_wave_book()
+    wave = book.active(sender_id)
+    if wave is None:
+        return "no_wave"
+    pending_ids = list(wave.media_ids[-wave.pending :]) if wave.pending else []
+    for mid in pending_ids:
+        _recognize_one(
+            sender_id=sender_id,
+            media_id=mid,
+            filename="",
+            vision_fn=vision_fn,
+            download_fn=download_fn,
+            put_fn=put_fn,
+        )
+    return finish_wave(sender_id, store=store, seed_message=seed_message)
+
+
+def finish_queued_image(payload: dict[str, Any]) -> dict[str, Any]:
+    """
+    Worker 识别槽：认完一张，pending 为 0 则出业务结果。
+
+    payload.__vision / __bytes 仅单测注入，生产不得依赖。
+    """
+    sender_id = str(payload.get("sender_id") or "")
+    media_id = str(payload.get("media_id") or "")
+    filename = str(payload.get("filename") or "image.jpg")
+    vision_hook = payload.get("__vision")
+    raw = payload.get("__bytes")
+
+    def _vision(**_kwargs: Any) -> dict[str, Any]:
+        if isinstance(vision_hook, dict):
+            return dict(vision_hook)
+        from agent.config import get_settings
+        from agent.llm.mode_vision import invoke_vision
+
+        settings = get_settings()
+        return invoke_vision(
+            payload=_kwargs.get("payload") or {},
+            allow_network=bool(settings.llm_allow_network and settings.llm_data_usage_confirmed),
+        )
+
+    def _download(mid: str) -> tuple[bytes, str]:
+        if isinstance(raw, (bytes, bytearray)):
+            return bytes(raw), ""
+        from agent.channel.outbox.sender import WeComAppClient
+        from agent.config import get_settings
+
+        settings = get_settings()
+        client = WeComAppClient(
+            corp_id=settings.wecom_corp_id,
+            secret=settings.wecom_secret,
+            agent_id=int(settings.wecom_agent_id or "0") or 0,
+        )
+        return client.download_media(mid)
+
+    _recognize_one(
+        sender_id=sender_id,
+        media_id=media_id,
+        filename=filename,
+        vision_fn=_vision,
+        download_fn=_download,
+        put_fn=None,
+    )
+    book = get_image_wave_book()
+    wave = book.active(sender_id)
+    if wave is None:
+        logger.warning(
+            "image_inquiry Worker 认完但无活跃波次 sender=%s media=%s（多半 HTTP/Worker 账不一致）",
+            sender_id,
+            media_id,
+        )
+        return {"ok": False, "phase": "no_active_wave", "pending": 0}
+    if wave.pending > 0:
+        return {"ok": True, "phase": "recognizing", "pending": wave.pending}
+    phase = finish_wave(sender_id, seed_message=InboundMessage(
+        sender_id=sender_id,
+        message_id=str(payload.get("message_id") or media_id),
+        content="",
+        msg_type="image",
+        media={"media_id": media_id, "filename": filename},
+    ))
+    return {"ok": True, "phase": phase}
+
+
+def _recognize_one(
+    *,
+    sender_id: str,
+    media_id: str,
+    filename: str,
+    vision_fn: Optional[VisionFn],
+    download_fn: Optional[DownloadFn],
+    put_fn: Optional[PutFn],
+) -> None:
+    """下载、可选落对象存储、读图，写入波次。"""
+    import base64
+
+    from agent.llm.mode_vision import sniff_image_mime
+
+    book = get_image_wave_book()
+    err = ""
+    raw = b""
+    if download_fn is None:
+        raw, err = b"", "no_download"
+    else:
+        raw, err = download_fn(media_id)
+    if not raw:
+        logger.info(
+            "image_inquiry 无字节 sender=%s err=%s",
+            sender_id,
+            err or "media_empty",
+        )
+        book.add_vision(
+            sender_id,
+            "",
+            object_key="",
+            filename=filename,
+            error=err or "media_empty",
+        )
+        return
+    mime = sniff_image_mime(raw) or "image/jpeg"
+    object_key = ""
+    if put_fn is not None:
+        key = f"inquiry-image/{sender_id}/{media_id or filename}"
+        try:
+            stored = put_fn(key=key, data=raw, content_type=mime)
+            if stored.get("ok"):
+                object_key = str(stored.get("key") or key)
+        except Exception:
+            logger.exception("原图落存储失败 sender=%s", sender_id)
+    else:
+        object_key = f"inquiry-image/{sender_id}/{media_id or filename}"
+        try:
+            from agent.config import get_settings
+            from agent.storage.s3_client import S3ObjectStorage
+
+            S3ObjectStorage.from_settings(get_settings(), allow_network=False).put_bytes(
+                key=object_key, data=raw, content_type=mime
+            )
+        except Exception:
+            logger.info("原图 dry_run 存储跳过 sender=%s", sender_id)
+
+    vision: dict[str, Any] = {"ok": False, "text": "", "empty": True, "error": err or "no_vision"}
+    if vision_fn is not None:
+        b64 = base64.b64encode(raw).decode("ascii")
+        vision = vision_fn(payload={"image_b64": b64, "mime": mime, "media_id": media_id}) or vision
+    else:
+        from agent.config import get_settings
+        from agent.llm.mode_vision import invoke_vision
+
+        settings = get_settings()
+        vision = invoke_vision(
+            payload={"image_b64": base64.b64encode(raw).decode("ascii"), "mime": mime},
+            allow_network=bool(settings.llm_allow_network and settings.llm_data_usage_confirmed),
+        )
+    text = str(vision.get("text") or "")
+    if not vision.get("ok"):
+        err = str(vision.get("error") or err or "vision_failed")
+    logger.info(
+        "image_inquiry 千问 bytes=%s mime=%s ok=%s empty=%s text_len=%s err=%s",
+        len(raw),
+        mime,
+        bool(vision.get("ok")),
+        bool(vision.get("empty")),
+        len(text.strip()),
+        err or "-",
+    )
+    book.add_vision(
+        sender_id,
+        text,
+        object_key=object_key,
+        filename=filename,
+        error=err,
+    )
+
+
+def finish_wave(
+    sender_id: str,
+    *,
+    store=None,
+    seed_message: Optional[InboundMessage] = None,
+) -> str:
+    """
+    本波图都认完：拼口播、当文字询价、必要时把说明和补问合成一条。
+    """
+    from agent.handlers.text_inquiry import handle_text_inquiry
+    from agent.policy.air_text_flow import get_air_text_flow
+    from agent.policy.land_text_flow import get_land_text_flow
+    from agent.policy.sea_text_flow import get_sea_text_flow
+
+    book = get_image_wave_book()
+    wave = book.active(sender_id)
+    oral = book.compose_oral(sender_id)
+    reason = ""
+    if not oral.strip():
+        reason = copy.IMAGE_TECH_FAIL if (wave and wave.errors) else copy.IMAGE_UNREADABLE
+
+    msg_store = store or get_message_store()
+    prefix_used = {"v": False}
+
+    class _StoreProxy:
+        def enqueue_outbound(self, **kwargs):
+            body = str(kwargs.get("content") or "")
+            extra = dict(kwargs.get("payload") or {})
+            if reason and not prefix_used["v"] and body:
+                body = copy.image_fail_then_clarify(reason, body)
+                kwargs["content"] = body
+                prefix_used["v"] = True
+            return msg_store.enqueue_outbound(
+                touser=kwargs.get("touser") or sender_id,
+                content=body,
+                dedupe_key=str(kwargs.get("dedupe_key") or ""),
+                payload=extra,
+            )
+
+    inbound = InboundMessage(
+        sender_id=sender_id,
+        message_id=(seed_message.message_id if seed_message else f"img:{sender_id}:{int(time.time())}"),
+        content=oral,
+        msg_type="text",
+        media=dict(seed_message.media) if seed_message else {},
+        raw=dict(seed_message.raw) if seed_message else {},
+    )
+    phase = handle_text_inquiry(inbound, store=_StoreProxy())
+    if reason and not prefix_used["v"]:
+        msg_store.enqueue_outbound(
+            touser=sender_id,
+            content=copy.image_fail_then_clarify(reason, copy.ASK_TRANSPORT),
+            dedupe_key=f"img-empty:{inbound.message_id}",
+        )
+        phase = "need_mode"
+    book.mark_business_done(sender_id)
+    book.forget_text(sender_id)
+    sess = (
+        get_sea_text_flow().session_of(sender_id)
+        or get_air_text_flow().session_of(sender_id)
+        or get_land_text_flow().session_of(sender_id)
+    )
+    wo = (sess.work_order_no if sess else "") or ""
+    if wo and wave:
+        ledger = None
+        for engine in (get_sea_text_flow(), get_air_text_flow(), get_land_text_flow()):
+            hit = engine.session_of(sender_id)
+            if hit and hit.work_order_no == wo:
+                ledger = engine.ledger
+                break
+        if ledger is not None:
+            for i, key in enumerate(wave.object_keys):
+                name = wave.filenames[i] if i < len(wave.filenames) else "image.jpg"
+                try:
+                    ledger.attach_file(
+                        work_order_no=wo,
+                        file_name=name or "image.jpg",
+                        object_key=key,
+                        content_type="image/jpeg",
+                        sender_id=sender_id,
+                    )
+                except Exception:
+                    logger.exception("挂图片附件失败 wo=%s", wo)
+    logger.info("image_inquiry 波次完成 phase=%s oral_len=%s", phase, len(oral))
+    return phase

inquiry-agent/agent/handlers/text_inquiry.py

@@ -21,11 +21,23 @@ from agent.policy.sea_text_flow import get_sea_text_flow
 logger = logging.getLogger(__name__)
 
 
+def _front_bookmark(sess) -> bool:
+    """补问/核对手稿优先于已出票旧书签。空的 need_mode 不算（那是丢会话后的误伤）。"""
+    if sess is None:
+        return False
+    phase = (sess.phase or "").strip()
+    line = (sess.business_line or "").upper()
+    if phase == "need_mode" and not line:
+        return False
+    return phase in {"clarify", "need_mode", "need_land_options", "wait_confirm"}
+
+
 def _pick_text_flow(*, sender_id: str, text: str, injected_mode: str = ""):
     """
     按当前书签或本句运输方式选空运/海运/陆运流程。
 
     陆运核对卡上的确定优先于旧海运书签，避免「确定」被海运/空运吃掉。
+    图片识图在 Worker 写下的补问书签，HTTP 必须从 Redis 读到，否则「托盘」会变成再问运输方式。
     """
     from agent.schema.land_options import looks_like_confirm
 
@@ -38,10 +50,15 @@ def _pick_text_flow(*, sender_id: str, text: str, injected_mode: str = ""):
         if retry:
             return land
     sea = get_sea_text_flow()
+    air = get_air_text_flow()
     sea_sess = sea.session_of(sender_id)
+    land_sess = land.session_of(sender_id)
+    air_sess = air.session_of(sender_id)
+    for sess, flow in ((land_sess, land), (air_sess, air), (sea_sess, sea)):
+        if _front_bookmark(sess):
+            return flow
     if sea_sess and (sea_sess.business_line or "").upper() == "SEA":
         return sea
-    land_sess = land.session_of(sender_id)
     if land_sess and (land_sess.business_line or "").upper() == "LAND":
         return land
     mode = (injected_mode or detect_transport_mode(text) or "").upper()
@@ -49,7 +66,7 @@ def _pick_text_flow(*, sender_id: str, text: str, injected_mode: str = ""):
         return sea
     if mode == "LAND":
         return land
-    return get_air_text_flow()
+    return air
 
 
 def is_pending_land_confirm(message: InboundMessage) -> bool:
@@ -126,6 +143,9 @@ def handle_text_inquiry(
     if not message.has_sender():
         logger.warning("text_inquiry 拒绝：无 sender_id")
         return "reject_no_sender"
+    from agent.handlers.image_inquiry import note_private_text
+
+    note_private_text(message.sender_id, message.content or "")
     settings = get_settings()
     msg_store = store or get_message_store()
     engine = flow or _pick_text_flow(

inquiry-agent/agent/http_app.py

@@ -82,6 +82,7 @@ def main() -> None:
         level=logging.INFO,
         format="%(asctime)s %(levelname)s [%(name)s] %(message)s",
     )
+    logging.getLogger("httpx").setLevel(logging.WARNING)
     settings = get_settings()
     # test 环境默认允许本机注入，便于并存期联调
     if settings.ytd_env == "test" and not settings.dev_inject_enabled:

inquiry-agent/agent/jobs/__init__.py

@@ -19,7 +19,7 @@ from agent.jobs.archive import archive_stream, process_archive_stub
 from agent.jobs.libreoffice import LibreOfficeConverter
 from agent.jobs.pdf_job import pdf_stream, process_pdf_job
 from agent.jobs.quote_fill import fill_quote_workbook
-from agent.jobs.recognition import process_recognition_stub
+from agent.jobs.recognition import process_recognition
 from agent.jobs.wake_consumer import claim_and_apply_wake, wake_stream
 from agent.llm import LlmGateway, LlmMode
 from agent.redis_coord.runtime import RedisRuntime
@@ -211,7 +211,7 @@ class WorkerRuntime:
                     return
                 entry_id, data = claimed[0]
                 try:
-                    process_recognition_stub(data)
+                    process_recognition(data)
                 finally:
                     stream.ack(entry_id)
 

inquiry-agent/agent/jobs/recognition.py

@@ -1,7 +1,7 @@
 """
 识别任务壳：入队 Redis Stream，Worker HTTP 槽认领后跑占位处理。
 
-本文件职责：enqueue / process_stub；正式识别（附件/OCR/字段）后续按 payload.kind 分支。
+本文件职责：识别入队；Worker 槽消费。kind=vision 跑私聊图片波次，其它 kind 仍占位。
 禁止：回调线程同步等识别完成；禁止本模块写主账六态。
 """
 
@@ -62,6 +62,21 @@ def enqueue_recognition(
     return RecognitionEnqueueResult(accepted=True, entry_id=entry_id)
 
 
+def process_recognition(data: dict[str, Any]) -> dict[str, Any]:
+    """
+    Worker 槽：vision 跑图片波次，其它 kind 仍走占位壳。
+
+    禁止在本函数写六态。
+    """
+    kind = str(data.get("kind") or "")
+    payload = data.get("payload") if isinstance(data.get("payload"), dict) else {}
+    if kind == "vision":
+        from agent.handlers.image_inquiry import finish_queued_image
+
+        return finish_queued_image(payload)
+    return process_recognition_stub(data)
+
+
 def process_recognition_stub(data: dict[str, Any]) -> dict[str, Any]:
     """
     识别占位：回显 kind，不读附件、不调模型。

inquiry-agent/agent/llm/extract_text.py

@@ -23,12 +23,36 @@ _MODE_WORDS = (
 _LABEL_LINE = re.compile(
     r"^(起运地|起运港|目的地|目的港|品名|件数|毛重|体积|重量|重量\(KG\)|体积\(CBM\)|"
     r"包装方式|包装类型|货量|货物数量|整柜或拼柜|箱型箱量|报价日期|车型/数量|车型数量|"
-    r"贸易条款|货好时间|商品海关编码|海关编码|HS编码|HS|是否含油|是否含电|是否含磁)\s*[:：]\s*(.+)$"
+    r"贸易条款|货好时间|商品海关编码|海关编码|HS编码|HS|是否含油|是否含电|是否含磁|"
+    r"客户名称|货值|是否为危险品|通关口岸（非必填）|通关口岸)\s*[:：]\s*(.+)$"
 )
 _LABEL_INLINE = re.compile(
     r"(起运地|起运港|目的地|目的港|品名|件数|毛重|体积|重量|重量\(KG\)|体积\(CBM\)|"
     r"包装方式|包装类型|货量|货物数量|整柜或拼柜|箱型箱量|报价日期|车型/数量|车型数量|"
-    r"贸易条款|货好时间|商品海关编码|海关编码|HS编码|HS|是否含油|是否含电|是否含磁)\s*[:：]\s*(\S+)"
+    r"贸易条款|货好时间|商品海关编码|海关编码|HS编码|HS|是否含油|是否含电|是否含磁|"
+    r"客户名称|货值|是否为危险品|通关口岸（非必填）|通关口岸)\s*[:：]\s*(\S+)"
+)
+# 陆运核对卡非必填：销售可能写「通关口岸皇岗」或照抄「通关口岸（非必填）：皇岗」。长词优先。
+_OPTIONAL_INQUIRY_LABELS = (
+    "通关口岸（非必填）",
+    "通关口岸",
+)
+# 群里协同标签，冒号可省略：「客户名称张三」「货值 10万」。长词优先。
+_COLLAB_LABELS = (
+    "是否为危险品",
+    "商品海关编码",
+    "车型/数量",
+    "包装方式",
+    "客户名称",
+    "海关编码",
+    "贸易条款",
+    "货好时间",
+    "是否含油",
+    "是否含电",
+    "是否含磁",
+    "HS编码",
+    "货值",
+    "HS",
 )
 
 
@@ -64,6 +88,23 @@ def parse_labeled_facts(text: str) -> dict[str, str]:
     return facts
 
 
+def _merge_spoken_collab(text: str, facts: dict[str, str]) -> dict[str, str]:
+    """
+    私聊已经说到的协同项写进事实，进群才能预填。
+
+    不覆盖已有询价键。不拿去补问、不挡查价。
+    """
+    from agent.schema.field_validate import harvest_oral_collab, normalize_facts
+
+    merged = dict(facts or {})
+    extra = dict(harvest_oral_collab(text))
+    extra.update(normalize_facts(harvest_collab_labels(text)))
+    for key, val in extra.items():
+        if val and not str(merged.get(key) or "").strip():
+            merged[key] = val
+    return merged
+
+
 def extract_inquiry_snapshot(
     text: str,
     *,
@@ -80,11 +121,13 @@ def extract_inquiry_snapshot(
     标签行覆盖 B 的同名字段。B 未出现的字段保持空，不本地猜港口。
     件数/毛重/体积若原话已带 TMS 单位（件/KGS/CBM），空字段可机读补上。
     """
-    from agent.schema.field_validate import harvest_oral_measures, normalize_facts
+    from agent.schema.field_validate import harvest_oral_collab, harvest_oral_measures, normalize_facts
 
     if injected_facts is not None:
         mode = (injected_mode or detect_transport_mode(text) or "").upper()
         facts = harvest_oral_measures(text, normalize_facts(dict(injected_facts)))
+        facts = _merge_spoken_collab(text, facts)
+        facts = _merge_optional_inquiry_labels(text, facts)
         subtype = str(facts.get("运输类型") or "").strip()
         return {
             "business_line": mode,
@@ -109,6 +152,8 @@ def extract_inquiry_snapshot(
         mode = mode or str(b_payload.get("business_line") or "")
         source = "deepseek_b"
     facts = harvest_oral_measures(text, normalize_facts(facts))
+    facts = _merge_spoken_collab(text, facts)
+    facts = _merge_optional_inquiry_labels(text, 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
@@ -123,37 +168,117 @@ def extract_inquiry_snapshot(
     }
 
 
+def harvest_collab_labels(text: str) -> dict[str, str]:
+    """
+    群里协同标签：有无冒号都收。
+
+    「客户名称张三 包装方式编织袋 货值 10万」切成三项。
+    只说「纸箱」没有标签则不收。值取到下一个标签之前。
+    """
+    return _harvest_labels(text, _COLLAB_LABELS)
+
+
+def harvest_inquiry_optional_labels(text: str) -> dict[str, str]:
+    """
+    陆运询价非必填标签：通关口岸。
+
+    核对卡写「通关口岸（非必填）：」，销售可能照抄或只写「通关口岸皇岗」。
+    有无冒号都收。不猜口岸名。
+    """
+    return _harvest_labels(text, _OPTIONAL_INQUIRY_LABELS)
+
+
+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
+    return merged
+
+
+def _harvest_labels(text: str, labels: tuple[str, ...]) -> dict[str, str]:
+    """按标签切原话。同一起点只留最长标签。"""
+    raw = text or ""
+    hits: list[tuple[int, str]] = []
+    for label in labels:
+        start = 0
+        while True:
+            pos = raw.find(label, start)
+            if pos < 0:
+                break
+            hits.append((pos, label))
+            start = pos + len(label)
+    if not hits:
+        return {}
+    hits.sort()
+    # 同一起点只留最长标签，避免 HS 抢 HS编码、通关口岸抢「通关口岸（非必填）」。
+    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)
+        chunk = raw[begin:end]
+        chunk = re.sub(r"^[\s:：,，、]+", "", chunk)
+        chunk = re.sub(r"[\s,，、]+$", "", chunk)
+        if chunk:
+            out[label] = chunk
+    return out
+
+
 def extract_collab_fields(
     text: str,
     *,
     allow_b: bool = False,
     injected: Optional[dict[str, str]] = None,
+    allowed_keys: Optional[tuple[str, ...]] = None,
 ) -> dict[str, str]:
     """
-    群里协同字段：标签「键：值」+ 口语含油/含电/含磁、货好时间。
+    群里协同字段：标签（冒号可省）+ 口语含油/含电/含磁/危险品、货好时间。
 
-    闲聊、报价金额不进这里。商品海关编码 / HS 折成 HS编码。
-    allow_b 保留给调用方；协同六项不走询价七项的 B 合同。
+    allowed_keys 传入则只返回本单该有的项。不传则海运六项 ∪ 陆运协同键，给意图分类。
+    闲聊、只说「纸箱」不进这里。商品海关编码 / HS 折成 HS编码。
     """
     from agent.schema.field_validate import (
-        SEA_COLLAB_FIELDS,
+        ALL_COLLAB_KEYS,
         harvest_oral_collab,
         normalize_facts,
     )
 
     del allow_b
+    keys = tuple(allowed_keys) if allowed_keys is not None else ALL_COLLAB_KEYS
     if injected:
         return {
             k: str(v).strip()
             for k, v in normalize_facts(dict(injected)).items()
-            if k in SEA_COLLAB_FIELDS and str(v or "").strip()
+            if k in keys and str(v or "").strip()
         }
     labeled = normalize_facts(parse_labeled_facts(text))
+    labeled.update(normalize_facts(harvest_collab_labels(text)))
     oral = harvest_oral_collab(text)
+    if "是否为危险品" in oral:
+        if allowed_keys is not None and "是否为危险品" not in keys:
+            oral.pop("是否为危险品", None)
+        elif allowed_keys is None:
+            raw = text or ""
+            if "普货" in raw and not any(
+                w in raw for w in ("危险", "非危", "有危", "含危", "是否为危险品")
+            ):
+                oral.pop("是否为危险品", None)
     merged = dict(oral)
     merged.update({k: v for k, v in labeled.items() if v})
     out: dict[str, str] = {}
-    for key in SEA_COLLAB_FIELDS:
+    for key in keys:
         val = (merged.get(key) or "").strip()
         if val:
             out[key] = val

inquiry-agent/agent/llm/http_client.py

@@ -19,6 +19,42 @@ from agent.llm.modes import LlmMode
 logger = logging.getLogger(__name__)
 
 
+def message_plain_text(msg: dict[str, Any]) -> str:
+    """
+    把 assistant message 收成纯文本。
+
+    千问视觉的 content 经常是 [{text: ...}] 数组，不是字符串；
+    思考模式打开时正文可能只在 reasoning_content。两者都漏会让识图变「没看清」。
+    """
+    if not isinstance(msg, dict):
+        return ""
+    text = _content_to_text(msg.get("content"))
+    if text:
+        return text
+    reason = msg.get("reasoning_content")
+    if isinstance(reason, str) and reason.strip():
+        return reason.strip()
+    return ""
+
+
+def _content_to_text(content: Any) -> str:
+    """字符串或千问多模态数组 → 纯文本。"""
+    if isinstance(content, str):
+        return content.strip()
+    if isinstance(content, list):
+        parts: list[str] = []
+        for item in content:
+            if isinstance(item, str) and item.strip():
+                parts.append(item.strip())
+            elif isinstance(item, dict):
+                piece = item.get("text") if "text" in item else item.get("content")
+                got = _content_to_text(piece)
+                if got:
+                    parts.append(got)
+        return "\n".join(parts).strip()
+    return ""
+
+
 @dataclass
 class LlmHttpResult:
     ok: bool
@@ -72,7 +108,7 @@ class LlmHttpClient:
         mode_val = mode.value if isinstance(mode, LlmMode) else str(mode)
         if provider == "qwen":
             if mode_val == LlmMode.VISION.value:
-                return self.settings.qwen_vl_model or "qwen-vl-plus"
+                return self.settings.qwen_vl_model or "qwen3-vl-plus"
             return self.settings.qwen_text_model or "qwen-plus"
         return self.settings.deepseek_model or "deepseek-chat"
 
@@ -115,6 +151,9 @@ class LlmHttpClient:
             "messages": messages,
             "temperature": temperature,
         }
+        if provider == "qwen":
+            # 识图只要可见原文。思考开着时 content 常为空，销售侧就会「认不到」。
+            body["enable_thinking"] = False
         if tools:
             body["tools"] = tools
         if tool_choice is not None:
@@ -147,7 +186,7 @@ class LlmHttpClient:
             tool_calls: list[dict[str, Any]] = []
             if choices:
                 msg = (choices[0].get("message") or {})
-                content = str(msg.get("content") or "")
+                content = message_plain_text(msg if isinstance(msg, dict) else {})
                 raw_calls = msg.get("tool_calls") or []
                 if isinstance(raw_calls, list):
                     tool_calls = [c for c in raw_calls if isinstance(c, dict)]

inquiry-agent/agent/llm/mode_extract_fields.py

@@ -33,6 +33,7 @@ _FIELD_KEYS = (
     "箱型箱量",
     "车型/数量",
     "车型数量",
+    "通关口岸",
 )
 
 
@@ -127,7 +128,10 @@ def build_extract_messages(text: str) -> list[dict[str, str]]:
                 "10件/3pcs→空运件数；海运同样数字+件也要写入货量；100KGS/180kg/1.5KG/100公斤→毛重；2CBM/4.2cbm/2.5立方/2方→体积；"
                 "托盘/散货/卡板/纸箱/木箱→包装方式（销售说什么记什么，散货也要抽）。"
                 "9月15日/9月15号/2026-09-15→报价日期，规范成 yyyy-MM-dd，未写年用今年。"
+                "陆运「通关口岸：皇岗」「通关口岸皇岗」「过皇岗」有说就写入通关口岸，没说留空，禁止编造口岸。"
                 "「从南京飞吉隆坡」起运港=南京、目的港=吉隆坡；「北京飞吉隆坡」起运港=北京。"
+                "原话是机场三字码或航线就原样记，禁止翻译成城市名："
+                "SZX-CGK / SZX/CGK / SZX到CGK → 起运港=SZX、目的港=CGK，不要写成深圳、雅加达。"
                 "换城市必须覆盖旧值，不要沿用上一句的港口。"
                 "运输方式仅在原话能判断时填 AIR/SEA/LAND，否则 UNKNOWN。"
             ),

inquiry-agent/agent/llm/mode_vision.py

@@ -1,30 +1,166 @@
 """
-千问视觉：读图 mode 壳。
+千问视觉：只读图上可见原文。
+
+本文件职责：把图片交给 qwen3-vl-plus，抄出看得见的文字。
+属于 llm 包。依赖：LlmHttpClient（生产）；单测注入 chat_fn。
+禁止：填询价字段 JSON、算最终缺项、发明港口/重量/日期；禁止在回调线程长等（由 Worker 槽调用）。
 """
 
 from __future__ import annotations
 
-from typing import Any
+import logging
+from typing import Any, Optional
 
 from agent.llm.modes import LlmMode
 
+logger = logging.getLogger(__name__)
+
 MODE = LlmMode.VISION
 
+VISION_SYSTEM = (
+    "只读出图片上能看见的文字和数字，原样抄写，一行不漏。"
+    "英文、机场三字码、航线（如 SZX-CGK、SZX/CGK）必须保持原样，禁止翻译成中文城市名。"
+    "看不清的字不要猜。不要补询价缺项，不要发明港口、重量或日期。"
+    "只有整张图完全没有文字时才输出空；只要看得见字就必须抄出来。"
+)
+
+
+def sniff_image_mime(raw: bytes) -> str:
+    """
+    按文件头认图片类型。企微截图经常是 PNG，不能一律当 JPEG 送给千问。
+    """
+    blob = raw or b""
+    if len(blob) < 12:
+        return ""
+    if blob.startswith(b"\x89PNG\r\n\x1a\n"):
+        return "image/png"
+    if blob.startswith(b"\xff\xd8\xff"):
+        return "image/jpeg"
+    if blob[:4] == b"RIFF" and blob[8:12] == b"WEBP":
+        return "image/webp"
+    if blob.startswith(b"GIF87a") or blob.startswith(b"GIF89a"):
+        return "image/gif"
+    if blob[4:8] == b"ftyp":
+        return "image/heic"
+    if blob.startswith(b"BM"):
+        return "image/bmp"
+    return ""
+
 
 def mode_meta() -> dict[str, Any]:
     return {
         "mode": MODE.value,
         "provider": "qwen",
         "purpose": "vision",
-        "implemented": False,
+        "implemented": True,
+    }
+
+
+def _pack_text(text: str, *, stub: bool = False) -> dict[str, Any]:
+    body = (text or "").strip()
+    return {
+        "ok": True,
+        "stub": stub,
+        "mode": MODE.value,
+        "text": body,
+        "empty": not bool(body),
+        "error": "",
     }
 
 
 def invoke_shell(*, payload: dict[str, Any]) -> dict[str, Any]:
+    """
+    无网/单测入口：assistant_text 当作图上可见原文。
+
+    没有该键时保持失败码，避免假装已经读图。
+    """
+    if "assistant_text" in (payload or {}):
+        packed = _pack_text(str(payload.get("assistant_text") or ""), stub=True)
+        return packed
     return {
         "ok": False,
         "stub": True,
         "mode": MODE.value,
+        "text": "",
+        "empty": True,
         "error": "vision_shell_not_implemented",
-        "echo_keys": sorted(payload.keys()),
+        "echo_keys": sorted((payload or {}).keys()),
     }
+
+
+def _vision_messages(*, image_b64: str, mime: str, image_url: str) -> list[dict[str, Any]]:
+    """拼兼容 OpenAI 的图文消息。优先 data URL，其次 http 图链。"""
+    url = (image_url or "").strip()
+    if not url and image_b64:
+        kind = (mime or "image/jpeg").strip() or "image/jpeg"
+        url = f"data:{kind};base64,{image_b64}"
+    user_content: list[dict[str, Any]] = []
+    if url:
+        user_content.append({"type": "image_url", "image_url": {"url": url}})
+    user_content.append({"type": "text", "text": "请只抄写图上可见文字。"})
+    return [
+        {"role": "system", "content": VISION_SYSTEM},
+        {"role": "user", "content": user_content},
+    ]
+
+
+def invoke_vision(
+    *,
+    payload: dict[str, Any],
+    chat_fn: Optional[Any] = None,
+    allow_network: bool = False,
+) -> dict[str, Any]:
+    """
+    读一张图，返回可见原文。
+
+    chat_fn：单测注入，签名同 LlmHttpClient.chat。
+    allow_network=False 且无 chat_fn：走 invoke_shell，不打真网。
+    """
+    data = dict(payload or {})
+    image_b64 = str(data.get("image_b64") or "")
+    image_url = str(data.get("image_url") or "")
+    mime = str(data.get("mime") or "").strip() or "image/jpeg"
+    if chat_fn is None and not allow_network:
+        return invoke_shell(payload=data)
+    if not image_b64 and not image_url:
+        logger.info("vision 拒绝：没有图")
+        return {
+            "ok": False,
+            "stub": False,
+            "mode": MODE.value,
+            "text": "",
+            "empty": True,
+            "error": "no_image",
+        }
+    if chat_fn is None:
+        from agent.llm.http_client import LlmHttpClient
+
+        client = LlmHttpClient.from_settings()
+        chat_fn = client.chat
+    messages = _vision_messages(
+        image_b64=image_b64,
+        mime=mime,
+        image_url=image_url,
+    )
+    result = chat_fn(mode=MODE, messages=messages)
+    if not getattr(result, "ok", False):
+        err = str(getattr(result, "error", "") or "vision_failed")
+        logger.info("vision 失败 err=%s", err)
+        return {
+            "ok": False,
+            "stub": False,
+            "mode": MODE.value,
+            "text": "",
+            "empty": True,
+            "error": err,
+        }
+    packed = _pack_text(str(getattr(result, "content", "") or ""))
+    packed["stub"] = False
+    logger.info(
+        "vision 读完 ok=%s empty=%s text_len=%s mime=%s",
+        True,
+        packed.get("empty"),
+        len(str(packed.get("text") or "")),
+        mime,
+    )
+    return packed

inquiry-agent/agent/policy/air_group_ops.py

@@ -155,7 +155,25 @@ def send_air_brief_and_sheet(
     if room and not view.get("collab_chat_id"):
         view["collab_chat_id"] = room
     facts = dict(view.get("facts") or {})
+    from agent.schema.field_validate import AIR_COLLAB_FIELDS, pick_collab_from_facts
+
+    seeded = pick_collab_from_facts(facts, AIR_COLLAB_FIELDS)
     collab = dict(view.get("collab_facts") or {})
+    if seeded:
+        changed = False
+        for k, v in seeded.items():
+            if v and not str(collab.get(k) or "").strip():
+                collab[k] = v
+                changed = True
+        if changed:
+            view["collab_facts"] = collab
+            patcher = getattr(getattr(flow, "ledger", None), "patch_collab_facts", None)
+            if callable(patcher):
+                patcher(
+                    work_order_no=no,
+                    facts=seeded,
+                    sender_id=str(view.get("sales_wecom_id") or view.get("salesWecomId") or ""),
+                )
     view["collab_facts"] = collab
     view_q = dict(view.get("quote") or {})
     override = dict(quote_override or {})

inquiry-agent/agent/policy/air_text_flow.py

@@ -12,7 +12,7 @@ from __future__ import annotations
 import logging
 import threading
 import uuid
-from dataclasses import dataclass, field
+from dataclasses import asdict, dataclass, field
 from typing import Any, Callable, Optional
 
 from agent.ledger.http_ledger import HttpLedger
@@ -55,6 +55,43 @@ class FlowSession:
     pending_product_quote: dict[str, Any] = field(default_factory=dict)
 
 
+def _session_payload(sess: FlowSession) -> dict[str, Any]:
+    """书签转 JSON。tuple 改 list，方便 HTTP/Worker 共用。"""
+    data = asdict(sess)
+    data["allowed"] = list(sess.allowed or ())
+    data["invited_userids"] = list(sess.invited_userids or ())
+    data["invited_product_names"] = list(sess.invited_product_names or ())
+    data["facts"] = {str(k): str(v) for k, v in (sess.facts or {}).items()}
+    return data
+
+
+def _session_from_payload(raw: dict[str, Any]) -> FlowSession:
+    """Redis JSON → 书签。缺字段用默认，避免旧键打崩。"""
+    return FlowSession(
+        sender_id=str(raw.get("sender_id") or ""),
+        thread_id=str(raw.get("thread_id") or raw.get("sender_id") or ""),
+        wait_version=int(raw.get("wait_version") or 0),
+        phase=str(raw.get("phase") or "idle"),
+        business_line=str(raw.get("business_line") or ""),
+        facts={str(k): str(v) for k, v in dict(raw.get("facts") or {}).items()},
+        work_order_no=str(raw.get("work_order_no") or ""),
+        first_or_same=str(raw.get("first_or_same") or "first"),
+        history_work_order_no=str(raw.get("history_work_order_no") or ""),
+        quote=dict(raw.get("quote") or {}),
+        status=str(raw.get("status") or ""),
+        allowed=tuple(str(x) for x in (raw.get("allowed") or ())),
+        immutable_text=str(raw.get("immutable_text") or ""),
+        collab_chat_id=str(raw.get("collab_chat_id") or ""),
+        collab_facts={str(k): str(v) for k, v in dict(raw.get("collab_facts") or {}).items()},
+        invited_userids=tuple(str(x) for x in (raw.get("invited_userids") or ())),
+        invited_sales_name=str(raw.get("invited_sales_name") or ""),
+        invited_product_names=tuple(str(x) for x in (raw.get("invited_product_names") or ())),
+        deal_version=int(raw.get("deal_version") or 0),
+        collab_ended=bool(raw.get("collab_ended")),
+        pending_product_quote=dict(raw.get("pending_product_quote") or {}),
+    )
+
+
 class AirTextInquiryFlow:
     """
     空运文字询价 Owner。海运走 SeaTextInquiryFlow，陆运走 LandTextInquiryFlow。
@@ -63,12 +100,15 @@ class AirTextInquiryFlow:
     价格卡按工单号另存一份当时会话：点哪张卡就生成哪张卡当时的报价单。
     """
 
-    def __init__(self, ledger: Any = None, group_client: Any = None) -> None:
+    bookmark_kind = "air"
+
+    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:
@@ -83,8 +123,16 @@ class AirTextInquiryFlow:
         return self._group_client
 
     def session_of(self, sender_id: str) -> Optional[FlowSession]:
+        sid = (sender_id or "").strip()
         with self._lock:
-            return self._by_sender.get(sender_id)
+            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_by_work_order(self, work_order_no: str) -> Optional[FlowSession]:
         """用工单号找回书签，不校验 sender。群成交卡校验用。"""
@@ -126,17 +174,14 @@ class AirTextInquiryFlow:
         必须校验 sender_id，禁止用别人的工单号串会话。
         """
         no = copy.work_order_from_card_meta(card_meta, text=text)
+        if not no:
+            return self.session_of(sender_id)
         with self._lock:
-            if no:
-                hit = self._by_ticket.get(no)
-                if hit and hit.sender_id == sender_id:
-                    return hit
-            elif not no:
-                return self._by_sender.get(sender_id)
-        if no:
-            # 卡上有工单号就只动这一单；进程刚重启时从主账收回，禁止落到别人/最新单。
-            return self._load_ticket_session(sender_id, no)
-        return None
+            hit = self._by_ticket.get(no)
+            if hit and hit.sender_id == sender_id:
+                return hit
+        # 卡上有工单号就只动这一单；进程刚重启时从主账收回，禁止落到别人/最新单。
+        return self._load_ticket_session(sender_id, no)
 
     def _save(self, sess: FlowSession) -> None:
         with self._lock:
@@ -144,6 +189,53 @@ class AirTextInquiryFlow:
             no = (sess.work_order_no or "").strip()
             if no:
                 self._by_ticket[no] = sess
+        self._persist_bookmark(sess)
+
+    def drop_current_bookmark(self, sender_id: str) -> Optional[FlowSession]:
+        """
+        发图新开：丢掉当前书签，已出号会话仍留在 _by_ticket。
+
+        未出号的补问/核对草稿从此不再被文字当「当前单」。
+        HTTP 与 Worker 必须同时丢掉 Redis 里那份，否则补「托盘」会问运输方式。
+        """
+        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 _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,
+        )
 
     def _load_ticket_session(self, sender_id: str, work_order_no: str) -> Optional[FlowSession]:
         """

inquiry-agent/agent/policy/image_wave.py

@@ -0,0 +1,386 @@
+"""
+私聊图片波次账。
+
+本文件职责：记住「这一波」图和紧挨的字，供图片询价拼口播。
+属于 policy 包。生产波次内容落 Redis（HTTP 与 Worker 共用）；单测可纯内存。
+禁止：调模型、建单、写六态、在这里等几秒；禁止一把大锁串全部销售。
+
+10 秒只约束发图前那一句。发图之后只看业务结果出没出。
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import threading
+import uuid
+from dataclasses import asdict, dataclass, field
+from typing import Any, Optional, Protocol
+
+# 规格：发图前紧挨的一句字，间隔不超过 10 秒才带到新单。
+PRECEDE_SECONDS = 10.0
+# 波次短协调：超时后当新开，不当六态。
+_WAVE_TTL_SEC = 2 * 3600
+logger = logging.getLogger(__name__)
+
+
+class WaveKV(Protocol):
+    """HTTP 与 Worker 共用的短存。单测可注入内存假实现。"""
+
+    def get_json(self, kind: str, sender_id: str) -> Optional[dict[str, Any]]: ...
+
+    def set_json(self, kind: str, sender_id: str, data: dict[str, Any]) -> None: ...
+
+    def delete(self, kind: str, sender_id: str) -> None: ...
+
+
+@dataclass
+class LastPrivateText:
+    """同一发送人最近一句私聊文字。"""
+
+    sender_id: str
+    text: str
+    ts: float
+
+
+@dataclass
+class ImageWave:
+    """
+    一张新询价对应的图片波次。
+
+    pending：还在认的图张数；为 0 且未 business_done 时才能出业务结果。
+    """
+
+    sender_id: str
+    wave_id: str
+    ack_sent: bool = False
+    business_done: bool = False
+    media_ids: list[str] = field(default_factory=list)
+    vision_texts: list[str] = field(default_factory=list)
+    pending: int = 0
+    precede_text: str = ""
+    follow_text: str = ""
+    object_keys: list[str] = field(default_factory=list)
+    filenames: list[str] = field(default_factory=list)
+    errors: list[str] = field(default_factory=list)
+
+
+class ImageWaveBook:
+    """
+    图片波次账。按 sender 一把锁，不同销售互不等。
+
+    生产必须挂 Redis：收图在 HTTP 进程、识图在 Worker，两边不能各记一份内存。
+    单测不传 kv，仍走进程内字典。
+    """
+
+    def __init__(self, kv: Optional[WaveKV] = None) -> None:
+        self._lock = threading.Lock()
+        self._kv = kv
+        self._last_text: dict[str, LastPrivateText] = {}
+        self._waves: dict[str, ImageWave] = {}
+
+    def _hydrate(self, sender_id: str) -> None:
+        """从 Redis 覆盖本进程缓存。无 kv 时不做事。"""
+        sid = (sender_id or "").strip()
+        if not sid or self._kv is None:
+            return
+        try:
+            raw_wave = self._kv.get_json("wave", sid)
+            if raw_wave:
+                self._waves[sid] = _wave_from_dict(raw_wave)
+            else:
+                self._waves.pop(sid, None)
+            raw_text = self._kv.get_json("text", sid)
+            if raw_text and raw_text.get("text"):
+                self._last_text[sid] = LastPrivateText(
+                    sender_id=sid,
+                    text=str(raw_text.get("text") or ""),
+                    ts=float(raw_text.get("ts") or 0.0),
+                )
+            else:
+                self._last_text.pop(sid, None)
+        except Exception:
+            logger.exception("读图片波次失败 sender=%s", sid)
+
+    def _persist(self, sender_id: str) -> None:
+        """把当前 sender 的波次和前一句写回 Redis。"""
+        sid = (sender_id or "").strip()
+        if not sid or self._kv is None:
+            return
+        try:
+            wave = self._waves.get(sid)
+            if wave is None:
+                self._kv.delete("wave", sid)
+            else:
+                self._kv.set_json("wave", sid, asdict(wave))
+            row = self._last_text.get(sid)
+            if row is None:
+                self._kv.delete("text", sid)
+            else:
+                self._kv.set_json("text", sid, {"text": row.text, "ts": row.ts})
+        except Exception:
+            logger.exception("写图片波次失败 sender=%s", sid)
+
+    def note_text(self, sender_id: str, text: str, now: float) -> None:
+        """记下最近一句私聊文字，供 10 秒内发图带走。空句不记。"""
+        body = (text or "").strip()
+        sid = (sender_id or "").strip()
+        if not sid or not body:
+            return
+        with self._lock:
+            self._hydrate(sid)
+            self._last_text[sid] = LastPrivateText(sender_id=sid, text=body, ts=float(now))
+            self._persist(sid)
+
+    def forget_text(self, sender_id: str) -> None:
+        """
+        图片口播不是销售刚打的字，认完后清掉，避免 10 秒内再发图把旧口播带走。
+        """
+        sid = (sender_id or "").strip()
+        if not sid:
+            return
+        with self._lock:
+            self._hydrate(sid)
+            self._last_text.pop(sid, None)
+            self._persist(sid)
+
+    def precede_text(self, sender_id: str, now: float) -> str:
+        """发图前那一句：超过 10 秒返回空。"""
+        sid = (sender_id or "").strip()
+        with self._lock:
+            self._hydrate(sid)
+            row = self._last_text.get(sid)
+        if row is None:
+            return ""
+        if float(now) - row.ts > PRECEDE_SECONDS:
+            return ""
+        return row.text
+
+    def active(self, sender_id: str) -> Optional[ImageWave]:
+        """未出业务结果的当前波；没有则 None。"""
+        sid = (sender_id or "").strip()
+        with self._lock:
+            self._hydrate(sid)
+            wave = self._waves.get(sid)
+            if wave is None or wave.business_done:
+                return None
+            return wave
+
+    def open_or_join(
+        self, sender_id: str, media_id: str, *, now: float
+    ) -> tuple[ImageWave, str]:
+        """
+        第一张图开波；业务发出前的后续图并入；发出后再发图开新波。
+
+        返回 how：opened / joined / new_after_done。
+        副作用：pending +1。不发送企微。
+        """
+        sid = (sender_id or "").strip()
+        mid = (media_id or "").strip()
+        precede = self.precede_text(sid, now)
+        with self._lock:
+            self._hydrate(sid)
+            cur = self._waves.get(sid)
+            if cur is not None and not cur.business_done:
+                if mid:
+                    cur.media_ids.append(mid)
+                cur.pending += 1
+                self._persist(sid)
+                return cur, "joined"
+            how = "new_after_done" if cur is not None and cur.business_done else "opened"
+            wave = ImageWave(
+                sender_id=sid,
+                wave_id=uuid.uuid4().hex,
+                media_ids=[mid] if mid else [],
+                pending=1,
+                precede_text=precede,
+            )
+            self._waves[sid] = wave
+            self._persist(sid)
+            return wave, how
+
+    def mark_ack(self, sender_id: str) -> None:
+        """已经回过「正在识别图片」，连发第二张不得再 ack。"""
+        sid = (sender_id or "").strip()
+        with self._lock:
+            self._hydrate(sid)
+            wave = self._waves.get(sid)
+            if wave is not None:
+                wave.ack_sent = True
+                self._persist(sid)
+
+    def append_follow_text(self, sender_id: str, text: str) -> bool:
+        """业务结果发出前的字并进本波。发出后返回 False，交给文字询价。"""
+        body = (text or "").strip()
+        sid = (sender_id or "").strip()
+        if not sid or not body:
+            return False
+        with self._lock:
+            self._hydrate(sid)
+            wave = self._waves.get(sid)
+            if wave is None or wave.business_done:
+                return False
+            if wave.follow_text:
+                wave.follow_text = f"{wave.follow_text}\n{body}"
+            else:
+                wave.follow_text = body
+            self._persist(sid)
+            return True
+
+    def add_vision(
+        self,
+        sender_id: str,
+        text: str,
+        *,
+        object_key: str = "",
+        filename: str = "",
+        error: str = "",
+    ) -> ImageWave:
+        """
+        一张图认完。pending 减一，可见原文追加。
+
+        调用：Worker 识别槽。不得在本方法出发业务回复。
+        """
+        sid = (sender_id or "").strip()
+        with self._lock:
+            self._hydrate(sid)
+            wave = self._waves.get(sid)
+            if wave is None:
+                wave = ImageWave(sender_id=sid, wave_id=uuid.uuid4().hex, pending=1)
+                self._waves[sid] = wave
+            body = (text or "").strip()
+            if body:
+                wave.vision_texts.append(body)
+            if object_key:
+                wave.object_keys.append(object_key)
+            if filename:
+                wave.filenames.append(filename)
+            if error:
+                wave.errors.append(error)
+            if wave.pending > 0:
+                wave.pending -= 1
+            self._persist(sid)
+            return wave
+
+    def compose_oral(self, sender_id: str) -> str:
+        """precede + 各张可见原文 + 后到的字，换行拼接。"""
+        sid = (sender_id or "").strip()
+        with self._lock:
+            self._hydrate(sid)
+            wave = self._waves.get(sid)
+        if wave is None:
+            return ""
+        parts = []
+        if wave.precede_text.strip():
+            parts.append(wave.precede_text.strip())
+        for chunk in wave.vision_texts:
+            if chunk.strip():
+                parts.append(chunk.strip())
+        if wave.follow_text.strip():
+            parts.append(wave.follow_text.strip())
+        return "\n".join(parts)
+
+    def mark_business_done(self, sender_id: str) -> None:
+        """业务补问/卡片已发出。之后再发图必须新开。"""
+        sid = (sender_id or "").strip()
+        with self._lock:
+            self._hydrate(sid)
+            wave = self._waves.get(sid)
+            if wave is not None:
+                wave.business_done = True
+                wave.pending = 0
+                self._persist(sid)
+
+
+_BOOK: Optional[ImageWaveBook] = None
+_BOOK_GUARD = threading.Lock()
+
+
+def _wave_from_dict(raw: dict[str, Any]) -> ImageWave:
+    """Redis JSON → 波次对象。缺字段用默认，避免旧键打崩。"""
+    return ImageWave(
+        sender_id=str(raw.get("sender_id") or ""),
+        wave_id=str(raw.get("wave_id") or uuid.uuid4().hex),
+        ack_sent=bool(raw.get("ack_sent")),
+        business_done=bool(raw.get("business_done")),
+        media_ids=[str(x) for x in (raw.get("media_ids") or [])],
+        vision_texts=[str(x) for x in (raw.get("vision_texts") or [])],
+        pending=int(raw.get("pending") or 0),
+        precede_text=str(raw.get("precede_text") or ""),
+        follow_text=str(raw.get("follow_text") or ""),
+        object_keys=[str(x) for x in (raw.get("object_keys") or [])],
+        filenames=[str(x) for x in (raw.get("filenames") or [])],
+        errors=[str(x) for x in (raw.get("errors") or [])],
+    )
+
+
+class RedisWaveKV:
+    """
+    把图片波次落到本前缀 Redis，TTL 2 小时。
+
+    HTTP 收图、Worker 认图必须读同一把键，否则第二张图会丢。
+    """
+
+    def __init__(self, client: Any) -> None:
+        self._client = client
+
+    def get_json(self, kind: str, sender_id: str) -> Optional[dict[str, Any]]:
+        from agent.redis_coord.keys import IMAGE_WAVE, IMAGE_WAVE_TEXT
+
+        suffix = IMAGE_WAVE if kind == "wave" else IMAGE_WAVE_TEXT
+        raw = self._client.raw.get(self._client.key(suffix, sender_id))
+        if not raw:
+            return None
+        if isinstance(raw, bytes):
+            raw = raw.decode("utf-8")
+        data = json.loads(raw)
+        return dict(data) if isinstance(data, dict) else None
+
+    def set_json(self, kind: str, sender_id: str, data: dict[str, Any]) -> None:
+        from agent.redis_coord.keys import IMAGE_WAVE, IMAGE_WAVE_TEXT
+
+        suffix = IMAGE_WAVE if kind == "wave" else IMAGE_WAVE_TEXT
+        self._client.raw.set(
+            self._client.key(suffix, sender_id),
+            json.dumps(data, ensure_ascii=False),
+            ex=_WAVE_TTL_SEC,
+        )
+
+    def delete(self, kind: str, sender_id: str) -> None:
+        from agent.redis_coord.keys import IMAGE_WAVE, IMAGE_WAVE_TEXT
+
+        suffix = IMAGE_WAVE if kind == "wave" else IMAGE_WAVE_TEXT
+        self._client.raw.delete(self._client.key(suffix, sender_id))
+
+
+def _production_kv() -> Optional[WaveKV]:
+    """生产挂 Redis；连不上则退回内存并打日志。"""
+    try:
+        from agent.redis_coord.runtime import get_redis_runtime
+
+        rt = get_redis_runtime()
+        if rt is None or rt.client is None:
+            return None
+        return RedisWaveKV(rt.client)
+    except Exception:
+        logger.exception("图片波次 Redis 不可用，本进程退回内存账（双进程会丢第二张图）")
+        return None
+
+
+def get_image_wave_book() -> ImageWaveBook:
+    """
+    进程内一份句柄；生产波次内容在 Redis，HTTP 与 Worker 才能对上。
+    """
+    global _BOOK
+    with _BOOK_GUARD:
+        if _BOOK is None:
+            _BOOK = ImageWaveBook(kv=_production_kv())
+        return _BOOK
+
+
+def reset_image_wave_book_for_test(kv: Optional[WaveKV] = None) -> ImageWaveBook:
+    """单测重置。传入同一 kv 可模拟 HTTP/Worker 两份句柄。"""
+    global _BOOK
+    with _BOOK_GUARD:
+        _BOOK = ImageWaveBook(kv=kv)
+        return _BOOK

inquiry-agent/agent/policy/inquiry_copy.py

@@ -17,9 +17,10 @@ from agent.schema.field_validate import (
     AIR_COLLAB_FIELDS,
     AIR_REQUIRED_FIELDS,
     LAND_DISPLAY,
-    SEA_COLLAB_DISPLAY,
     SEA_COLLAB_FIELDS,
     SEA_REQUIRED_FIELDS,
+    collab_display_name,
+    collab_fields_for_ticket,
     display_name,
 )
 from agent.schema.land_options import option_prompt as _land_option_prompt
@@ -99,8 +100,50 @@ _CONTINUE_HINT = (
 
 
 ASK_TRANSPORT = "请补充运输方式：海运 / 空运 / 陆运。"
+# 私聊图片询价：先回识别中；看不清/失败与补问拼成一条，销售侧不加 [test]。
+IMAGE_RECOGNIZING = "正在识别图片"
+IMAGE_UNREADABLE = "图片没看清询价信息，请换一张更清楚的，或直接打字。"
+IMAGE_TECH_FAIL = "图片暂时没认出来，请再发一张或直接打字。"
+
+
+def image_fail_then_clarify(reason: str, clarify: str) -> str:
+    """
+    看不清或技术失败时，说明 + 文字询价补问写在同一条回复。
+
+    禁止只回说明就把人丢在那里。两端空白去掉，中间空一行。
+    """
+    head = (reason or "").strip()
+    body = (clarify or "").strip()
+    if head and body:
+        return f"{head}\n\n{body}"
+    return head or body
+
 # 旧一句只问运输类型；销售要一次看到类型、线路、分类。残留调用也走完整清单。
 ASK_LAND_SUBTYPE = _land_option_prompt(current={})
+
+
+def land_sample_image_payload() -> dict[str, Any]:
+    """
+    填写样例表图片出站。
+
+    走企微 image，不走文件气泡。路径随发布包，出站线程再读字节。
+    禁止在回调线程上传。
+    """
+    from agent.schema.land_options import sample_image_path
+
+    path = sample_image_path()
+    return {
+        "msgtype": "image",
+        "filename": "填写样例.png",
+        "filepath": str(path),
+    }
+
+
+def land_sample_image_fallback() -> str:
+    """图片失败时回文字样例，避免销售什么都看不到。"""
+    from agent.schema.land_options import sample_fallback_text
+
+    return sample_fallback_text()
 FIRST_CHAT = "你好，我是询价机器人。请直接描述客户需求，例如「上海到洛杉矶空运，3 件 180kg，下周三前到」，我会解析需求并查询 TMS 标准报价。"
 ROUTE_MODEL_FAILED = "刚才没听清，请再说一次：是新询价，还是要接着办哪张工单？"
 LAND_CONFIRM_TAIL = "请回复“确定”进行TMS查价，内容有误时，请重新发内容。"
@@ -111,9 +154,8 @@ LAND_TMS_RETRYING = "正在重新查询TMS报价。工单：{wo}"
 _LAND_LCL_KEYS = ("起运港", "目的港", "品名", "体积", "毛重", "报价日期")
 _LAND_FTL_KEYS = ("起运港", "目的港", "品名", "车型/数量", "报价日期")
 _LAND_HK_FULL_KEYS = ("起运港", "目的港", "品名", "报价日期")
-_LAND_LCL_TYPES = frozenset({"国内运输拼车", "跨境集拼"})
-_LAND_FTL_TYPES = frozenset({"国内运输整车", "跨境整车"})
-_LAND_CUSTOMS_TYPES = frozenset({"跨境集拼", "跨境整车", "中港"})
+_LAND_FTL_TYPES = frozenset({"国内运输整车", "跨境整车", "中港零担/集拼"})
+_LAND_CUSTOMS_TYPES = frozenset({"跨境集拼", "跨境整车", "中港整车", "中港零担/集拼"})
 _LAND_EXAMPLES = {
     "起运港": "广州",
     "目的港": "深圳",
@@ -127,11 +169,8 @@ _LAND_EXAMPLES = {
 
 def _land_cargo_keys(land_type: str, facts: dict[str, str] | None = None) -> tuple[str, ...]:
     """当前运输类型要展示的货物内部键。中港整车不要车型/数量。"""
-    if land_type == "中港":
-        load = str((facts or {}).get("中港装载方式") or "").strip()
-        if load == "零担/集拼":
-            return _LAND_FTL_KEYS
-        # 整车或还没选定装载：卡片/补问都不出现车型/数量
+    _ = facts
+    if land_type == "中港整车":
         return _LAND_HK_FULL_KEYS
     if land_type in _LAND_FTL_TYPES:
         return _LAND_FTL_KEYS
@@ -142,8 +181,9 @@ def land_confirm_card(facts: dict[str, str] | None) -> str:
     """
     确定查价前的核对卡。没有工单号。
 
-    行列出运输类型/线路/分类（中港再加装载）和该类型货物字段。
+    行列出运输类型/线路/分类和该类型货物字段。
     国内不出现通关口岸；跨境/中港通关口岸非必填，没填留空行。
+    不再单独列出中港装载方式。
     """
     src = dict(facts or {})
     land_type = (src.get("运输类型") or "").strip()
@@ -152,8 +192,6 @@ def land_confirm_card(facts: dict[str, str] | None) -> str:
         f"线路类别（必填）：{(src.get('线路类别') or '').strip()}",
         f"运输分类（必填）：{(src.get('运输分类') or '').strip()}",
     ]
-    if land_type == "中港":
-        lines.append(f"中港装载方式（必填）：{(src.get('中港装载方式') or '').strip()}")
     for key in _land_cargo_keys(land_type, src):
         label = LAND_DISPLAY.get(key, key)
         val = (src.get(key) or "").strip()
@@ -170,6 +208,29 @@ def land_group_no_staff() -> str:
     return "没有可拉的产品人员。请先在后台「员工管理」维护本线路的陆运产品岗后再试。"
 
 
+def land_group_pair_rejected() -> str:
+    """类型+线路不在填写样例里，禁止拉群。"""
+    return "拉群失败，请排查填写内容，并重新发起工单。"
+
+
+def _land_optional_keys(land_type: str) -> tuple[str, ...]:
+    """当前类型询价非必填内部键。国内没有通关口岸。"""
+    if land_type in _LAND_CUSTOMS_TYPES:
+        return ("通关口岸",)
+    return ()
+
+
+def _land_field_line(key: str, value: str, *, required: bool, example: str = "") -> str:
+    """销售侧一行：名称（必填/非必填）：值或示例。"""
+    label = LAND_DISPLAY.get(key, key)
+    mark = "必填" if required else "非必填"
+    if value:
+        return f"{label}（{mark}）：{value}"
+    if example:
+        return f"{label}（{mark}）：(如：{example}）"
+    return f"{label}（{mark}）："
+
+
 def ask_land_fields(
     *,
     facts: dict[str, str] | None = None,
@@ -179,30 +240,37 @@ 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 _land_cargo_keys(land_type, src):
         val = (src.get(key) or "").strip()
         if val:
-            lines.append(f"{LAND_DISPLAY.get(key, key)}：{val}")
+            lines.append(_land_field_line(key, val, required=True))
+    for key in optional:
+        val = (src.get(key) or "").strip()
+        if val:
+            lines.append(_land_field_line(key, val, required=False))
+    pending = list(missing)
+    for key in optional:
+        if not (src.get(key) or "").strip() and key not in pending:
+            pending.append(key)
+    if not pending:
+        return ""
     lines.append("")
     lines.append("待补充：")
-    if not missing:
-        return ""
-    for key in missing:
-        label = display_name(key, "LAND")
+    for key in pending:
+        required = key not in optional
         example = _LAND_EXAMPLES.get(key, "")
-        if example:
-            lines.append(f"{label}：(如：{example}）")
-        else:
-            lines.append(f"{label}：")
+        lines.append(_land_field_line(key, "", required=required, example=example))
     return "\n".join(lines)
 
 
@@ -342,7 +410,8 @@ def inquiry_field_lines(
     """
     询价确认字段行（销售可见）。
 
-    起运/目的展示销售原文（南京/北京），三字码只给出站 TMS，不对销售展示。
+    起运/目的展示销售原文：说南京就南京，图上/原话是 SZX 就展示 SZX。
+    originCode 只给出站 TMS，不拿来改写成城市名再给销售看。
     """
     origin = (facts.get("起运港") or facts.get("起运地") or "-").strip()
     dest = (facts.get("目的港") or facts.get("目的地") or "-").strip()
@@ -373,10 +442,6 @@ def inquiry_field_lines(
             f"线路类别：{(facts.get('线路类别') or '-').strip() or '-'}",
             f"运输分类：{(facts.get('运输分类') or '-').strip() or '-'}",
         ]
-        if land_type == "中港":
-            lines.append(
-                f"中港装载方式：{(facts.get('中港装载方式') or '-').strip() or '-'}"
-            )
         for key in _land_cargo_keys(land_type, facts):
             label = LAND_DISPLAY.get(key, key)
             val = (facts.get(key) or "-").strip() or "-"
@@ -1151,8 +1216,10 @@ def sea_group_brief(
     lines.append("")
     lines.append("协同补充字段：")
     collab = dict(collab_facts or {})
-    for key in SEA_COLLAB_FIELDS:
-        label = SEA_COLLAB_DISPLAY.get(key, key)
+    line = "LAND" if mode == "陆运" else "SEA"
+    collab_keys = collab_fields_for_ticket(facts, line) or SEA_COLLAB_FIELDS
+    for key in collab_keys:
+        label = collab_display_name(key, line)
         val = (collab.get(key) or collab.get(label) or "").strip() or "-"
         lines.append(f"{label}：{val}")
     tms = dict(tms_quote) if tms_quote is not None else {}
@@ -1298,14 +1365,10 @@ def scheme_horizontal_rows(
 
 
 def _land_scheme_type_line(facts: dict[str, str]) -> str:
-    """方案卡「类型」格：运输类型、中港装载、线路类别、运输分类。"""
+    """方案卡「类型」格：运输类型、线路类别、运输分类。"""
     src = facts or {}
     land_type = (src.get("运输类型") or "-").strip() or "-"
     bits = [land_type]
-    if land_type == "中港":
-        load = (src.get("中港装载方式") or "").strip()
-        if load:
-            bits.append(load)
     route = (src.get("线路类别") or "-").strip() or "-"
     klass = (src.get("运输分类") or "-").strip() or "-"
     bits.extend([route, klass])
@@ -1576,7 +1639,15 @@ def group_quote_not_recognized() -> str:
     return "没有识别到报价，请产品再发一次价格或 Excel/PDF"
 
 
-def group_fields_not_recognized() -> str:
+def group_fields_not_recognized(
+    keys: tuple[str, ...] | None = None,
+    business_line: str = "",
+) -> str:
+    """没认出协同字段。陆运列出本单可补项；海运保持原六项口。"""
+    if keys:
+        line = (business_line or "").strip().upper() or "SEA"
+        names = [collab_display_name(k, line) for k in keys]
+        return "听到了。请直接说" + "、".join(names) + "。"
     return "听到了。请直接说贸易条款、货好时间、商品海关编码、是否含油/含电/含磁。"
 
 

inquiry-agent/agent/policy/land_text_flow.py

@@ -1,7 +1,7 @@
 """
 陆运文字询价确定性流程。
 
-本文件职责：选项联动 → 按类型补货物 → 无工单号核对卡 →（确定后）建单查 TMS。
+本文件职责：三项选项各自必填（不联动）→ 按类型补货物 → 无工单号核对卡 →（确定后）建单查 TMS。
 查价后半段复用海运拉群/报价单/成交，拉人按线路类别，禁止陆运岗全员。
 禁止：字段齐了就建单；直连 TMS；把海运「出卡即查价」改掉。
 调用：handler 在 Worker/inbox 线程；禁止回调线程同步跑。
@@ -23,6 +23,8 @@ from agent.schema.land_options import (
     harvest_land_options,
     looks_like_confirm,
     option_prompt,
+    option_prompt_needs_sample,
+    sample_pair_allowed,
     validate_land_options,
 )
 from agent.schema.tms_land_query import assemble_land_query
@@ -39,6 +41,8 @@ class LandTextInquiryFlow(SeaTextInquiryFlow):
     前半段与海运不同：核对通过前不建单。后半段套海运建群，匹配改走线路类别。
     """
 
+    bookmark_kind = "land"
+
     def on_text(
         self,
         *,
@@ -112,7 +116,7 @@ class LandTextInquiryFlow(SeaTextInquiryFlow):
         subtype = str(snap.get("land_subtype") or facts.get("运输类型") or "").strip()
         if subtype and not str(facts.get("运输类型") or "").strip():
             facts["运输类型"] = subtype
-        option_keys = ("运输类型", "线路类别", "运输分类", "中港装载方式")
+        option_keys = ("运输类型", "线路类别", "运输分类")
         base = dict(sess.facts) if sess else {}
         cargo = {k: v for k, v in facts.items() if k not in option_keys}
         merged = dict(base)
@@ -184,6 +188,23 @@ class LandTextInquiryFlow(SeaTextInquiryFlow):
             sess=sess,
         )
 
+    def _reply_land_options(
+        self,
+        reply: ReplyFn,
+        facts: dict[str, str],
+        *,
+        message: str = "",
+    ) -> None:
+        """
+        选项补问：先发文字清单，需要样例时再发表格图。
+
+        图片失败由出站回退文字样例。不在本方法上传素材。
+        """
+        body = (message or "").strip() or option_prompt(current=facts)
+        reply(body)
+        if option_prompt_needs_sample(current=facts):
+            reply(copy.land_sample_image_fallback(), copy.land_sample_image_payload())
+
     def _confirm_or_hold(
         self,
         *,
@@ -221,7 +242,7 @@ class LandTextInquiryFlow(SeaTextInquiryFlow):
         """
         销售确定之后才建陆运工单并查 TMS。
 
-        核对卡阶段没有工单号。运输分类、中港装载方式写入工单 facts，不进入 assemble 出站。
+        核对卡阶段没有工单号。运输分类写入工单 facts，不进入 assemble 出站。
         """
         import uuid
 
@@ -363,10 +384,10 @@ class LandTextInquiryFlow(SeaTextInquiryFlow):
             facts=merged, business_line="LAND", land_subtype=land_type
         )
         cargo = dict(check["facts"])
-        cargo.update({k: merged[k] for k in ("运输类型", "线路类别", "运输分类", "中港装载方式") if merged.get(k)})
+        cargo.update({k: merged[k] for k in ("运输类型", "线路类别", "运输分类") if merged.get(k)})
         tms = assemble_land_query(cargo)
         cargo = dict(tms.get("facts") or cargo)
-        cargo.update({k: merged[k] for k in ("运输类型", "线路类别", "运输分类", "中港装载方式") if merged.get(k)})
+        cargo.update({k: merged[k] for k in ("运输类型", "线路类别", "运输分类") if merged.get(k)})
         missing = list(check.get("missing") or [])
         for key in tms.get("missing") or []:
             if key not in missing and key not in {"运输类型", "线路类别"}:
@@ -412,7 +433,11 @@ class LandTextInquiryFlow(SeaTextInquiryFlow):
             return "wait_confirm"
         if state.get("stage") == "options_error":
             err = (state["opt"].get("errors") or [{}])[0]
-            reply(str(err.get("message") or option_prompt(current=facts)))
+            self._reply_land_options(
+                reply,
+                facts,
+                message=str(err.get("message") or ""),
+            )
             self._save(
                 FlowSession(
                     sender_id=sender_id,
@@ -427,12 +452,11 @@ class LandTextInquiryFlow(SeaTextInquiryFlow):
             self._clear_parked_confirm(sender_id)
             return "need_land_options"
         if state["stage"] == "options_missing":
-            body = option_prompt(current=facts)
             logger.info(
                 "land.options_missing missing=%s",
                 (state.get("opt") or {}).get("missing") or [],
             )
-            reply(body)
+            self._reply_land_options(reply, facts)
             self._save(
                 FlowSession(
                     sender_id=sender_id,
@@ -574,6 +598,32 @@ class LandTextInquiryFlow(SeaTextInquiryFlow):
 
         clear_pending_land_confirm(sender_id)
 
+    def _pull_collab_group(
+        self,
+        sess: FlowSession,
+        reply: ReplyFn,
+        card_meta: Optional[dict[str, Any]] = None,
+    ) -> str:
+        """
+        点「拉产品进群协同」：先核填写样例，再按线路拉人建群。
+
+        类型+线路对不上样例：不建群，告诉销售重新开单。按钮保持可点。
+        对得上则走海运同一套建群。
+        """
+        facts = dict(sess.facts or {})
+        land_type = str(facts.get("运输类型") or "").strip()
+        route = str(facts.get("线路类别") or "").strip()
+        if not sample_pair_allowed(land_type, route):
+            logger.info(
+                "land.group.reject pair type=%s route=%s wo=%s",
+                land_type or "-",
+                route or "-",
+                sess.work_order_no or "-",
+            )
+            reply(copy.land_group_pair_rejected())
+            return sess.phase or "wait_collab"
+        return super()._pull_collab_group(sess, reply, card_meta=card_meta)
+
     def _match_sea_staff(self, sess: FlowSession) -> list[dict[str, Any]]:
         """按本单线路类别整词匹配产品·陆运，禁止全员。"""
         matcher = getattr(self._ledger, "match_land_staff", None)

inquiry-agent/agent/policy/quote_overlay.py

@@ -24,6 +24,10 @@ _COLLAB_LABEL_MARKERS: tuple[tuple[str, tuple[str, ...]], ...] = (
     ("是否含油", ("是否含油", "has_oil")),
     ("是否含电", ("是否含电", "has_battery")),
     ("是否含磁", ("是否含磁", "has_magnet")),
+    ("客户名称", ("客户名称", "customer_name")),
+    ("货值", ("货值", "cargo_value")),
+    ("是否为危险品", ("是否为危险品", "is_dangerous")),
+    ("包装方式", ("包装方式", "包装类型")),
 )
 
 

inquiry-agent/agent/policy/sea_group_ops.py

@@ -14,7 +14,12 @@ from typing import Any, Optional
 from agent.llm.extract_text import extract_collab_fields, extract_product_quote
 from agent.policy.quote_overlay import overlay_labeled_facts
 from agent.policy import inquiry_copy as copy
-from agent.schema.field_validate import SEA_COLLAB_DISPLAY, SEA_COLLAB_FIELDS
+from agent.schema.field_validate import (
+    SEA_COLLAB_DISPLAY,
+    SEA_COLLAB_FIELDS,
+    collab_display_name,
+    collab_fields_for_ticket,
+)
 
 # 主账填表按英文码取格子；中文键必须同时带上，否则货好时间/条款写不进表。
 _COLLAB_FILL_ALIASES = {
@@ -24,6 +29,10 @@ _COLLAB_FILL_ALIASES = {
     "是否含油": ("是否含油", "has_oil"),
     "是否含电": ("是否含电", "has_battery"),
     "是否含磁": ("是否含磁", "has_magnet"),
+    "客户名称": ("客户名称", "customer_name"),
+    "货值": ("货值", "cargo_value"),
+    "是否为危险品": ("是否为危险品", "is_dangerous"),
+    "包装方式": ("包装方式", "包装类型"),
 }
 
 logger = logging.getLogger(__name__)
@@ -91,6 +100,24 @@ def ticket_no(ticket: object) -> str:
     return str(ticket_view(ticket).get("work_order_no") or "").strip()
 
 
+def _collab_keys_of(ticket: object, sess: Any = None) -> tuple[str, tuple[str, ...]]:
+    """本单业务线 + 允许写入的协同内部键。"""
+    view = ticket_view(ticket)
+    facts = dict(view.get("facts") or {})
+    if sess is not None and getattr(sess, "facts", None):
+        facts.update(dict(sess.facts or {}))
+    line = str(
+        view.get("business_line")
+        or view.get("businessLine")
+        or (getattr(sess, "business_line", "") if sess is not None else "")
+        or "SEA"
+    ).strip().upper() or "SEA"
+    keys = collab_fields_for_ticket(facts, line)
+    if not keys and line != "LAND":
+        keys = SEA_COLLAB_FIELDS
+    return line, keys
+
+
 def is_terminal(ticket: object) -> bool:
     return ticket_status(ticket) in TERMINAL
 
@@ -378,6 +405,13 @@ def send_brief_and_sheet(
     quote = dict(override or sess_q or view_q or tms or manual)
     sales_name, products, mention_ids = resolve_group_mentions(flow, ticket, sess)
     has_tms = bool(tms.get("total") or _quote_fee_rows(tms))
+    line = str(
+        view.get("business_line")
+        or view.get("businessLine")
+        or (sess.business_line if sess is not None else "")
+        or "SEA"
+    ).strip().upper()
+    mode = "陆运" if line == "LAND" else "海运"
     brief = copy.sea_group_brief(
         work_order_no=no,
         facts=facts,
@@ -388,9 +422,16 @@ def send_brief_and_sheet(
         has_price=has_tms,
         tms_quote=tms,
         product_quote=manual,
+        transport_mode=mode,
     )
     send_group_text(client, chat_id, brief, mention_userids=mention_ids)
-    send_filled_sheet(flow=flow, ticket=ticket, sess=sess, quote_override=quote)
+    send_filled_sheet(
+        flow=flow,
+        ticket=ticket,
+        sess=sess,
+        quote_override=quote,
+        business_line=line,
+    )
     return "brief_and_sheet"
 
 
@@ -400,7 +441,7 @@ def send_filled_sheet(
     ticket: object,
     sess: Any = None,
     quote_override: Optional[dict[str, Any]] = None,
-    business_line: str = "SEA",
+    business_line: str = "",
     chat_id: str = "",
 ) -> dict[str, Any]:
     """
@@ -429,7 +470,7 @@ def send_filled_sheet(
         flow=flow,
         chat_id=room,
         work_order_no=no,
-        facts=_facts_for_fill(facts, collab),
+        facts=_facts_for_fill(facts, collab, business_line=line or "SEA"),
         quote=quote,
         sales_id=str(
             (sess.sender_id if sess else "")
@@ -441,15 +482,20 @@ def send_filled_sheet(
     )
 
 
-def _collab_value(collab: dict[str, str], key: str) -> str:
-    """协同六项：内部键、对外展示名都能取到。"""
+def _collab_value(collab: dict[str, str], key: str, business_line: str = "SEA") -> str:
+    """协同项：内部键、对外展示名都能取到。"""
     src = dict(collab or {})
     val = str(src.get(key) or "").strip()
     if val:
         return val
-    label = SEA_COLLAB_DISPLAY.get(key, "")
+    label = collab_display_name(key, business_line)
     if label:
         val = str(src.get(label) or "").strip()
+    if val:
+        return val
+    sea_label = SEA_COLLAB_DISPLAY.get(key, "")
+    if sea_label:
+        val = str(src.get(sea_label) or "").strip()
     return val
 
 
@@ -466,15 +512,22 @@ def _oil_elec_mag_text(collab: dict[str, str]) -> str:
     return f"油：{oil or '-'}；电：{elec or '-'}；磁：{mag or '-'}"
 
 
-def _facts_for_fill(facts: dict[str, str], collab: dict[str, str]) -> dict[str, str]:
+def _facts_for_fill(
+    facts: dict[str, str],
+    collab: dict[str, str],
+    business_line: str = "SEA",
+) -> dict[str, str]:
     """
-    询价七项 + 协同六项。中文键和填表英文码一起带上。
+    询价字段 + 本单协同项。中文键和填表英文码一起带上。
     主账 writeFactsByLabel 只认 cargo_ready_date / trade_terms，只传「货好时间」会空着。
+    模板没有格子的新项照样带上，overlay 找不到标签就跳过。
     """
     merged = dict(facts or {})
     src = dict(collab or {})
-    for key in SEA_COLLAB_FIELDS:
-        val = _collab_value(src, key)
+    line = (business_line or "SEA").strip().upper() or "SEA"
+    keys = collab_fields_for_ticket(facts, line) or SEA_COLLAB_FIELDS
+    for key in keys:
+        val = _collab_value(src, key, line)
         if not val:
             continue
         for alias in _COLLAB_FILL_ALIASES.get(key, (key,)):
@@ -549,7 +602,7 @@ def apply_collab_update(
     sender_id: str,
     sess: Any = None,
 ) -> str:
-    """补协同六项。终态要说明原因。"""
+    """补本单协同项。终态要说明原因。不该有的项当没补到。"""
     ticket = hydrate_ticket(flow, ticket)
     no = ticket_no(ticket)
     chat_id = str((sess.collab_chat_id if sess else "") or ticket_view(ticket).get("collab_chat_id") or "")
@@ -557,11 +610,21 @@ def apply_collab_update(
     if is_terminal(ticket):
         send_group_text(client, chat_id, copy.ticket_locked(work_order_no=no, status=ticket_status(ticket)))
         return "ticket_locked"
-    fields = extract_collab_fields(text, allow_b=True)
+    line, keys = _collab_keys_of(ticket, sess)
+    fields = extract_collab_fields(text, allow_b=True, allowed_keys=keys)
     if not fields:
-        send_group_text(client, chat_id, copy.group_fields_not_recognized())
+        send_group_text(
+            client,
+            chat_id,
+            copy.group_fields_not_recognized(keys=keys or None, business_line=line),
+        )
         return "collab_not_recognized"
-    flow.apply_collab_facts(work_order_no=no, facts=fields, sender_id=sender_id)
+    flow.apply_collab_facts(
+        work_order_no=no,
+        facts=fields,
+        sender_id=sender_id,
+        allowed_keys=keys,
+    )
     if sess is not None:
         merged = dict(sess.collab_facts)
         merged.update(fields)
@@ -603,9 +666,15 @@ def apply_product_quote(
     if not is_product(ticket, sender_id, flow=flow):
         send_group_text(client, chat_id, copy.group_only_product_can_quote())
         return "quote_not_product"
-    fields = extract_collab_fields(text)
+    fields = extract_collab_fields(text, allowed_keys=_collab_keys_of(ticket, sess)[1])
     if fields:
-        flow.apply_collab_facts(work_order_no=no, facts=fields, sender_id=sender_id)
+        line, keys = _collab_keys_of(ticket, sess)
+        flow.apply_collab_facts(
+            work_order_no=no,
+            facts=fields,
+            sender_id=sender_id,
+            allowed_keys=keys,
+        )
         if sess is not None:
             merged = dict(sess.collab_facts)
             merged.update(fields)
@@ -772,7 +841,7 @@ def _persist_quote_sheet(
         result = renderer(
             work_order_no=no,
             business_line=line,
-            facts=_facts_for_fill(facts, collab),
+            facts=_facts_for_fill(facts, collab, business_line=line or "SEA"),
             quote=quote,
             format="xlsx",
             sales_wecom_id=str(

inquiry-agent/agent/policy/sea_text_flow.py

@@ -40,6 +40,8 @@ class SeaTextInquiryFlow(AirTextInquiryFlow):
     复用空运后半段（跳过协同 / 出方案卡 / Excel PDF / 成交），只改查价前与卡片、拉群。
     """
 
+    bookmark_kind = "sea"
+
     def __init__(self, ledger: Any = None, group_client: Any = None) -> None:
         super().__init__(ledger=ledger)
         self._group_client = group_client
@@ -514,6 +516,24 @@ class SeaTextInquiryFlow(AirTextInquiryFlow):
                 product_ids=product_ids,
             )
 
+        from agent.schema.field_validate import collab_fields_for_ticket, pick_collab_from_facts
+
+        line = (sess.business_line or "SEA").upper()
+        keys = collab_fields_for_ticket(sess.facts, line)
+        seeded = pick_collab_from_facts(sess.facts, keys)
+        if seeded:
+            merged = dict(sess.collab_facts or {})
+            for k, v in seeded.items():
+                if v and not str(merged.get(k) or "").strip():
+                    merged[k] = v
+            sess.collab_facts = merged
+            self.apply_collab_facts(
+                work_order_no=no,
+                facts=seeded,
+                sender_id=sess.sender_id,
+                allowed_keys=keys,
+            )
+
         brief = copy.sea_group_brief(
             work_order_no=no,
             facts=sess.facts,
@@ -638,6 +658,7 @@ class SeaTextInquiryFlow(AirTextInquiryFlow):
         work_order_no: str,
         facts: dict[str, str],
         sender_id: str,
+        allowed_keys: tuple[str, ...] | None = None,
     ) -> dict[str, Any]:
         """
         群里静默写入协同字段。不改六态、不回复、不出报价版本。
@@ -647,10 +668,11 @@ class SeaTextInquiryFlow(AirTextInquiryFlow):
         from agent.schema.field_validate import SEA_COLLAB_FIELDS
 
         no = (work_order_no or "").strip()
+        keys = tuple(allowed_keys) if allowed_keys is not None else SEA_COLLAB_FIELDS
         cleaned = {
             k: str(v).strip()
             for k, v in (facts or {}).items()
-            if k in SEA_COLLAB_FIELDS and str(v or "").strip()
+            if k in keys and str(v or "").strip()
         }
         if not no or not cleaned:
             return {"ok": False, "updated": []}

inquiry-agent/agent/redis_coord/flow_bookmark.py

@@ -0,0 +1,133 @@
+"""
+询价当前书签短协调：HTTP 收字、Worker 识图后补问必须读同一份。
+
+不当六态、不当 checkpoint。只存当前 sender 的补问/核对草稿。
+禁止：写入现网 Redis 前缀；把已出号工单只放这里当主账。
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+from typing import Any, Optional, Protocol
+
+from agent.redis_coord.keys import FLOW_BOOKMARK
+from agent.redis_coord.runtime import get_redis_runtime
+
+logger = logging.getLogger(__name__)
+
+_TTL_SEC = 24 * 3600
+
+
+class BookmarkStore(Protocol):
+    """单测可注入；生产走 Redis。"""
+
+    def save(self, kind: str, sender_id: str, payload: dict[str, Any]) -> None: ...
+
+    def load(self, kind: str, sender_id: str) -> Optional[dict[str, Any]]: ...
+
+    def delete(self, kind: str, sender_id: str) -> None: ...
+
+
+class DictBookmarkStore:
+    """两份 Flow 实例共用同一字典，模拟 HTTP/Worker。"""
+
+    def __init__(self) -> None:
+        self._d: dict[str, dict[str, Any]] = {}
+
+    def save(self, kind: str, sender_id: str, payload: dict[str, Any]) -> None:
+        self._d[f"{kind}:{sender_id}"] = json.loads(json.dumps(payload))
+
+    def load(self, kind: str, sender_id: str) -> Optional[dict[str, Any]]:
+        row = self._d.get(f"{kind}:{sender_id}")
+        return json.loads(json.dumps(row)) if row is not None else None
+
+    def delete(self, kind: str, sender_id: str) -> None:
+        self._d.pop(f"{kind}:{sender_id}", None)
+
+
+def _redis_ok() -> bool:
+    try:
+        return get_redis_runtime().client.backend == "redis"
+    except Exception:
+        return False
+
+
+def save_flow_bookmark(
+    kind: str,
+    sender_id: str,
+    payload: dict[str, Any],
+    *,
+    store: Optional[BookmarkStore] = None,
+) -> None:
+    """写下当前书签。store 优先；否则仅真 Redis。"""
+    sid = (sender_id or "").strip()
+    name = (kind or "").strip() or "air"
+    if not sid or not isinstance(payload, dict):
+        return
+    if store is not None:
+        store.save(name, sid, payload)
+        return
+    if not _redis_ok():
+        return
+    try:
+        client = get_redis_runtime().client
+        client.raw.set(
+            client.key(FLOW_BOOKMARK, name, sid),
+            json.dumps(payload, ensure_ascii=False),
+            ex=_TTL_SEC,
+        )
+    except Exception:
+        logger.exception("写询价书签失败 kind=%s sender=%s", name, sid)
+
+
+def load_flow_bookmark(
+    kind: str,
+    sender_id: str,
+    *,
+    store: Optional[BookmarkStore] = None,
+) -> dict[str, Any]:
+    """取出当前书签。没有则空。"""
+    sid = (sender_id or "").strip()
+    name = (kind or "").strip() or "air"
+    if not sid:
+        return {}
+    if store is not None:
+        return dict(store.load(name, sid) or {})
+    if not _redis_ok():
+        return {}
+    try:
+        client = get_redis_runtime().client
+        raw = client.raw.get(client.key(FLOW_BOOKMARK, name, sid))
+        if not raw:
+            return {}
+        if isinstance(raw, bytes):
+            raw = raw.decode("utf-8")
+        data = json.loads(raw)
+        return dict(data) if isinstance(data, dict) else {}
+    except Exception:
+        logger.exception("读书签失败 kind=%s sender=%s", name, sid)
+        return {}
+
+
+def delete_flow_bookmark(
+    kind: str,
+    sender_id: str,
+    *,
+    store: Optional[BookmarkStore] = None,
+) -> None:
+    """发图新开或结束当前草稿时删掉。已出号会话不靠这把键。"""
+    sid = (sender_id or "").strip()
+    name = (kind or "").strip() or "air"
+    if not sid:
+        return
+    if store is not None:
+        store.delete(name, sid)
+        return
+    if not _redis_ok():
+        return
+    try:
+        client = get_redis_runtime().client
+        client.raw.delete(client.key(FLOW_BOOKMARK, name, sid))
+    except Exception:
+        logger.exception("删书签失败 kind=%s sender=%s", name, sid)

inquiry-agent/agent/redis_coord/keys.py

@@ -57,6 +57,11 @@ QUOTE_FILE_READY = "quote_file_ready"
 PENDING_PRODUCT_QUOTE = "pending_product_quote"
 # 陆运核对卡已出、等确定查价：短协调，不当六态、不当工单号
 PENDING_LAND_CONFIRM = "pending_land_confirm"
+# 空运/海运/陆运当前书签：HTTP 与 Worker 共用，不当六态
+FLOW_BOOKMARK = "flow_bookmark"
+# 私聊图片波次：HTTP 与 Worker 必须共用，不当六态
+IMAGE_WAVE = "image_wave"
+IMAGE_WAVE_TEXT = "image_wave_text"
 # 空运群同一句：BOT 与存档补听去重，不当六态
 AIR_INBOUND_MSG = "air_inbound_msg"
 AIR_INBOUND_TEXT = "air_inbound_text"

inquiry-agent/agent/routing/dispatch.py

@@ -14,6 +14,7 @@ from agent.channel.wecom.models import InboundMessage
 from agent.handlers import abnormal, card_action, close_deal, continue_thread, new_inquiry
 from agent.handlers.echo_text import handle_echo_text
 from agent.handlers.group_collab import handle_group_collab
+from agent.handlers.image_inquiry import handle_image_inquiry
 from agent.handlers.text_inquiry import handle_text_inquiry, handle_text_other, is_land_confirm_action
 from agent.routing.decision import RouteDecision
 
@@ -56,6 +57,11 @@ def _wrap_new(message: InboundMessage) -> None:
     new_inquiry.handle_new_inquiry(message)
 
 
+def _wrap_image(message: InboundMessage) -> None:
+    """私聊图片询价：认图后当口播新开，复用文字流程。"""
+    handle_image_inquiry(message)
+
+
 def _wrap_continue(message: InboundMessage, *, thread_id: str = "") -> None:
     continue_thread.handle_continue_thread(message, thread_id=thread_id)
 
@@ -112,7 +118,7 @@ HANDLER_REGISTRY: dict[str, HandlerFn] = {
     "ordinary_text_inquiry": _wrap_text_inquiry,
     "ordinary_text_other": _wrap_text_other,
     "attachment_inquiry": _wrap_new,
-    "image_inquiry": _wrap_new,
+    "image_inquiry": _wrap_image,
     "multi_segment_transport": _wrap_new,
     "tms_read_only_quote": _wrap_continue,
     "new_inquiry": _wrap_new,
@@ -203,6 +209,18 @@ def dispatch_inbound(message: InboundMessage, decision: RouteDecision) -> str:
         HANDLER_REGISTRY["reject_no_sender"](message)
         return "reject_no_sender"
 
+    kind = (message.msg_type or "text").lower()
+    if kind == "image":
+        handle_image_inquiry(message)
+        return "image_inquiry"
+    if kind in {"text", ""}:
+        from agent.handlers.image_inquiry import append_wave_text, is_active_image_wave
+
+        if is_active_image_wave(message.sender_id) and (message.content or "").strip():
+            append_wave_text(message.sender_id, message.content or "")
+            logger.info("dispatch 图片波次收字 sender=%s", message.sender_id)
+            return "image_wave_follow_text"
+
     fn = resolve_handler(decision.intent)
     if fn is None:
         logger.warning("dispatch 未知意图 intent=%s → noop", decision.intent)

inquiry-agent/agent/schema/field_validate.py

@@ -64,6 +64,90 @@ AIR_COLLAB_FIELDS = ("是否含电", "是否含磁")
 SEA_COLLAB_DISPLAY = {
     "HS编码": "商品海关编码",
 }
+# 陆运摘要标题：海关编码，不要写成海运的商品海关编码。
+LAND_COLLAB_DISPLAY = {
+    "HS编码": "海关编码",
+}
+# 填写样例允许拉群的类型+线路 → 协同内部键（顺序与规格一致）。
+_LAND_COLLAB_BY_PAIR: dict[tuple[str, str], tuple[str, ...]] = {
+    ("国内运输拼车", "国内长途/零担"): (
+        "客户名称",
+        "包装方式",
+        "货值",
+        "是否为危险品",
+    ),
+    ("国内运输拼车", "国内城配/拖车/打包"): (
+        "客户名称",
+        "包装方式",
+        "货值",
+        "是否为危险品",
+    ),
+    ("国内运输整车", "国内长途/零担"): (
+        "客户名称",
+        "货值",
+        "包装方式",
+        "是否为危险品",
+    ),
+    ("国内运输整车", "国内城配/拖车/打包"): (
+        "客户名称",
+        "货值",
+        "包装方式",
+        "是否为危险品",
+    ),
+    ("跨境集拼", "中港/中亚/中欧"): (
+        "客户名称",
+        "HS编码",
+        "贸易条款",
+        "包装方式",
+        "是否为危险品",
+    ),
+    ("跨境集拼", "东南亚"): (
+        "客户名称",
+        "HS编码",
+        "贸易条款",
+        "包装方式",
+        "是否为危险品",
+    ),
+    ("跨境整车", "中港/中亚/中欧"): (
+        "客户名称",
+        "HS编码",
+        "贸易条款",
+        "货值",
+        "包装方式",
+        "是否为危险品",
+    ),
+    ("跨境整车", "东南亚"): (
+        "客户名称",
+        "HS编码",
+        "贸易条款",
+        "货值",
+        "包装方式",
+        "是否为危险品",
+    ),
+    ("中港整车", "中港/中亚/中欧"): (
+        "客户名称",
+        "HS编码",
+        "车型/数量",
+        "贸易条款",
+        "货值",
+        "包装方式",
+        "是否为危险品",
+    ),
+    ("中港零担/集拼", "中港/中亚/中欧"): (
+        "客户名称",
+        "HS编码",
+        "贸易条款",
+        "包装方式",
+        "是否为危险品",
+    ),
+}
+# 意图分类用：海运六项 ∪ 陆运全部协同键。
+ALL_COLLAB_KEYS = tuple(
+    dict.fromkeys(
+        list(SEA_COLLAB_FIELDS)
+        + ["客户名称", "包装方式", "货值", "是否为危险品", "车型/数量"]
+    )
+)
 
 # 口语里已经带上 TMS 文档规定的单位时，补进空字段。
 # 只认「数字+单位」，不猜港口，不把长宽高换算成立方。
@@ -102,6 +186,11 @@ _ALIASES = {
     "HS CODE": "HS编码",
     "商品海关编码": "HS编码",
     "海关编码": "HS编码",
+    "客户名称": "客户名称",
+    "货值": "货值",
+    "是否为危险品": "是否为危险品",
+    "通关口岸": "通关口岸",
+    "通关口岸（非必填）": "通关口岸",
 }
 
 
@@ -224,11 +313,75 @@ def harvest_oral_measures(text: str, facts: dict[str, str] | None = None) -> dic
     return harvest_named_air_route(raw, merged)
 
 
+def land_collab_fields(land_type: str, route: str) -> tuple[str, ...]:
+    """
+    本单陆运协同内部键，顺序与规格一致。
+
+    对不上填写样例返回空元组，调用方不得按海运六项去列。
+    海关编码内部仍是 HS编码。副作用：无。
+    """
+    from agent.schema.land_options import canonicalize_route, canonicalize_transport_type
+
+    t = canonicalize_transport_type(land_type) or (land_type or "").strip()
+    r = canonicalize_route(route) or (route or "").strip()
+    return _LAND_COLLAB_BY_PAIR.get((t, r), ())
+
+
+def collab_fields_for_ticket(facts: dict[str, str] | None, business_line: str) -> tuple[str, ...]:
+    """
+    按业务线取出群摘要 / 写入允许的协同键。
+
+    LAND 看事实里的运输类型+线路类别；SEA/AIR 用固定名单。
+    """
+    line = (business_line or "").strip().upper()
+    if line == "AIR":
+        return AIR_COLLAB_FIELDS
+    if line == "LAND":
+        src = dict(facts or {})
+        return land_collab_fields(
+            str(src.get("运输类型") or ""),
+            str(src.get("线路类别") or ""),
+        )
+    return SEA_COLLAB_FIELDS
+
+
+def collab_display_name(key: str, business_line: str) -> str:
+    """协同项对外标题。陆运 HS编码 → 海关编码；海运 → 商品海关编码。"""
+    line = (business_line or "").strip().upper()
+    if line == "LAND":
+        return LAND_COLLAB_DISPLAY.get(key, key)
+    return SEA_COLLAB_DISPLAY.get(key, key)
+
+
+def pick_collab_from_facts(
+    facts: dict[str, str] | None,
+    keys: tuple[str, ...],
+) -> dict[str, str]:
+    """
+    从私聊事实里取出已有协同键，供进群预填。
+
+    只取 keys 里非空值。海关编码/商品海关编码经 normalize 折成 HS编码。
+    不在 keys 的询价项（如跨境整车车型/数量）不会被取出。
+    不覆盖调用方已有协同值；本函数只返回候选。
+    """
+    allowed = set(keys or ())
+    if not allowed:
+        return {}
+    src = normalize_facts(facts)
+    out: dict[str, str] = {}
+    for key in keys:
+        val = (src.get(key) or "").strip()
+        if val:
+            out[key] = val
+    return out
+
+
 def harvest_oral_collab(text: str) -> dict[str, str]:
     """
-    群里口语协同三项：含油/含电/含磁、货好时间。
+    群里口语协同：含油/含电/含磁、货好时间、是否为危险品。
 
-    只认原话里已经说死的词（不含电、随时可提），不猜贸易条款和海关编码。
+    只认原话里已经说死的词（不含电、随时可提、普货），不猜贸易条款和海关编码。
+    危险品：否词（不是危险品/非危/普货）先于是词，避免「不是危险品」收成是。
     """
     raw = text or ""
     out: dict[str, str] = {}
@@ -236,6 +389,11 @@ def harvest_oral_collab(text: str) -> dict[str, str]:
         ("是否含电", ("不含电", "不带电", "无电"), ("含电", "带电")),
         ("是否含油", ("不含油", "不带油", "无油"), ("含油", "带油")),
         ("是否含磁", ("不含磁", "不带磁", "无磁"), ("含磁", "带磁")),
+        (
+            "是否为危险品",
+            ("不是危险品", "非危险品", "非危", "普货"),
+            ("危险品", "有危", "含危"),
+        ),
     )
     for key, no_words, yes_words in yes_no:
         if any(w in raw for w in no_words):
@@ -279,7 +437,8 @@ def _pricing_required_keys(
     land_subtype: str,
     facts: dict[str, str] | None = None,
 ) -> list[str]:
-    """当前激活的首次查价内部键列表。中港再按装载方式收窄。"""
+    """当前激活的首次查价内部键列表。中港整车/零担是两个运输类型。"""
+    _ = facts
     line = (business_line or "").strip().upper()
     pricing = contract.get("pricing_required") or {}
     if line == "SEA":
@@ -291,14 +450,6 @@ def _pricing_required_keys(
         block = land.get(land_subtype) or {}
         if not block:
             return []
-        if land_subtype == "中港":
-            from agent.schema.land_options import canonicalize_hk_load
-
-            load = canonicalize_hk_load(str((facts or {}).get("中港装载方式") or ""))
-            by_load = block.get("by_hk_load") or {}
-            picked = by_load.get(load) if load else None
-            if isinstance(picked, list):
-                return list(picked)
         return list(block.get("always") or [])
     return []
 

inquiry-agent/agent/schema/land_options.py

@@ -1,61 +1,64 @@
 """
-陆运选项规则：运输类型、线路类别、运输分类、中港装载方式。
+陆运选项规则：运输类型、线路类别、运输分类。
 
-本文件职责：从销售原话收取标准项、校验联动、生成补问清单、判断「确定查价」意图。
-联动错误只纠正那一项能选什么，不整单作废。
-禁止：改六态；发明枚举；把选项当货物字段。
+本文件职责：从销售原话收取标准项、校验是否在名单内、生成补问清单（含填写样例表）、判断「确定查价」意图。
+三项各自必填，组合不拦截查价。拉群必须对上填写样例。运输分类句里的「中港/东南亚」不得冒充类型或线路。
+禁止：改六态；发明枚举；把选项当货物字段；再要单独的中港装载方式。
 线程：纯函数，无共享可变状态，可在 Worker 槽调用。
 """
 
 from __future__ import annotations
 
+from pathlib import Path
 from typing import Any
 
-# 与后台员工管理陆运「负责线路」同一套字，禁止改字。
 TRANSPORT_TYPES = (
     "国内运输拼车",
     "国内运输整车",
     "跨境集拼",
     "跨境整车",
-    "中港",
+    "中港整车",
+    "中港零担/集拼",
 )
-DOMESTIC_ROUTES = ("国内长途/零担", "国内城配/拖车/打包")
-CROSS_ROUTES = ("中港/中亚/中欧", "东南亚")
-ROUTES_BY_TYPE = {
-    "国内运输拼车": DOMESTIC_ROUTES,
-    "国内运输整车": DOMESTIC_ROUTES,
-    "跨境集拼": CROSS_ROUTES,
-    "跨境整车": CROSS_ROUTES,
-    "中港": CROSS_ROUTES,
-}
-CROSS_CLASSES = (
-    "普件，中港、东南亚常用线路/常规车型",
-    "普件，中亚、中欧、东南亚涉及国内+国外段",
-    "普件，中亚、中欧、东南亚涉及境外段清派等特殊要求",
+ROUTES = (
+    "国内长途/零担",
+    "国内城配/拖车/打包",
+    "中港/中亚/中欧",
+    "东南亚",
+)
+LAND_CLASSES = (
+    "普件(东南亚、中港常用线路/常规车型)",
+    "普件(东南亚、中亚、中欧涉及国内+国外段)",
+    "普件(东南亚、中亚、中欧涉及境外段清派等特殊要求)",
     "急件",
 )
-DOMESTIC_CLASSES = (
-    "普件，常用线路/常规车型",
-    "普件，非常用线路/临时线路/特种车型",
-    "急件",
-)
-CLASSES_BY_ROUTE = {
-    "中港/中亚/中欧": CROSS_CLASSES,
-    "东南亚": CROSS_CLASSES,
-    "国内长途/零担": DOMESTIC_CLASSES,
-    "国内城配/拖车/打包": DOMESTIC_CLASSES,
-}
-HK_LOADS = ("整车", "零担/集拼")
 
-# 口语别名 → 标准运输类型。长词优先，避免「拼车」误伤「整车」。
+# 填写样例，同时作为拉群门禁：对不上的类型+线路可以查价，但不能拉群。
+SAMPLE_PAIRS = (
+    ("国内运输拼车", "国内长途/零担"),
+    ("国内运输拼车", "国内城配/拖车/打包"),
+    ("国内运输整车", "国内长途/零担"),
+    ("国内运输整车", "国内城配/拖车/打包"),
+    ("跨境集拼", "中港/中亚/中欧"),
+    ("跨境集拼", "东南亚"),
+    ("跨境整车", "中港/中亚/中欧"),
+    ("跨境整车", "东南亚"),
+    ("中港整车", "中港/中亚/中欧"),
+    ("中港零担/集拼", "中港/中亚/中欧"),
+)
+
+# 口语别名 → 标准运输类型。长词优先。禁止把单独「中港」收成类型。
 _TYPE_ALIASES = (
     ("国内运输拼车", "国内运输拼车"),
     ("国内运输整车", "国内运输整车"),
+    ("中港零担/集拼", "中港零担/集拼"),
+    ("中港零担", "中港零担/集拼"),
+    ("中港集拼", "中港零担/集拼"),
+    ("中港整车", "中港整车"),
     ("国内拼车", "国内运输拼车"),
     ("国内整车", "国内运输整车"),
     ("跨境集拼", "跨境集拼"),
     ("跨境整车", "跨境整车"),
-    ("中港", "中港"),
 )
 
 _ROUTE_ALIASES = (
@@ -67,20 +70,9 @@ _ROUTE_ALIASES = (
     ("东南亚", "东南亚"),
 )
 
-# 分类全称从长到短；「急件」单独出现才算急件，避免普件句被误收。
+# 分类全称从长到短；「急件」单独出现才算急件。
 _CLASS_CANON = tuple(
-    sorted(
-        {*(CROSS_CLASSES + DOMESTIC_CLASSES)},
-        key=len,
-        reverse=True,
-    )
-)
-
-_HK_LOAD_ALIASES = (
-    ("零担/集拼", "零担/集拼"),
-    ("零担", "零担/集拼"),
-    ("集拼", "零担/集拼"),
-    ("整车", "整车"),
+    sorted(LAND_CLASSES, key=len, reverse=True)
 )
 
 _CONFIRM_PHRASES = ("可以查了", "确定", "确认", "查价")
@@ -92,6 +84,7 @@ def canonicalize_transport_type(raw: str) -> str:
     把销售原话收成标准运输类型。
 
     参数：任意一句。返回 TRANSPORT_TYPES 之一或空串。
+    只说「中港」不算选完，避免分类句里的中港把类型改掉。
     副作用：无。先匹配更长别名。
     """
     text = (raw or "").strip()
@@ -104,7 +97,7 @@ def canonicalize_transport_type(raw: str) -> str:
 
 
 def canonicalize_route(raw: str) -> str:
-    """把原话收成标准线路类别；对不上返回空串。"""
+    """把原话收成标准线路类别；对不上返回空串。不把单独「中港」收成中港/中亚/中欧。"""
     text = (raw or "").strip()
     if not text:
         return ""
@@ -136,14 +129,25 @@ def canonicalize_land_class(raw: str) -> str:
     return ""
 
 
+def sample_pair_allowed(land_type: str, route: str) -> bool:
+    """
+    类型+线路是否在填写样例里。
+
+    查价不调用本函数；点「拉产品进群协同」时必须为真，否则不建群。
+    入参先收成标准项。副作用：无。
+    """
+    t = canonicalize_transport_type(land_type) or (land_type or "").strip()
+    r = canonicalize_route(route) or (route or "").strip()
+    return (t, r) in SAMPLE_PAIRS
+
+
 def canonicalize_hk_load(raw: str) -> str:
-    """中港装载方式：整车或零担/集拼。"""
-    text = (raw or "").strip()
-    if not text:
-        return ""
-    for needle, canon in _HK_LOAD_ALIASES:
-        if needle in text:
-            return canon
+    """
+    旧中港装载方式不再收取。
+
+    保留函数名避免旧调用处报缺符号；恒返回空串。新单用运输类型「中港整车」「中港零担/集拼」。
+    """
+    _ = raw
     return ""
 
 
@@ -151,9 +155,9 @@ def harvest_land_options(text: str, current: dict[str, str] | None = None) -> di
     """
     从一句原话收取陆运选项，合并到已有值上。
 
+    先收运输分类并把命中句从原文挖掉，再收类型/线路，避免分类里的东南亚、中港冒充选项。
     已有非空项默认保留；本句明确写出另一标准项时才覆盖。
-    已有值对不上标准枚举则丢掉，避免模型把「陆运」写成运输类型后整张清单重问。
-    不改货物字段。返回新 dict，不修改 current。
+    已有值对不上标准枚举则丢掉。不改货物字段。返回新 dict，不修改 current。
     """
     merged = {
         k: str(v).strip()
@@ -162,23 +166,22 @@ def harvest_land_options(text: str, current: dict[str, str] | None = None) -> di
     }
     _keep_canonical_options(merged)
     blob = text or ""
-    found_type = canonicalize_transport_type(blob)
-    found_route = canonicalize_route(blob)
     found_class = canonicalize_land_class(blob)
-    found_load = canonicalize_hk_load(blob)
+    remainder = blob.replace(found_class, " ") if found_class else blob
+    found_type = canonicalize_transport_type(remainder)
+    found_route = canonicalize_route(remainder)
     if found_type:
         merged["运输类型"] = found_type
     if found_route:
         merged["线路类别"] = found_route
     if found_class:
         merged["运输分类"] = found_class
-    if found_load:
-        merged["中港装载方式"] = found_load
+    merged.pop("中港装载方式", None)
     return merged
 
 
 def _keep_canonical_options(merged: dict[str, str]) -> None:
-    """只留能对上枚举的选项；模型胡写的「陆运」不能冒充运输类型。"""
+    """只留能对上枚举的选项；模型胡写的「陆运」不能冒充运输类型。旧中港两项作废。"""
     t = canonicalize_transport_type(merged.get("运输类型") or "")
     if t:
         merged["运输类型"] = t
@@ -194,138 +197,124 @@ def _keep_canonical_options(merged: dict[str, str]) -> None:
         merged["运输分类"] = c
     else:
         merged.pop("运输分类", None)
-    load = canonicalize_hk_load(merged.get("中港装载方式") or "")
-    if load:
-        merged["中港装载方式"] = load
-    else:
-        merged.pop("中港装载方式", None)
+    merged.pop("中港装载方式", None)
 
 
 def validate_land_options(opts: dict[str, str]) -> dict[str, Any]:
     """
-    校验选项是否齐全且联动正确。
+    校验三项是否都在名单内。组合不拦截。
 
-    返回 ok / missing / errors。errors 每项 {field, message}，只纠正错的那一项。
-    中港缺装载方式进 missing，不改货物字段。
+    返回 ok / missing / errors。对不上名单的项当缺项，不写联动纠正句。
     副作用：无。
     """
     src = {k: str(v).strip() for k, v in dict(opts or {}).items() if str(v).strip()}
-    t = canonicalize_transport_type(src.get("运输类型") or "") or src.get("运输类型") or ""
-    r = canonicalize_route(src.get("线路类别") or "") or src.get("线路类别") or ""
-    c = canonicalize_land_class(src.get("运输分类") or "") or src.get("运输分类") or ""
-    load = canonicalize_hk_load(src.get("中港装载方式") or "") or src.get("中港装载方式") or ""
+    t = canonicalize_transport_type(src.get("运输类型") or "")
+    r = canonicalize_route(src.get("线路类别") or "")
+    c = canonicalize_land_class(src.get("运输分类") or "")
 
     missing: list[str] = []
-    errors: list[dict[str, str]] = []
-
     if t not in TRANSPORT_TYPES:
         missing.append("运输类型")
         t = ""
-
-    if t:
-        allowed_routes = ROUTES_BY_TYPE.get(t) or ()
-        if not r:
-            missing.append("线路类别")
-        elif r not in allowed_routes:
-            errors.append(
-                {
-                    "field": "线路类别",
-                    "message": "线路类别只能选：" + "、".join(f"「{x}」" for x in allowed_routes),
-                }
-            )
-            r = ""
-    elif r:
-        # 类型还没有，线路先记下但不判联动
-        pass
-
-    if r:
-        allowed_cls = CLASSES_BY_ROUTE.get(r) or ()
-        if not c:
-            missing.append("运输分类")
-        elif c not in allowed_cls:
-            errors.append(
-                {
-                    "field": "运输分类",
-                    "message": "运输分类只能选：" + "、".join(f"「{x}」" for x in allowed_cls),
-                }
-            )
-            c = ""
-    elif "运输分类" not in missing and not c:
+    if r not in ROUTES:
+        missing.append("线路类别")
+        r = ""
+    if c not in LAND_CLASSES:
         missing.append("运输分类")
+        c = ""
 
-    if t == "中港":
-        if load not in HK_LOADS:
-            missing.append("中港装载方式")
-    ok = not missing and not errors
+    ok = not missing
     return {
         "ok": ok,
         "missing": missing,
-        "errors": errors,
+        "errors": [],
         "normalized": {
             "运输类型": t,
             "线路类别": r,
             "运输分类": c,
-            "中港装载方式": load if t == "中港" else "",
+            "中港装载方式": "",
         },
     }
 
 
+def _class_prompt_lines() -> list[str]:
+    lines = ["运输分类（必填）：可选择回复，"]
+    for i, name in enumerate(LAND_CLASSES, start=1):
+        lines.append(f"{i}.{name}")
+    return lines
+
+
+def option_prompt_needs_sample(*, current: dict[str, str]) -> bool:
+    """运输类型或线路类别缺任一，就要发填写样例图。只缺分类不发。"""
+    src = {k: str(v).strip() for k, v in dict(current or {}).items() if str(v).strip()}
+    t = canonicalize_transport_type(src.get("运输类型") or "")
+    r = canonicalize_route(src.get("线路类别") or "")
+    return t not in TRANSPORT_TYPES or r not in ROUTES
+
+
+def sample_image_path() -> Path:
+    """填写样例表 PNG。和源码放一起，随 agent 发布包走。"""
+    return Path(__file__).resolve().parent.parent / "assets" / "land_sample_pairs.png"
+
+
+def sample_fallback_text() -> str:
+    """
+    图片发不出去时的文字兜底。
+
+    一行一种搭配，不再画框线。
+    """
+    lines = ["填写样例"]
+    for i, (typ, route) in enumerate(SAMPLE_PAIRS, start=1):
+        lines.append(f"{i}.{typ} + {route}")
+    return "\n".join(lines)
+
+
+def _sample_table() -> str:
+    """兼容旧名：补问正文不再内嵌样例，只作图片失败兜底。"""
+    return sample_fallback_text()
+
+
 def option_prompt(*, current: dict[str, str]) -> str:
     """
     按已识别项生成补问清单。
 
-    全缺：一次列出类型、线路、分类、中港装载。
-    类型已知：不再问类型，只列合法线路/分类。
+    缺哪项列哪项。运输类型或线路类别缺任一则提示见图，样例图另发出站。只缺分类不附图。
     不修改 current。
     """
     src = {k: str(v).strip() for k, v in dict(current or {}).items() if str(v).strip()}
-    t = canonicalize_transport_type(src.get("运输类型") or "") or src.get("运输类型") or ""
-    r = canonicalize_route(src.get("线路类别") or "") or src.get("线路类别") or ""
+    t = canonicalize_transport_type(src.get("运输类型") or "")
+    r = canonicalize_route(src.get("线路类别") or "")
+    c = canonicalize_land_class(src.get("运输分类") or "")
+    need_type = t not in TRANSPORT_TYPES
+    need_route = r not in ROUTES
+    need_class = c not in LAND_CLASSES
     lines: list[str] = []
 
-    if t not in TRANSPORT_TYPES:
-        lines.append("请一次选择：运输类型、线路类别、运输分类。")
+    if need_type and need_route and need_class:
+        lines.append("请一次选择：运输类型、线路类别、运输分类")
         lines.append("")
-        lines.append("运输类型：")
-        lines.append("、".join(TRANSPORT_TYPES))
+        lines.append("运输类型（必填）：可选择回复，" + "、".join(TRANSPORT_TYPES))
         lines.append("")
-        lines.append("线路类别（跟运输类型联动）：")
-        lines.append("国内运输拼车 / 国内运输整车：" + "、".join(DOMESTIC_ROUTES))
-        lines.append("跨境集拼 / 跨境整车 / 中港：" + "、".join(CROSS_ROUTES))
+        lines.append("线路类别（必填）：可选择回复，" + "、".join(ROUTES))
         lines.append("")
-        lines.append("运输分类（跟线路类别联动）：")
-        lines.append("中港/中亚/中欧、东南亚：" + "；".join(CROSS_CLASSES))
-        lines.append("国内长途/零担、国内城配/拖车/打包：" + "；".join(DOMESTIC_CLASSES))
+        lines.extend(_class_prompt_lines())
         lines.append("")
-        lines.append("运输类型选中港时，还要选择：" + "、".join(HK_LOADS))
+        lines.append("填写样例见图。")
         return "\n".join(lines).strip()
 
-    # 类型已认出来：不再问运输类型
-    allowed_routes = ROUTES_BY_TYPE.get(t) or ()
-    if r not in allowed_routes:
-        lines.append(f"运输类型已选「{t}」。请选择线路类别：")
-        lines.append("、".join(allowed_routes))
-        route_for_class = ""
-    else:
-        route_for_class = r
-
-    if route_for_class:
-        allowed_cls = CLASSES_BY_ROUTE.get(route_for_class) or ()
-        if src.get("运输分类") not in allowed_cls:
-            lines.append(f"线路类别已选「{route_for_class}」。请选择运输分类：")
-            lines.append("；".join(allowed_cls))
-    elif t in TRANSPORT_TYPES:
-        # 线路未定，把该类型下各线路的分类一并列出，方便一次粘贴
-        seen: set[str] = set()
-        lines.append("运输分类（跟线路类别联动）：")
-        for route in allowed_routes:
-            if route in seen:
-                continue
-            seen.add(route)
-            lines.append(f"{route}：" + "；".join(CLASSES_BY_ROUTE.get(route) or ()))
-
-    if t == "中港" and (src.get("中港装载方式") not in HK_LOADS):
-        lines.append("中港请再选择装载方式：" + "、".join(HK_LOADS))
+    if need_type:
+        lines.append("运输类型（必填）：可选择回复，" + "、".join(TRANSPORT_TYPES))
+    if need_route:
+        if lines:
+            lines.append("")
+        lines.append("线路类别（必填）：可选择回复，" + "、".join(ROUTES))
+    if need_class:
+        if lines:
+            lines.append("")
+        lines.extend(_class_prompt_lines())
+    if need_type or need_route:
+        lines.append("")
+        lines.append("填写样例见图。")
     return "\n".join(lines).strip()
 
 

inquiry-agent/agent/schema/tms_air_query.py

@@ -47,6 +47,35 @@ _PACKS = {
 }
 
 _IATA = re.compile(r"^[A-Za-z]{3}$")
+_IATA_PAIR = re.compile(
+    r"(?<![A-Za-z])([A-Za-z]{3})\s*[-–—/]\s*([A-Za-z]{3})(?![A-Za-z])"
+)
+# 原话里的 AAA-BBB 不一定是航线，货币/单位不能当成起运目的。
+_IATA_PAIR_BLOCK = frozenset(
+    {
+        "USD",
+        "CNY",
+        "EUR",
+        "GBP",
+        "HKD",
+        "JPY",
+        "CBM",
+        "KGS",
+        "PCS",
+        "PKG",
+        "PDF",
+        "JPG",
+        "PNG",
+        "AND",
+        "THE",
+        "FOR",
+        "NOT",
+        "MAX",
+        "MIN",
+        "GMT",
+        "UTC",
+    }
+)
 _NUMBER = re.compile(r"(\d+(?:\.\d+)?)")
 
 
@@ -67,12 +96,24 @@ def resolve_airport_code(raw: str) -> str:
 
 def harvest_named_air_route(text: str, facts: dict[str, str] | None = None) -> dict[str, str]:
     """
-    当前原话里的「从南京飞吉隆坡 / 北京飞吉隆坡」覆盖起运地、目的地。
+    当前原话里的航线覆盖起运地、目的地。
 
-    只认对照表里的地名，不猜没写过的港口。换城市必须丢掉旧三字码。
+    图上/原话写了 SZX-CGK 这类三字码对，必须保持英文，禁止改写成深圳/雅加达。
+    「从南京飞吉隆坡」仍按对照表收中文地名。换城市必须丢掉旧三字码。
     """
     merged = dict(facts or {})
     raw = text or ""
+    pair = _IATA_PAIR.search(raw)
+    if pair:
+        origin_code = pair.group(1).upper()
+        dest_code = pair.group(2).upper()
+        blocked = origin_code in _IATA_PAIR_BLOCK or dest_code in _IATA_PAIR_BLOCK
+        if origin_code != dest_code and not blocked:
+            merged["起运港"] = origin_code
+            merged["目的港"] = dest_code
+            merged.pop("originCode", None)
+            merged.pop("destinationCode", None)
+            return merged
     names = sorted(_AIRPORTS.keys(), key=len, reverse=True)
     for origin in names:
         for dest in names:

inquiry-agent/agent/schema/tms_land_query.py

@@ -4,8 +4,8 @@
 本文件职责：核对卡通过后、建单查价前，确认当前运输类型的货物字段能拼出查价正文。
 带去：运输类型、线路类别、始发/目的/品名/报价日期，以及该类型的体积重量或车型/数量；通关口岸有值才带。
 出站类型/线路必须译成 TMS 6.1.1 英文枚举，线路键名 routeCategory。工单与卡片仍用中文。
-中港整车不带车型/数量。车型/数量是一个字段，出站再拆。
-禁止：把运输分类、中港装载方式放进出站；直连 TMS；改六态。
+中港整车、中港零担/集拼出站类型都是 CHINA_HK。中港整车不带车型/数量。
+禁止：把运输分类放进出站；直连 TMS；改六态。
 线程：纯函数。
 """
 
@@ -16,20 +16,21 @@ import re
 from typing import Any
 
 from agent.schema.field_validate import apply_quote_date_default, normalize_facts
-from agent.schema.land_options import canonicalize_hk_load
 
 logger = logging.getLogger(__name__)
 
 LCL_TYPES = frozenset({"国内运输拼车", "跨境集拼"})
-FTL_TYPES = frozenset({"国内运输整车", "跨境整车"})
-CUSTOMS_TYPES = frozenset({"跨境集拼", "跨境整车", "中港"})
+FTL_TYPES = frozenset({"国内运输整车", "跨境整车", "中港零担/集拼"})
+HK_FTL_TYPE = "中港整车"
+CUSTOMS_TYPES = frozenset({"跨境集拼", "跨境整车", "中港整车", "中港零担/集拼"})
 
 _TYPE_ENUM = {
     "国内运输拼车": "DOMESTIC_LTL",
     "国内运输整车": "DOMESTIC_FTL",
     "跨境集拼": "CROSS_BORDER_LTL",
     "跨境整车": "CROSS_BORDER_FTL",
-    "中港": "CHINA_HK",
+    "中港整车": "CHINA_HK",
+    "中港零担/集拼": "CHINA_HK",
 }
 _ROUTE_ENUM = {
     "国内长途/零担": "DOMESTIC_LONG_LTL",
@@ -91,13 +92,12 @@ def assemble_land_query(facts: dict[str, Any] | None) -> dict[str, Any]:
     """
     把已采集陆运字段拼成 TMS 查价参数。
 
-    不齐则 ok=False，missing 给销售补问。齐了也不带分类/中港装载。
+    不齐则 ok=False，missing 给销售补问。齐了也不带分类。
     改了运输类型后，outbound 只带当前类型需要的键。
     """
     src = apply_quote_date_default(
         normalize_facts({str(k): v for k, v in dict(facts or {}).items() if v is not None})
     )
-    # 选项键不走港口别名折叠，再从原文补回
     raw = {str(k): str(v).strip() for k, v in dict(facts or {}).items() if v is not None}
     land_type = (raw.get("运输类型") or src.get("运输类型") or "").strip()
     route = (raw.get("线路类别") or src.get("线路类别") or "").strip()
@@ -106,7 +106,6 @@ def assemble_land_query(facts: dict[str, Any] | None) -> dict[str, Any]:
     cargo = (src.get("品名") or "").strip()
     quote_date = (src.get("报价日期") or "").strip()
     customs = (raw.get("通关口岸") or src.get("通关口岸") or "").strip()
-    hk_load = canonicalize_hk_load(raw.get("中港装载方式") or src.get("中港装载方式") or "")
 
     missing: list[str] = []
     if not land_type:
@@ -141,10 +140,9 @@ def assemble_land_query(facts: dict[str, Any] | None) -> dict[str, Any]:
             missing.append("毛重")
         else:
             outbound["weightKg"] = wgt
-    elif land_type == "中港" and hk_load != "零担/集拼":
-        # 中港整车（或尚未选定装载）不采集、不带出车型/数量
+    elif land_type == HK_FTL_TYPE:
         pass
-    elif land_type in FTL_TYPES or land_type == "中港":
+    elif land_type in FTL_TYPES:
         combo = (
             src.get("车型/数量")
             or raw.get("车型/数量")

inquiry-agent/agent/worker_app.py

@@ -30,6 +30,8 @@ def main() -> None:
         level=logging.INFO,
         format="%(asctime)s %(levelname)s [%(name)s] %(message)s",
     )
+    # Worker 自己下素材，必须关掉 httpx INFO，否则 access_token 会进日志。
+    logging.getLogger("httpx").setLevel(logging.WARNING)
     settings = get_settings()
     graph_runtime = get_graph_runtime(settings)
     redis_runtime = get_redis_runtime(settings)

inquiry-agent/tests/test_air_group_collab.py

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

inquiry-agent/tests/test_air_text_inquiry.py

@@ -149,6 +149,33 @@ class AirTextInquiryTests(unittest.TestCase):
         self.assertNotIn("请补充运输方式", joined)
         self.assertEqual(self.flow.session_of("u_fly").business_line, "AIR")
 
+    def test_worker_clarify_http_pallet_keeps_air(self) -> None:
+        """识图在 Worker 写下补问，HTTP 补「托盘」必须仍是空运，不能再问运输方式。"""
+        from agent.redis_coord.flow_bookmark import DictBookmarkStore
+
+        store = DictBookmarkStore()
+        worker = AirTextInquiryFlow(MemoryLedger(), bookmark_store=store)
+        http = AirTextInquiryFlow(MemoryLedger(), bookmark_store=store)
+        replies: list[str] = []
+
+        def reply(text: str, extra=None) -> None:
+            replies.append(text)
+
+        phase = worker.on_text(
+            sender_id="u_img",
+            text="空运 深圳到雅加达 发光二级管组件 11P 2400KG 20cbm",
+            reply=reply,
+        )
+        self.assertEqual(phase, "clarify")
+        replies.clear()
+        phase = http.on_text(sender_id="u_img", text="托盘", reply=reply)
+        self.assertNotEqual(phase, "need_mode")
+        self.assertNotIn("请补充运输方式", "\n".join(replies))
+        sess = http.session_of("u_img")
+        self.assertIsNotNone(sess)
+        self.assertEqual(sess.business_line, "AIR")
+        self.assertEqual(sess.facts.get("包装方式"), "托盘")
+
     def test_missing_mode_asks_transport(self) -> None:
         phase = self.flow.on_text(
             sender_id="u1",

inquiry-agent/tests/test_extract_oral.py

@@ -15,7 +15,7 @@ if ROOT not in sys.path:
 
 os.environ.setdefault("LEDGER_BACKEND", "memory")
 
-from agent.llm.extract_text import extract_inquiry_snapshot
+from agent.llm.extract_text import extract_inquiry_snapshot, parse_labeled_facts
 from agent.llm.mode_extract_fields import invoke_shell, parse_extract_payload
 from agent.policy.air_text_flow import AirTextInquiryFlow
 from agent.ledger.memory_ledger import MemoryLedger
@@ -65,6 +65,22 @@ class OralExtractTests(unittest.TestCase):
         self.assertEqual(snap["business_line"], "AIR")
         self.assertFalse(snap["facts"].get("起运港"))
 
+    def test_labeled_customs_port(self) -> None:
+        self.assertEqual(parse_labeled_facts("通关口岸：皇岗").get("通关口岸"), "皇岗")
+        snap = extract_inquiry_snapshot("通关口岸（非必填）：皇岗")
+        self.assertEqual(snap["facts"].get("通关口岸"), "皇岗")
+        snap2 = extract_inquiry_snapshot("通关口岸皇岗")
+        self.assertEqual(snap2["facts"].get("通关口岸"), "皇岗")
+
+    def test_b_keeps_land_customs_port(self) -> None:
+        parsed = parse_extract_payload(
+            {
+                "transport_mode": "LAND",
+                "fields": {"起运港": "广州", "通关口岸": "皇岗"},
+            }
+        )
+        self.assertEqual(parsed["facts"].get("通关口岸"), "皇岗")
+
     def test_assemble_air_query_zhuhai_clark(self) -> None:
         from agent.schema.tms_air_query import assemble_air_query
 
@@ -134,6 +150,46 @@ class OralExtractTests(unittest.TestCase):
         self.assertNotIn("起运地：NKG", shown)
         self.assertNotIn("起运地：南京", shown)
 
+    def test_iata_pair_kept_not_rewritten_to_city(self) -> None:
+        from agent.schema.field_validate import harvest_oral_measures
+        from agent.schema.tms_air_query import assemble_air_query
+        from agent.policy import inquiry_copy as copy
+
+        oral = "空运 SZX-CGK 发光二级管组件 11P 2400KG 20cbm"
+        facts = harvest_oral_measures(
+            oral,
+            {"起运港": "深圳", "目的港": "雅加达", "品名": "发光二级管组件"},
+        )
+        self.assertEqual(facts["起运港"], "SZX")
+        self.assertEqual(facts["目的港"], "CGK")
+        assembled = assemble_air_query(
+            {
+                **facts,
+                "件数": "11P",
+                "毛重": "2400KG",
+                "体积": "20cbm",
+                "包装方式": "托盘",
+                "报价日期": "2026-09-18",
+            }
+        )
+        self.assertEqual(assembled["facts"]["originCode"], "SZX")
+        self.assertEqual(assembled["facts"]["destinationCode"], "CGK")
+        card = copy.inquiry_card(work_order_no="WO1", facts=assembled["facts"])
+        self.assertIn("起运地：SZX", card)
+        self.assertIn("目的地：CGK", card)
+        self.assertNotIn("深圳", card)
+        self.assertNotIn("雅加达", card)
+
+    def test_currency_pair_is_not_air_route(self) -> None:
+        from agent.schema.field_validate import harvest_oral_measures
+
+        facts = harvest_oral_measures(
+            "空运 货值 USD-CNY 茶叶",
+            {"起运港": "上海", "目的港": "洛杉矶"},
+        )
+        self.assertEqual(facts["起运港"], "上海")
+        self.assertEqual(facts["目的港"], "洛杉矶")
+
     def test_oral_pack_and_quote_date_from_current_sentence(self) -> None:
         from agent.schema.field_validate import harvest_oral_measures
 
@@ -225,6 +281,13 @@ class OralExtractTests(unittest.TestCase):
         self.assertEqual(norm["毛重"], "180")
         self.assertEqual(norm["体积"], "2.5")
 
+    def test_extract_prompt_keeps_iata(self) -> None:
+        from agent.llm.mode_extract_fields import build_extract_messages
+
+        blob = str(build_extract_messages("空运 SZX-CGK"))
+        self.assertIn("SZX-CGK", blob)
+        self.assertIn("禁止翻译成城市名", blob)
+
     def test_chat_fn_fixture_drives_flow(self) -> None:
         def chat_fn(**kwargs):
             return SimpleNamespace(

inquiry-agent/tests/test_image_inquiry.py

@@ -0,0 +1,272 @@
+"""
+私聊图片询价：识图当口播、新开一票、复用文字流程。
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+import unittest
+
+ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
+if ROOT not in sys.path:
+    sys.path.insert(0, ROOT)
+
+os.environ.setdefault("YTD_ENV", "test")
+os.environ.setdefault("REDIS_BACKEND", "memory")
+os.environ.setdefault("CHECKPOINT_BACKEND", "memory")
+os.environ.setdefault("MESSAGE_STORE_BACKEND", "memory")
+os.environ.setdefault("LLM_DATA_USAGE_CONFIRMED", "false")
+os.environ.setdefault("LLM_ALLOW_NETWORK", "false")
+os.environ.setdefault("LEDGER_BACKEND", "memory")
+
+from agent.channel.queue import MemoryMessageStore
+from agent.channel.wecom.models import InboundMessage
+from agent.channel.outbox.sender import WeComAppClient
+from agent.handlers.image_inquiry import (
+    flush_active_wave,
+    handle_image_inquiry,
+)
+from agent.handlers.text_inquiry import handle_text_inquiry
+from agent.ledger.memory_ledger import MemoryLedger
+from agent.policy import inquiry_copy as copy
+from agent.policy.air_text_flow import AirTextInquiryFlow, FlowSession, reset_air_text_flow_for_test
+from agent.policy.image_wave import reset_image_wave_book_for_test
+from agent.policy.land_text_flow import get_land_text_flow, reset_land_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
+from tests.test_sea_text_inquiry import COMPLETE_SEA
+
+SEA_ORAL = """海运
+起运港：上海
+目的港：洛杉矶
+品名：普货
+货量：20吨
+整柜或拼柜：整柜
+箱型箱量：1x40HQ
+"""
+
+LAND_ORAL = """陆运
+国内运输拼车
+国内长途/零担
+急件
+起运港：广州
+目的港：深圳
+品名：衣服
+体积：100CBM
+毛重：50kg
+"""
+
+
+def _vision(text: str, *, ok: bool = True, error: str = ""):
+    def _fn(**_kwargs):
+        body = (text or "").strip()
+        return {"ok": ok, "text": body, "empty": not bool(body), "error": error}
+
+    return _fn
+
+
+def _img(sender: str, mid: str, msgid: str = "") -> InboundMessage:
+    return InboundMessage(
+        sender_id=sender,
+        message_id=msgid or mid,
+        content="",
+        msg_type="image",
+        media={"media_id": mid, "filename": f"{mid}.jpg"},
+    )
+
+
+class ImageBookmarkTests(unittest.TestCase):
+    def test_drop_keeps_ticket_session(self) -> None:
+        flow = AirTextInquiryFlow(MemoryLedger())
+        sess = FlowSession(
+            sender_id="s1", thread_id="s1", work_order_no="WO1", phase="quoted"
+        )
+        flow._save(sess)
+        dropped = flow.drop_current_bookmark("s1")
+        self.assertEqual(dropped.work_order_no, "WO1")
+        self.assertIsNone(flow.session_of("s1"))
+        self.assertIsNotNone(flow.session_by_work_order("WO1"))
+
+    def test_download_empty_media_id(self) -> None:
+        client = WeComAppClient(corp_id="c", secret="s", agent_id=1)
+        raw, err = client.download_media("")
+        self.assertEqual(raw, b"")
+        self.assertEqual(err, "media_empty")
+
+
+class ImageInquiryFlowTests(unittest.TestCase):
+    def setUp(self) -> None:
+        reset_sea_text_flow_for_test()
+        reset_land_text_flow_for_test()
+        reset_air_text_flow_for_test()
+        reset_image_wave_book_for_test()
+        self.store = MemoryMessageStore()
+
+    def _bodies(self) -> list[str]:
+        return [i.content for i in self.store._outbox.values()]
+
+    def test_only_image_asks_mode(self) -> None:
+        phase = handle_image_inquiry(
+            _img("s1", "mid-1"),
+            store=self.store,
+            vision_fn=_vision(""),
+            download_fn=lambda _mid: (b"x", ""),
+        )
+        bodies = self._bodies()
+        self.assertEqual(bodies[0], copy.IMAGE_RECOGNIZING)
+        self.assertIn("没看清", bodies[-1])
+        self.assertIn("海运", bodies[-1])
+        self.assertIn("空运", bodies[-1])
+        self.assertEqual(phase, "need_mode")
+
+    def test_sea_visible_creates_ticket(self) -> None:
+        handle_image_inquiry(
+            _img("s1", "mid-sea"),
+            store=self.store,
+            vision_fn=_vision(SEA_ORAL),
+            download_fn=lambda _mid: (b"x", ""),
+            put_fn=lambda **k: {"ok": True, "key": k.get("key") or "k1"},
+        )
+        sea = get_sea_text_flow()
+        sess = sea.session_of("s1")
+        self.assertIsNotNone(sess)
+        self.assertTrue((sess.work_order_no or "").strip())
+        ticket = sea.ledger.get_ticket(work_order_no=sess.work_order_no)
+        self.assertTrue(ticket.attachments)
+
+    def test_burst_two_images_one_ack(self) -> None:
+        handle_image_inquiry(
+            _img("s1", "mid-a", "m-a"),
+            store=self.store,
+            vision_fn=_vision(""),
+            download_fn=lambda _mid: (b"x", ""),
+            process_now=False,
+        )
+        handle_image_inquiry(
+            _img("s1", "mid-b", "m-b"),
+            store=self.store,
+            vision_fn=_vision(SEA_ORAL),
+            download_fn=lambda _mid: (b"x", ""),
+            process_now=False,
+        )
+        flush_active_wave(
+            "s1",
+            store=self.store,
+            vision_fn=_vision(SEA_ORAL),
+            download_fn=lambda _mid: (b"x", ""),
+            seed_message=_img("s1", "mid-b", "m-b"),
+        )
+        acks = [c for c in self._bodies() if c == copy.IMAGE_RECOGNIZING]
+        self.assertEqual(len(acks), 1)
+        sess = get_sea_text_flow().session_of("s1")
+        self.assertTrue(sess and sess.work_order_no)
+
+    def test_precede_text_joins_new_ticket(self) -> None:
+        handle_text_inquiry(
+            InboundMessage(
+                sender_id="s1",
+                message_id="txt1",
+                content="空运 上海到纽约 3件 180KGS 2CBM 托盘",
+            ),
+            store=self.store,
+        )
+        handle_image_inquiry(
+            _img("s1", "mid-air"),
+            store=self.store,
+            vision_fn=_vision("包装方式：托盘"),
+            download_fn=lambda _mid: (b"x", ""),
+        )
+        from agent.policy.air_text_flow import get_air_text_flow
+
+        air = get_air_text_flow().session_of("s1")
+        self.assertIsNotNone(air)
+        self.assertEqual((air.business_line or "").upper(), "AIR")
+
+    def test_land_image_confirm_card_has_no_wo(self) -> None:
+        handle_image_inquiry(
+            _img("s1", "mid-land"),
+            store=self.store,
+            vision_fn=_vision(LAND_ORAL),
+            download_fn=lambda _mid: (b"x", ""),
+        )
+        last = self._bodies()[-1]
+        self.assertIn("请回复“确定”", last)
+        self.assertNotIn("WO", last)
+        sess = get_land_text_flow().session_of("s1")
+        self.assertEqual(sess.phase, "wait_confirm")
+        self.assertFalse((sess.work_order_no or "").strip())
+
+    def test_quoted_then_new_image_new_ticket(self) -> None:
+        handle_text_inquiry(
+            InboundMessage(sender_id="s1", message_id="old", content="海运出号"),
+            store=self.store,
+            injected_facts=COMPLETE_SEA,
+            injected_mode="SEA",
+        )
+        wo_old = get_sea_text_flow().session_of("s1").work_order_no
+        self.assertTrue(wo_old)
+        handle_image_inquiry(
+            _img("s1", "mid-new"),
+            store=self.store,
+            vision_fn=_vision(SEA_ORAL),
+            download_fn=lambda _mid: (b"x", ""),
+        )
+        wo_new = get_sea_text_flow().session_of("s1").work_order_no
+        self.assertTrue(wo_new)
+        self.assertNotEqual(wo_old, wo_new)
+        old = get_sea_text_flow().ledger.get_ticket(work_order_no=wo_old)
+        self.assertIsNotNone(old)
+
+    def test_group_image_not_this_handler(self) -> None:
+        msg = InboundMessage(
+            sender_id="s1",
+            message_id="g1",
+            content="",
+            msg_type="image",
+            chat_id="wr_group",
+            chat_type="group",
+            media={"media_id": "mid-g", "filename": "现场.jpg"},
+        )
+        name = dispatch_inbound(msg, RouteDecision(intent="ordinary_text_other"))
+        self.assertNotEqual(name, "image_inquiry")
+
+    def test_worker_kind_vision_finishes_wave(self) -> None:
+        from agent.handlers.image_inquiry import finish_queued_image
+        from agent.policy.image_wave import get_image_wave_book
+
+        book = reset_image_wave_book_for_test()
+        book.open_or_join("s1", "mid-1", now=1.0)
+        got = finish_queued_image(
+            {
+                "sender_id": "s1",
+                "media_id": "mid-1",
+                "message_id": "m1",
+                "__vision": {"ok": True, "text": "", "empty": True, "error": ""},
+                "__bytes": b"img",
+            }
+        )
+        self.assertTrue(got.get("ok"))
+        self.assertTrue(book._waves["s1"].business_done)
+
+    def test_png_download_sends_png_mime(self) -> None:
+        seen: dict = {}
+        png = b"\x89PNG\r\n\x1a\n" + b"\x00" * 20
+
+        def _fn(**kwargs):
+            seen.update(kwargs.get("payload") or {})
+            return {"ok": True, "text": "起运地：深圳", "empty": False, "error": ""}
+
+        handle_image_inquiry(
+            _img("s1", "mid-png"),
+            store=self.store,
+            vision_fn=_fn,
+            download_fn=lambda _mid: (png, ""),
+        )
+        self.assertEqual(seen.get("mime"), "image/png")
+        self.assertTrue(seen.get("image_b64"))
+
+
+if __name__ == "__main__":
+    unittest.main()

inquiry-agent/tests/test_image_wave.py

@@ -0,0 +1,103 @@
+"""
+私聊图片波次：前一句 10 秒、连发图、业务发出前收字。
+"""
+
+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.policy.image_wave import (
+    ImageWaveBook,
+    PRECEDE_SECONDS,
+    reset_image_wave_book_for_test,
+)
+
+
+class ImageWaveTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.book = reset_image_wave_book_for_test()
+
+    def test_precede_window_is_ten_seconds(self) -> None:
+        self.assertEqual(PRECEDE_SECONDS, 10.0)
+
+    def test_precede_within_10s(self) -> None:
+        self.book.note_text("s1", "空运 上海到纽约", now=100.0)
+        self.assertEqual(self.book.precede_text("s1", now=109.9), "空运 上海到纽约")
+
+    def test_precede_after_10s_dropped(self) -> None:
+        self.book.note_text("s1", "空运 上海到纽约", now=100.0)
+        self.assertEqual(self.book.precede_text("s1", now=110.1), "")
+
+    def test_forget_text_drops_precede(self) -> None:
+        self.book.note_text("s1", "空运 上海到纽约", now=100.0)
+        self.book.forget_text("s1")
+        self.assertEqual(self.book.precede_text("s1", now=100.5), "")
+
+    def test_second_image_joins_before_business(self) -> None:
+        w1, how = self.book.open_or_join("s1", "mid-a", now=1.0)
+        self.assertEqual(how, "opened")
+        w2, how2 = self.book.open_or_join("s1", "mid-b", now=1.2)
+        self.assertEqual(how2, "joined")
+        self.assertEqual(w1.wave_id, w2.wave_id)
+        self.assertEqual(w2.media_ids, ["mid-a", "mid-b"])
+
+    def test_image_after_business_is_new_wave(self) -> None:
+        self.book.open_or_join("s1", "mid-a", now=1.0)
+        self.book.mark_business_done("s1")
+        w2, how = self.book.open_or_join("s1", "mid-b", now=2.0)
+        self.assertEqual(how, "new_after_done")
+        self.assertEqual(w2.media_ids, ["mid-b"])
+
+    def test_follow_text_before_business(self) -> None:
+        self.book.open_or_join("s1", "mid-a", now=1.0)
+        self.assertTrue(self.book.append_follow_text("s1", "海运"))
+        self.book.add_vision("s1", "宁波到汉堡")
+        oral = self.book.compose_oral("s1")
+        self.assertIn("海运", oral)
+        self.assertIn("宁波到汉堡", oral)
+
+    def test_follow_text_after_business_rejected(self) -> None:
+        self.book.open_or_join("s1", "mid-a", now=1.0)
+        self.book.mark_business_done("s1")
+        self.assertFalse(self.book.append_follow_text("s1", "海运"))
+
+    def test_http_and_worker_share_kv(self) -> None:
+        """复现测服事故：HTTP 收图、Worker 认完必须看到同一份结束标记。"""
+        kv = _MemoryWaveKV()
+        http = reset_image_wave_book_for_test(kv=kv)
+        worker = ImageWaveBook(kv=kv)
+        http.open_or_join("s1", "mid-a", now=1.0)
+        self.assertIsNotNone(worker.active("s1"))
+        worker.add_vision("s1", "SZX-CGK 货交深圳")
+        worker.mark_business_done("s1")
+        self.assertIsNone(http.active("s1"))
+        _w2, how = http.open_or_join("s1", "mid-b", now=2.0)
+        self.assertEqual(how, "new_after_done")
+
+
+class _MemoryWaveKV:
+    """单测假 Redis：两份 ImageWaveBook 共用同一字典。"""
+
+    def __init__(self) -> None:
+        self._d: dict[str, dict] = {}
+
+    def get_json(self, kind: str, sender_id: str):
+        row = self._d.get(f"{kind}:{sender_id}")
+        return json.loads(json.dumps(row)) if row is not None else None
+
+    def set_json(self, kind: str, sender_id: str, data: dict) -> None:
+        self._d[f"{kind}:{sender_id}"] = json.loads(json.dumps(data))
+
+    def delete(self, kind: str, sender_id: str) -> None:
+        self._d.pop(f"{kind}:{sender_id}", None)
+
+
+if __name__ == "__main__":
+    unittest.main()

inquiry-agent/tests/test_land_collab_fields.py

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

inquiry-agent/tests/test_land_copy.py

@@ -53,13 +53,12 @@ class LandCopyTest(unittest.TestCase):
         )
         self.assertIn("通关口岸（非必填）：", text)
 
-    def test_hk_shows_load(self):
+    def test_hk_lcl_shows_vehicle_not_load_field(self):
         text = copy.land_confirm_card(
             {
-                "运输类型": "中港",
+                "运输类型": "中港零担/集拼",
                 "线路类别": "东南亚",
                 "运输分类": "急件",
-                "中港装载方式": "零担/集拼",
                 "起运港": "广州",
                 "目的港": "香港",
                 "品名": "衣服",
@@ -67,25 +66,26 @@ class LandCopyTest(unittest.TestCase):
                 "报价日期": "2026-09-17",
             }
         )
-        self.assertIn("零担/集拼", text)
+        self.assertIn("运输类型（必填）：中港零担/集拼", text)
         self.assertIn("车型/数量（必填）：9米6/1", text)
-        self.assertNotIn("\n数量（必填）", text)
+        self.assertNotIn("中港装载方式", text)
+        self.assertIn("通关口岸（非必填）", text)
 
     def test_hk_full_truck_hides_vehicle(self):
         text = copy.land_confirm_card(
             {
-                "运输类型": "中港",
+                "运输类型": "中港整车",
                 "线路类别": "东南亚",
                 "运输分类": "急件",
-                "中港装载方式": "整车",
                 "起运港": "广州",
                 "目的港": "香港",
                 "品名": "衣服",
                 "报价日期": "2026-09-17",
             }
         )
-        self.assertIn("整车", text)
+        self.assertIn("运输类型（必填）：中港整车", text)
         self.assertNotIn("车型/数量", text)
+        self.assertNotIn("中港装载方式", text)
 
     def test_domestic_ftl_one_field(self):
         text = copy.land_confirm_card(
@@ -106,6 +106,57 @@ class LandCopyTest(unittest.TestCase):
     def test_no_staff_copy(self):
         self.assertIn("陆运产品岗", copy.land_group_no_staff())
 
+    def test_ask_land_fields_marks_required_and_lists_optional(self):
+        text = copy.ask_land_fields(
+            facts={
+                "运输类型": "跨境集拼",
+                "线路类别": "东南亚",
+                "运输分类": "急件",
+                "起运港": "广州",
+                "目的港": "河内",
+                "品名": "衣服",
+                "毛重": "50kg",
+            },
+            missing_keys=["体积"],
+            land_subtype="跨境集拼",
+        )
+        self.assertIn("始发站（必填）：广州", text)
+        self.assertIn("待补充：", text)
+        self.assertIn("体积（必填）", text)
+        self.assertIn("通关口岸（非必填）", text)
+        self.assertNotIn(copy.LAND_CONFIRM_TAIL, text)
+
+    def test_ask_land_fields_domestic_has_no_customs(self):
+        text = copy.ask_land_fields(
+            facts={
+                "运输类型": "国内运输拼车",
+                "起运港": "广州",
+                "目的港": "深圳",
+                "品名": "衣服",
+                "毛重": "50kg",
+            },
+            missing_keys=["体积"],
+            land_subtype="国内运输拼车",
+        )
+        self.assertIn("体积（必填）", text)
+        self.assertNotIn("通关口岸", text)
+
+    def test_ask_land_fields_filled_customs_not_pending(self):
+        text = copy.ask_land_fields(
+            facts={
+                "运输类型": "中港整车",
+                "起运港": "广州",
+                "目的港": "香港",
+                "品名": "衣服",
+                "通关口岸": "皇岗",
+            },
+            missing_keys=["起运港"],
+            land_subtype="中港整车",
+        )
+        self.assertIn("通关口岸（非必填）：皇岗", text)
+        pending = text.split("待补充：", 1)[1]
+        self.assertNotIn("通关口岸", pending)
+
     def test_inquiry_card_lists_land_fields_and_wo(self):
         text = copy.inquiry_card(
             work_order_no="WO202609170006",
@@ -134,7 +185,7 @@ class LandCopyTest(unittest.TestCase):
         facts = {
             "运输类型": "国内运输拼车",
             "线路类别": "国内城配/拖车/打包",
-            "运输分类": "普件，常用线路/常规车型",
+            "运输分类": "普件(东南亚、中港常用线路/常规车型)",
             "起运港": "泉州市",
             "目的港": "南京市",
             "品名": "通信机柜及线缆配件",
@@ -201,6 +252,74 @@ class LandCopyTest(unittest.TestCase):
         self.assertIn("9米6/1", rows["货物"])
         self.assertNotIn("CBM", rows["货物"])
 
+    def test_land_brief_only_ltl_collab(self):
+        text = copy.sea_group_brief(
+            work_order_no="WO1",
+            facts={
+                "运输类型": "国内运输拼车",
+                "线路类别": "国内长途/零担",
+                "运输分类": "急件",
+                "起运港": "广州",
+                "目的港": "深圳",
+                "品名": "衣服",
+                "体积": "1",
+                "毛重": "1",
+                "报价日期": "2026-09-18",
+            },
+            quote=None,
+            collab_facts={"包装方式": "纸箱"},
+            sales_name="销",
+            product_names=["产"],
+            has_price=False,
+            transport_mode="陆运",
+        )
+        self.assertIn("包装方式：纸箱", text)
+        self.assertIn("客户名称：-", text)
+        self.assertIn("货值：-", text)
+        self.assertIn("是否为危险品：-", text)
+        self.assertNotIn("贸易条款", text)
+        self.assertNotIn("商品海关编码", text)
+        self.assertNotIn("海关编码", text)
+
+    def test_land_brief_ftl_longhaul_order(self):
+        text = copy.sea_group_brief(
+            work_order_no="WO2",
+            facts={
+                "运输类型": "国内运输整车",
+                "线路类别": "国内长途/零担",
+                "运输分类": "急件",
+                "起运港": "广州",
+                "目的港": "深圳",
+                "品名": "衣服",
+                "车型/数量": "9米/1",
+                "报价日期": "2026-09-18",
+            },
+            quote=None,
+            collab_facts={},
+            sales_name="销",
+            product_names=["产"],
+            has_price=False,
+            transport_mode="陆运",
+        )
+        collab = text.split("协同补充字段")[1]
+        self.assertLess(collab.find("客户名称"), collab.find("货值"))
+        self.assertLess(collab.find("货值"), collab.find("包装方式"))
+        self.assertNotIn("贸易条款", text)
+
+    def test_land_unrecognized_lists_keys(self):
+        keys = ("客户名称", "包装方式", "货值", "是否为危险品")
+        text = copy.group_fields_not_recognized(keys=keys, business_line="LAND")
+        self.assertIn("客户名称", text)
+        self.assertNotIn("货好时间", text)
+
+    def test_image_copy(self):
+        self.assertEqual(copy.IMAGE_RECOGNIZING, "正在识别图片")
+        self.assertIn("没看清", copy.IMAGE_UNREADABLE)
+        self.assertIn("暂时没认出来", copy.IMAGE_TECH_FAIL)
+        merged = copy.image_fail_then_clarify(copy.IMAGE_UNREADABLE, copy.ASK_TRANSPORT)
+        self.assertTrue(merged.startswith(copy.IMAGE_UNREADABLE))
+        self.assertIn(copy.ASK_TRANSPORT, merged)
+
 
 if __name__ == "__main__":
     unittest.main()

inquiry-agent/tests/test_land_field_validate.py

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

inquiry-agent/tests/test_land_group_collab.py

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

inquiry-agent/tests/test_land_options.py

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

inquiry-agent/tests/test_land_text_inquiry.py

@@ -47,9 +47,11 @@ class LandTextInquiryFrontTests(unittest.TestCase):
         self.ledger = CountingLedger()
         self.flow = LandTextInquiryFlow(self.ledger)
         self.replies: list[str] = []
+        self.extras: list[dict] = []
 
     def _reply(self, text: str, extra=None) -> str:
         self.replies.append(text)
+        self.extras.append(dict(extra or {}))
         return ""
 
     def _joined(self) -> str:
@@ -69,7 +71,12 @@ class LandTextInquiryFrontTests(unittest.TestCase):
         self.assertIn("国内运输拼车", text)
         self.assertIn("线路类别", text)
         self.assertIn("运输分类", text)
+        self.assertIn("中港整车", text)
+        self.assertIn("填写样例见图", text)
         self.assertNotIn("请补充陆运类型：", text)
+        img = [x for x in self.extras if x.get("msgtype") == "image"]
+        self.assertEqual(len(img), 1)
+        self.assertTrue(os.path.isfile(str(img[0].get("filepath") or "")))
 
     def test_known_type_does_not_reask_type(self) -> None:
         phase = self.flow.on_text(
@@ -82,8 +89,10 @@ class LandTextInquiryFrontTests(unittest.TestCase):
         self.assertEqual(phase, "need_land_options")
         self.assertNotIn("请选择运输类型", self._joined())
         self.assertIn("国内长途/零担", self._joined())
+        self.assertIn("填写样例见图", self._joined())
+        self.assertTrue(any(x.get("msgtype") == "image" for x in self.extras))
 
-    def test_wrong_route_keeps_type(self) -> None:
+    def test_unlinked_route_still_keeps_type(self) -> None:
         phase = self.flow.on_text(
             sender_id="u1",
             text="国内运输拼车，线路东南亚",
@@ -92,14 +101,16 @@ class LandTextInquiryFrontTests(unittest.TestCase):
             injected_mode="LAND",
         )
         self.assertEqual(phase, "need_land_options")
-        self.assertIn("国内长途/零担", self._joined())
+        self.assertIn("运输分类", self._joined())
+        self.assertNotIn("填写样例", self._joined())
         sess = self.flow.session_of("u1")
         self.assertEqual(sess.facts.get("运输类型"), "国内运输拼车")
+        self.assertEqual(sess.facts.get("线路类别"), "东南亚")
 
     def test_hk_lcl_asks_vehicle(self) -> None:
         phase = self.flow.on_text(
             sender_id="u1",
-            text="中港 东南亚 急件 零担",
+            text="中港零担/集拼 东南亚 急件",
             reply=self._reply,
             injected_facts={
                 "起运港": "广州",
@@ -111,7 +122,8 @@ class LandTextInquiryFrontTests(unittest.TestCase):
             injected_mode="LAND",
         )
         self.assertEqual(phase, "clarify")
-        self.assertIn("车型/数量", self._joined())
+        self.assertIn("车型/数量（必填）", self._joined())
+        self.assertIn("通关口岸（非必填）", self._joined())
         self.assertNotIn("\n数量：", self._joined())
         self.assertNotIn(copy.LAND_CONFIRM_TAIL, self._joined())
         self.assertEqual(self.ledger.create_calls, 0)
@@ -119,7 +131,7 @@ class LandTextInquiryFrontTests(unittest.TestCase):
     def test_hk_full_truck_skips_vehicle(self) -> None:
         phase = self.flow.on_text(
             sender_id="u1",
-            text="中港 东南亚 急件 整车",
+            text="中港整车 东南亚 急件",
             reply=self._reply,
             injected_facts={
                 "起运港": "广州",
@@ -172,6 +184,53 @@ class LandTextInquiryFrontTests(unittest.TestCase):
         self.assertEqual(phase, "wait_confirm")
         self.assertIn("通关口岸（非必填）：", self._joined())
 
+    def test_wait_confirm_can_fill_customs_port(self) -> None:
+        """核对卡空着通关口岸时，再发「通关口岸：皇岗」必须写进下一张卡。"""
+        self.flow.on_text(
+            sender_id="u1",
+            text="跨境集拼 东南亚 急件",
+            reply=self._reply,
+            injected_facts={
+                "起运港": "广州",
+                "目的港": "河内",
+                "品名": "衣服",
+                "体积": "100CBM",
+                "毛重": "50kg",
+            },
+            injected_mode="LAND",
+        )
+        self.replies.clear()
+        phase = self.flow.on_text(
+            sender_id="u1",
+            text="通关口岸：皇岗",
+            reply=self._reply,
+        )
+        self.assertEqual(phase, "wait_confirm")
+        self.assertIn("通关口岸（非必填）：皇岗", self._joined())
+        self.assertEqual(self.flow.session_of("u1").facts.get("通关口岸"), "皇岗")
+
+    def test_wait_confirm_fills_customs_from_card_label(self) -> None:
+        """销售按卡片原样写「通关口岸（非必填）：皇岗」也要收下。"""
+        self.flow.on_text(
+            sender_id="u1",
+            text="中港整车 中港/中亚/中欧 急件",
+            reply=self._reply,
+            injected_facts={
+                "起运港": "广州",
+                "目的港": "香港",
+                "品名": "衣服",
+            },
+            injected_mode="LAND",
+        )
+        self.replies.clear()
+        phase = self.flow.on_text(
+            sender_id="u1",
+            text="通关口岸（非必填）：皇岗",
+            reply=self._reply,
+        )
+        self.assertEqual(phase, "wait_confirm")
+        self.assertEqual(self.flow.session_of("u1").facts.get("通关口岸"), "皇岗")
+
     def test_missing_volume_no_confirm(self) -> None:
         phase = self.flow.on_text(
             sender_id="u1",
@@ -187,6 +246,8 @@ class LandTextInquiryFrontTests(unittest.TestCase):
         )
         self.assertEqual(phase, "clarify")
         self.assertNotIn(copy.LAND_CONFIRM_TAIL, self._joined())
+        self.assertIn("体积（必填）", self._joined())
+        self.assertNotIn("通关口岸", self._joined())
 
 
 class LandTextConfirmTests(unittest.TestCase):
@@ -403,7 +464,7 @@ class LandTextConfirmTests(unittest.TestCase):
         self.replies.clear()
         phase = self.flow.on_text(
             sender_id="u1",
-            text="国内城配/拖车/打包；普件，常用线路/常规车型",
+            text="国内城配/拖车/打包；普件(东南亚、中港常用线路/常规车型)",
             reply=self._reply,
             injected_facts={"品名": "通信机柜及线缆配件", "运输类型": "陆运"},
             injected_mode="LAND",
@@ -413,7 +474,7 @@ class LandTextConfirmTests(unittest.TestCase):
         self.assertNotIn("请一次选择：运输类型", text)
         self.assertIn("国内运输拼车", text)
         self.assertIn("国内城配/拖车/打包", text)
-        self.assertIn("普件，常用线路/常规车型", text)
+        self.assertIn("普件(东南亚、中港常用线路/常规车型)", text)
 
 
 class FakeGroupClient:
@@ -513,6 +574,36 @@ class LandTextPullTests(unittest.TestCase):
         self.assertEqual(self.groups.created, [])
         self.assertEqual(phase, "wait_collab")
 
+    def test_pull_rejects_unlisted_pair(self) -> None:
+        self.ledger.set_land_staff(
+            [
+                {"roleCode": "land", "status": "active", "wecomId": "p_sea", "name": "甲", "routes": "东南亚"},
+            ]
+        )
+        self.flow.on_text(
+            sender_id="u1",
+            text="国内运输拼车 东南亚 急件",
+            reply=self._reply,
+            injected_facts={
+                "起运港": "广州",
+                "目的港": "河内",
+                "品名": "衣服",
+                "体积": "100CBM",
+                "毛重": "50kg",
+            },
+            injected_mode="LAND",
+        )
+        self.flow.on_text(sender_id="u1", text="确定", reply=self._reply)
+        self.replies.clear()
+        phase = self.flow.on_button(
+            sender_id="u1",
+            action=copy.BTN_PULL_COLLAB,
+            reply=self._reply,
+        )
+        self.assertEqual(self.groups.created, [])
+        self.assertIn("拉群失败，请排查填写内容，并重新发起工单。", "\n".join(self.replies))
+        self.assertNotEqual(phase, "collab_group")
+
     def test_land_role_is_product_in_group(self) -> None:
         from agent.policy.sea_group_ops import is_product
 

inquiry-agent/tests/test_mode_vision.py

@@ -0,0 +1,102 @@
+"""
+千问视觉：只抄图上可见原文，不算缺项。
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+import unittest
+
+ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
+if ROOT not in sys.path:
+    sys.path.insert(0, ROOT)
+
+from agent.llm.mode_vision import invoke_shell, invoke_vision
+from agent.llm.modes import LlmMode
+
+
+class ModeVisionTests(unittest.TestCase):
+    def test_shell_reads_assistant_text(self) -> None:
+        got = invoke_shell(payload={"assistant_text": "海运 宁波到汉堡 服装 1x40HQ"})
+        self.assertTrue(got["ok"])
+        self.assertIn("宁波到汉堡", got["text"])
+        self.assertFalse(got["empty"])
+
+    def test_blank_is_empty(self) -> None:
+        got = invoke_shell(payload={"assistant_text": "  "})
+        self.assertTrue(got["ok"])
+        self.assertTrue(got["empty"])
+        self.assertEqual(got["text"], "")
+
+    def test_injected_chat_builds_image_message(self) -> None:
+        seen: dict = {}
+
+        class _R:
+            ok = True
+            content = "起运港上海"
+            error = ""
+
+        def chat(**kwargs):
+            seen.update(kwargs)
+            return _R()
+
+        got = invoke_vision(
+            payload={"image_b64": "abc", "mime": "image/jpeg"}, chat_fn=chat
+        )
+        self.assertEqual(got["text"], "起运港上海")
+        mode = seen.get("mode")
+        mode_val = mode.value if hasattr(mode, "value") else str(mode)
+        self.assertEqual(mode_val, LlmMode.VISION.value)
+        messages = seen.get("messages") or []
+        blob = str(messages)
+        self.assertIn("abc", blob)
+        self.assertIn("image", blob.lower())
+
+    def test_chat_fail(self) -> None:
+        class _R:
+            ok = False
+            content = ""
+            error = "http_timeout"
+
+        got = invoke_vision(
+            payload={"image_b64": "x"}, chat_fn=lambda **k: _R()
+        )
+        self.assertFalse(got["ok"])
+        self.assertEqual(got["error"], "http_timeout")
+
+    def test_sniff_png_not_jpeg(self) -> None:
+        from agent.llm.mode_vision import sniff_image_mime
+
+        png = b"\x89PNG\r\n\x1a\n" + b"\x00" * 16
+        jpg = b"\xff\xd8\xff\xe0" + b"\x00" * 16
+        self.assertEqual(sniff_image_mime(png), "image/png")
+        self.assertEqual(sniff_image_mime(jpg), "image/jpeg")
+        self.assertEqual(sniff_image_mime(b"short"), "")
+
+    def test_network_without_image_fails(self) -> None:
+        got = invoke_vision(payload={}, chat_fn=lambda **k: None, allow_network=True)
+        self.assertFalse(got["ok"])
+        self.assertEqual(got["error"], "no_image")
+
+    def test_plain_text_from_list_and_reasoning(self) -> None:
+        from agent.llm.http_client import message_plain_text
+
+        self.assertEqual(
+            message_plain_text({"content": [{"type": "text", "text": "起运地深圳"}]}),
+            "起运地深圳",
+        )
+        self.assertEqual(
+            message_plain_text({"content": "", "reasoning_content": "目的地雅加达"}),
+            "目的地雅加达",
+        )
+
+    def test_vision_prompt_keeps_iata(self) -> None:
+        from agent.llm.mode_vision import VISION_SYSTEM
+
+        self.assertIn("SZX-CGK", VISION_SYSTEM)
+        self.assertIn("禁止翻译", VISION_SYSTEM)
+
+
+if __name__ == "__main__":
+    unittest.main()

inquiry-agent/tests/test_sea_text_inquiry.py

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

inquiry-agent/tests/test_tms_land_query.py

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

inquiry-api/jeecg-module-inquiry/src/main/java/org/jeecg/modules/inquiry/tms/TmsLandEnumMapper.java

@@ -19,6 +19,8 @@ public final class TmsLandEnumMapper {
         TYPE.put("国内运输整车", "DOMESTIC_FTL");
         TYPE.put("跨境集拼", "CROSS_BORDER_LTL");
         TYPE.put("跨境整车", "CROSS_BORDER_FTL");
+        TYPE.put("中港整车", "CHINA_HK");
+        TYPE.put("中港零担/集拼", "CHINA_HK");
         TYPE.put("中港", "CHINA_HK");
         TYPE.put("DOMESTIC_LTL", "DOMESTIC_LTL");
         TYPE.put("DOMESTIC_FTL", "DOMESTIC_FTL");

inquiry-api/jeecg-module-inquiry/src/main/java/org/jeecg/modules/inquiry/tms/TmsLandSegmentMapper.java

@@ -15,9 +15,10 @@ import java.util.regex.Pattern;
 /**
  * 陆运 TMS 查价段：按运输类型拼出站字段。
  * <p>
- * 职责：把工单 facts 里的中文键写成 TMS 段；运输分类、中港装载方式只留在工单，不进出站。
- * 拼车/跨境集拼带体积重量；国内整车/跨境整车带「车型/数量」一项再拆成出站车型与台数；
- * 中港整车不带车型/数量，中港零担/集拼才带。通关口岸有值才放。
+ * 职责：把工单 facts 里的中文键写成 TMS 段；运输分类只留在工单，不进出站。
+ * 拼车/跨境集拼带体积重量；国内整车/跨境整车/中港零担带「车型/数量」一项再拆成出站车型与台数；
+ * 中港整车不带车型/数量。通关口岸有值才放。
+ * 中港整车、中港零担/集拼出站类型都是 CHINA_HK。
  * 类型/线路出站必须是 TMS 6.1.1 英文枚举，线路键名 routeCategory。
  * 禁止：按空运要包装件数；把分类当查价条件；直连 TMS。
  */
@@ -26,9 +27,9 @@ public final class TmsLandSegmentMapper {
     /** 体积+重量那一套。 */
     private static final Set<String> LCL = new HashSet<String>(
             Arrays.asList("国内运输拼车", "跨境集拼"));
-    /** 车型/数量那一套。中港不在此列，按装载方式另判。 */
+    /** 车型/数量那一套。中港零担走这里；中港整车另判。 */
     private static final Set<String> FTL = new HashSet<String>(
-            Arrays.asList("国内运输整车", "跨境整车"));
+            Arrays.asList("国内运输整车", "跨境整车", "中港零担/集拼"));
 
     private static final Pattern VEHICLE_SPLIT = Pattern.compile("\\s*[/xX×*＊]\\s*");
     private static final Pattern TAIL_QTY = Pattern.compile("^(.*?)[\\s,，]+(\\d+)\\s*(台|辆)?$");
@@ -37,7 +38,7 @@ public final class TmsLandSegmentMapper {
     }
 
     /**
-     * @param facts 工单/智能体中文 facts（含运输类型、线路类别、货物、中港装载方式）
+     * @param facts 工单/智能体中文 facts（含运输类型、线路类别、货物）
      * @return 可并入查价 segment 的键；不含 landClass / hkLoadType
      */
     public static Map<String, Object> build(Map<String, Object> facts) {
@@ -45,7 +46,6 @@ public final class TmsLandSegmentMapper {
         Map<String, Object> seg = new LinkedHashMap<String, Object>();
         String type = text(source.get("运输类型"));
         String route = text(source.get("线路类别"));
-        String hkLoad = text(source.get("中港装载方式"));
         put(seg, "landTransportType", TmsLandEnumMapper.transportType(type));
         put(seg, "routeCategory", TmsLandEnumMapper.routeCategory(route));
         put(seg, "origin", text(source.get("起运港")));
@@ -54,9 +54,9 @@ public final class TmsLandSegmentMapper {
         if (LCL.contains(type)) {
             put(seg, "volumeCbm", TmsV2RequestMapper.asPositiveDecimal(source.get("体积")));
             put(seg, "weightKg", TmsV2RequestMapper.asPositiveDecimal(source.get("毛重")));
-        } else if ("中港".equals(type) && !"零担/集拼".equals(hkLoad)) {
+        } else if ("中港整车".equals(type)) {
             // 中港整车：询价不要车型/数量，出站也不带
-        } else if (FTL.contains(type) || "中港".equals(type)) {
+        } else if (FTL.contains(type)) {
             putVehicleQty(source, seg);
         }
         String customs = text(source.get("通关口岸"));
@@ -68,9 +68,9 @@ public final class TmsLandSegmentMapper {
 
     /**
      * 陆运出站拦网：工单号、报价日期、起运目的品名、类型、线路，再加当前类型货物。
-     * 不要包装、不要运输分类、不要中港装载。中港整车不要车型/数量。
+     * 不要包装、不要运输分类。中港整车不要车型/数量。
      *
-     * @param facts 用于读中港装载方式；禁止把装载方式写进 segment
+     * @param facts 用于读运输类型；禁止把分类写进 segment
      */
     public static List<String> missingQuery(String workOrderNo, String quoteDate,
                                             Map<String, Object> segment,
@@ -93,14 +93,14 @@ public final class TmsLandSegmentMapper {
         }
         String typeEnum = text(seg.get("landTransportType"));
         String cnType = text(facts == null ? null : facts.get("运输类型"));
-        String hkLoad = text(facts == null ? null : facts.get("中港装载方式"));
         boolean lcl = LCL.contains(cnType)
                 || "DOMESTIC_LTL".equals(typeEnum)
                 || "CROSS_BORDER_LTL".equals(typeEnum);
-        boolean hk = "中港".equals(cnType) || "CHINA_HK".equals(typeEnum);
+        boolean hkFull = "中港整车".equals(cnType);
         boolean ftl = FTL.contains(cnType)
                 || "DOMESTIC_FTL".equals(typeEnum)
-                || "CROSS_BORDER_FTL".equals(typeEnum);
+                || "CROSS_BORDER_FTL".equals(typeEnum)
+                || "中港零担/集拼".equals(cnType);
         if (lcl) {
             if (seg.get("volumeCbm") == null) {
                 missing.add("volumeCbm");
@@ -108,9 +108,9 @@ public final class TmsLandSegmentMapper {
             if (seg.get("weightKg") == null) {
                 missing.add("weightKg");
             }
-        } else if (hk && !"零担/集拼".equals(hkLoad)) {
+        } else if (hkFull) {
             // 中港整车不拦车型/数量
-        } else if (ftl || hk) {
+        } else if (ftl) {
             if (blank(seg.get("vehicleType"))) {
                 missing.add("vehicleType");
             }

inquiry-api/jeecg-module-inquiry/src/test/java/org/jeecg/modules/inquiry/tms/TmsLandEnumMapperTest.java

@@ -15,6 +15,12 @@ public class TmsLandEnumMapperTest {
         assertEquals("DOMESTIC_CITY_DIST", TmsLandEnumMapper.routeCategory("国内城配/拖车/打包"));
     }
 
+    @Test
+    public void hkSplitTypesBothMapChinaHk() {
+        assertEquals("CHINA_HK", TmsLandEnumMapper.transportType("中港整车"));
+        assertEquals("CHINA_HK", TmsLandEnumMapper.transportType("中港零担/集拼"));
+    }
+
     @Test
     public void alreadyEnumStays() {
         assertEquals("DOMESTIC_LTL", TmsLandEnumMapper.transportType("DOMESTIC_LTL"));

inquiry-api/jeecg-module-inquiry/src/test/java/org/jeecg/modules/inquiry/tms/TmsLandSegmentMapperTest.java

@@ -38,15 +38,15 @@ public class TmsLandSegmentMapperTest {
     @Test
     public void landFtlUsesVehicleNotVolume() {
         Map<String, Object> facts = new HashMap<String, Object>();
-        facts.put("运输类型", "中港");
+        facts.put("运输类型", "中港零担/集拼");
         facts.put("线路类别", "中港/中亚/中欧");
-        facts.put("中港装载方式", "零担/集拼");
         facts.put("起运港", "广州");
         facts.put("目的港", "香港");
         facts.put("品名", "衣服");
         facts.put("车型/数量", "9米6/1");
         facts.put("体积", "99");
         Map<String, Object> seg = TmsLandSegmentMapper.build(facts);
+        assertEquals("CHINA_HK", seg.get("landTransportType"));
         assertEquals("9米6", seg.get("vehicleType"));
         assertNull(seg.get("volumeCbm"));
         assertFalse(seg.containsKey("hkLoadType"));
@@ -56,13 +56,13 @@ public class TmsLandSegmentMapperTest {
     @Test
     public void hkFullTruckOmitsVehicle() {
         Map<String, Object> facts = new HashMap<String, Object>();
-        facts.put("运输类型", "中港");
+        facts.put("运输类型", "中港整车");
         facts.put("线路类别", "中港/中亚/中欧");
-        facts.put("中港装载方式", "整车");
         facts.put("起运港", "广州");
         facts.put("目的港", "香港");
         facts.put("品名", "衣服");
         Map<String, Object> seg = TmsLandSegmentMapper.build(facts);
+        assertEquals("CHINA_HK", seg.get("landTransportType"));
         assertNull(seg.get("vehicleType"));
         assertNull(seg.get("quantity"));
         java.util.List<String> missing = TmsLandSegmentMapper.missingQuery(