fix(message): 发送前等 /s/vulcan 并校验 ack，避免 message send 静默失败

src/goofish_cli/commands/message/send.py

@@ -1,22 +1,31 @@
 """message send — 向指定会话发送一条消息（文本/图片）。
 
 写操作，走限流 + 熔断。未知 cid 时可传 --item-id 自动创建单聊。
+
+发送前必须等 `/s/vulcan`（见 `wait_ready`），并校验握手与发送回包。
+
+`ok` 的含义被严格限定为「服务端已接受该发送请求」（发送 ack `code=200`），
+**不等同于**消息最终落库或对方已收到；后两者只能由 `message history`
+回读或对端确认。拿不到 200 一律报错，不返回假成功。
 """
 
 import asyncio
 from typing import Any, Literal
 
 from goofish_cli.core import Session, Strategy, command
+from goofish_cli.core.errors import GoofishError
 from goofish_cli.core.guard import watch
 from goofish_cli.core.limiter import acquire
 from goofish_cli.core.token import get_access_token
 from goofish_cli.core.ws import (
     connect,
     create_chat,
     heartbeat_loop,
+    recv_ack,
     register,
     send_image,
     send_text,
+    wait_ready,
 )
 
 
@@ -25,7 +34,7 @@
     name="send",
     description="向会话发送消息（text/image）。text 必填，image 走 url+wh",
     strategy=Strategy.COOKIE,
-    columns=["cid", "toid", "kind", "ok", "mid"],
+    columns=["cid", "toid", "kind", "ok", "mid", "message_id"],
     write=True,
 )
 def send(
@@ -68,12 +77,19 @@ async def _send(
 ) -> dict[str, Any]:
     token = get_access_token(session)
     async with connect(session) as ws:
-        await register(ws, session, token)
+        handshake_mids = await register(ws, session, token)
         hb = asyncio.create_task(heartbeat_loop(ws))
         try:
+            # `/r/` 请求必须等 `/s/vulcan` 之后才发，否则服务端回 code 400。
+            # 顺带校验握手回包：/reg 非 200 抛错，ackDiff 非 200 只记警告。
+            if not await wait_ready(ws, mids=handshake_mids):
+                raise GoofishError("IM 连接未就绪（等待 /s/vulcan 超时），未发送")
+
             if item_id:
-                await create_chat(ws, myid=session.unb, toid=toid, item_id=item_id)
-                await asyncio.sleep(0.5)
+                create_mid = await create_chat(
+                    ws, myid=session.unb, toid=toid, item_id=item_id
+                )
+                await recv_ack(ws, create_mid)
 
             if kind == "text":
                 if not text:
@@ -95,8 +111,21 @@ async def _send(
                 )
             else:
                 raise ValueError(f"不支持的 kind: {kind}")
-            # 等一轮 ack 回包，避免 WS 提前关
-            await asyncio.sleep(1.0)
+            ack = await recv_ack(ws, mid)
         finally:
             hb.cancel()
-    return {"cid": cid, "toid": toid, "kind": kind, "ok": True, "mid": mid}
+
+    ack_code = (ack or {}).get("code")
+    if ack_code != 200:
+        detail = "无回包（超时）" if ack is None else f"ack_code={ack_code}"
+        raise GoofishError(f"发送未被服务端接受：{detail}")
+
+    body = (ack or {}).get("body") or {}
+    return {
+        "cid": cid,
+        "toid": toid,
+        "kind": kind,
+        "ok": True,
+        "mid": mid,
+        "message_id": str(body.get("messageId", "")),
+    }

src/goofish_cli/core/ws.py

@@ -24,6 +24,7 @@
 from loguru import logger
 from websockets.asyncio.client import ClientConnection
 
+from goofish_cli.core.errors import GoofishError
 from goofish_cli.core.session import Session
 from goofish_cli.core.sign import decrypt, generate_mid, generate_uuid
 from goofish_cli.core.token import IM_APP_KEY, get_access_token
@@ -70,8 +71,19 @@ async def connect(session: Session) -> AsyncIterator[ClientConnection]:
         yield ws
 
 
-async def register(ws: ClientConnection, session: Session, token: str) -> None:
-    """/reg + /r/SyncStatus/ackDiff。发完立即返回，后续由外层 recv 循环消化回包。"""
+async def register(
+    ws: ClientConnection, session: Session, token: str
+) -> dict[str, str]:
+    """发送 `/reg` + `/r/SyncStatus/ackDiff`，返回两者的 mid 供调用方校验回包。
+
+    发完立即返回：回包由外层 recv 循环消化。校验在 `wait_ready()` 里按返回的
+    mid 进行，而不是放在这里——`register()` 若自己消费回包，会吃掉
+    `list_user_messages()` 赖以触发请求的 `/s/vulcan` 帧。
+
+    因此目前只有走 `wait_ready()` 的发送路径会校验握手；
+    `list_user_messages()` 与 `run_forever()` 仍是忽略返回值的 fire-and-forget。
+    """
+    reg_mid = generate_mid()
     reg = {
         "lwp": "/reg",
         "headers": {
@@ -83,14 +95,15 @@ async def register(ws: ClientConnection, session: Session, token: str) -> None:
             "wv": "im:3,au:3,sy:6",
             "sync": "0,0;0;0;",
             "did": session.device_id,
-            "mid": generate_mid(),
+            "mid": reg_mid,
         },
     }
     await ws.send(json.dumps(reg))
     current_ms = int(time.time() * 1000)
+    ack_diff_mid = generate_mid()
     ack_diff = {
         "lwp": "/r/SyncStatus/ackDiff",
-        "headers": {"mid": generate_mid()},
+        "headers": {"mid": ack_diff_mid},
         "body": [
             {
                 "pipeline": "sync",
@@ -105,6 +118,7 @@ async def register(ws: ClientConnection, session: Session, token: str) -> None:
         ],
     }
     await ws.send(json.dumps(ack_diff))
+    return {"reg": reg_mid, "ack_diff": ack_diff_mid}
 
 
 async def heartbeat_loop(ws: ClientConnection, interval: float = 15.0) -> None:
@@ -134,6 +148,83 @@ def build_ack(msg: dict[str, Any]) -> dict[str, Any]:
     return ack
 
 
+async def _recv_json(ws: ClientConnection, *, timeout: float) -> dict[str, Any] | None:
+    """收一帧并解析成 dict。超时或非 JSON 返回 None。"""
+    try:
+        raw = await asyncio.wait_for(ws.recv(), timeout=timeout)
+    except TimeoutError:
+        return None
+    try:
+        parsed = json.loads(raw)
+    except (json.JSONDecodeError, TypeError):
+        return None
+    return parsed if isinstance(parsed, dict) else None
+
+
+async def wait_ready(
+    ws: ClientConnection,
+    *,
+    mids: dict[str, str] | None = None,
+    timeout: float = 15.0,
+) -> bool:
+    """等服务端推 `/s/vulcan`，期间对下行帧回 ack。就绪返回 True。
+
+    `/r/` 请求在 `/s/vulcan` 到达之前发出会被服务端以 `code 400` 拒绝。`/reg`
+    自身返回 200，所以「注册成功」并不代表连接已经可以发请求。
+    `list_user_messages()` 一直是等到 `/s/vulcan` 才发 `/r/` 请求。
+
+    传入 `register()` 返回的 mids 时顺带校验握手回包：
+
+    * `/reg` 非 200 直接抛错 —— 注册失败后继续发送没有意义。
+    * `ackDiff` 非 200 只记警告 —— 实测它恒为 400（与 `pts` 取值无关），而
+      `collect_session_cids()` 用同样的 ackDiff 也拿 400 却工作正常，sync
+      下推照常到达，因此不作为致命错误。
+    """
+    reg_mid = (mids or {}).get("reg")
+    ack_diff_mid = (mids or {}).get("ack_diff")
+    loop = asyncio.get_running_loop()
+    deadline = loop.time() + timeout
+    while True:
+        remaining = deadline - loop.time()
+        if remaining <= 0:
+            return False
+        frame = await _recv_json(ws, timeout=min(3.0, remaining))
+        if frame is None:
+            continue
+
+        frame_mid = (frame.get("headers") or {}).get("mid")
+        if reg_mid and frame_mid == reg_mid and frame.get("code") != 200:
+            raise GoofishError(
+                f"IM 注册失败：/reg 返回 code={frame.get('code')}，未发送"
+            )
+        if ack_diff_mid and frame_mid == ack_diff_mid and frame.get("code") != 200:
+            logger.debug("ackDiff rejected with code=%s (non-fatal)", frame.get("code"))
+
+        with suppress(Exception):
+            await ws.send(json.dumps(build_ack(frame)))
+        if frame.get("lwp") == "/s/vulcan":
+            return True
+
+
+async def recv_ack(
+    ws: ClientConnection, mid: str, *, timeout: float = 10.0
+) -> dict[str, Any] | None:
+    """读取 `mid` 对应的响应帧，期间继续对下行推送回 ack。超时返回 None。"""
+    loop = asyncio.get_running_loop()
+    deadline = loop.time() + timeout
+    while True:
+        remaining = deadline - loop.time()
+        if remaining <= 0:
+            return None
+        frame = await _recv_json(ws, timeout=min(3.0, remaining))
+        if frame is None:
+            continue
+        if (frame.get("headers") or {}).get("mid") == mid:
+            return frame
+        with suppress(Exception):
+            await ws.send(json.dumps(build_ack(frame)))
+
+
 async def send_text(
     ws: ClientConnection, *, myid: str, cid: str, toid: str, text: str
 ) -> str: