// McpChat.cpp : MCP「远程交谈」工具族 + CMcpServer 聊天会话状态机 // // 链路(Windows 客户端,零改动): // 主连接 [COMMAND_CHAT=161] // → 客户端新建子连接:先 [TOKEN_CONN_AUTH=246](子连接鉴权),再 [TOKEN_CHAT_START=62] // → 服务端在子连接回 [COMMAND_NEXT_CHAT=0] → 客户端弹出 IDD_CHAT // → 双向文本 [CP936 字节][NUL],无 token 前缀 // → 服务端 [COMMAND_CHAT_CLOSE=1] 关闭窗口 // // 客户端怪癖(见 docs/Mcp_Chat_Design.md §4): // 1. ChatManager.cpp 在任何长度判断前丢弃首字节为 TOKEN_HEARTBEAT(101, 'e') 的包, // 故 GBK 首字节为 101 时前置一个空格。 // 2. 窗口由子线程 CreateDialogParamA 创建,收到 [0] 后仍有约 150ms 建窗延迟; // 绑定后 1000ms 内不返回 chat_open,也不接受 chat_send。 // // 会话状态全部由 CMcpServer 持有(McpServer.h)。 #include "stdafx.h" #include "McpChat.h" #include "McpServer.h" #include "McpInternal.h" #include "context.h" #include "2015RemoteDlg.h" #include "LangManager.h" #include #include #include using namespace McpInternal; namespace { // ========== 错误码(与 McpServer.cpp / McpMachine.cpp 一致) ========== const int kErrTimeout = -32001; const int kErrOffline = -32002; const int kErrBusy = -32003; const int kErrSend = -32004; const int kErrUnsupported = -32005; const int kErrDisabled = -32006; const int kErrInvalid = -32602; // ========== 参数/协议上限 ========== // 客户端输入缓冲 char str[4096](ChatManager.cpp),最大 4095 字节; // 留出尾部 NUL 与余量,服务端上限 4000 GBK 字节。 const size_t kMaxSendBytes = 4000; // 单条入站文本上限:正常聊天远小于此,超过即丢弃(防异常客户端把 inbox 撑爆)。 const size_t kMaxIncomingBytes = 8u * 1024u; // inbox 上限(条数 / UTF-8 字节数),溢出丢最旧并计数。 const size_t kInboxMaxMessages = 100; const size_t kInboxMaxBytes = 256u * 1024u; // 绑定后等待客户端建窗的时间(ChatManager::WaitForDialogOpen 为 8s 等事件 + 150ms Sleep; // 服务端毫秒级回 [0],真正需要覆盖的是那 150ms 的建窗窗口)。 const int kChatOpenSettleMs = 1000; // 迟到 62 墓碑:吸收超时/关闭后才到达的子连接。 const int kChatOrphanMs = 5000; const int kOpenDefaultMs = 15000; const int kOpenMinMs = 2000; const int kOpenMaxMs = 60000; const int kWaitDefaultMs = 30000; const int kWaitMaxMs = 55000; const int kWaitMaxMessages = 50; const int kWaitDefaultMessages = 20; // ========== 小工具 ========== std::string Trim(const std::string& s) { size_t b = 0, e = s.size(); while (b < e && (s[b] == ' ' || s[b] == '\t' || s[b] == '\r' || s[b] == '\n')) ++b; while (e > b && (s[e - 1] == ' ' || s[e - 1] == '\t' || s[e - 1] == '\r' || s[e - 1] == '\n')) --e; return s.substr(b, e - b); } std::string OptStr(const Json::Value& args, const char* key) { return Trim(GetStringArg(args, key)); } bool IsWindowsClient(context* ctx) { if (!ctx) return false; CString t = ctx->GetAdditionalData(RES_CLIENT_TYPE); return !(t == "LNX" || t == "MAC" || t == "APK"); } struct HostTool { uint64_t dev = 0; context* mainCtx = nullptr; }; bool ResolveHost(const Json::Value& id, const Json::Value& args, CMy2015RemoteDlg* parent, HostTool& ht, bool requireWindows_, std::string& errJson) { uint64_t dev = 0; std::string err; if (!ParseHostIdArg(args, dev, err)) { errJson = BuildError(id, kErrInvalid, err); return false; } context* ctx = FindMainContext(parent, dev); if (!ctx) { errJson = BuildError(id, kErrOffline, "Host not found or offline: " + std::to_string(dev)); return false; } if (requireWindows_ && !IsWindowsClient(ctx)) { errJson = BuildError(id, kErrUnsupported, "Remote chat is only supported on Windows hosts"); return false; } ht.dev = dev; ht.mainCtx = ctx; return true; } bool GetOptIntArg(const Json::Value& id, const Json::Value& args, const char* key, int minV, int maxV, int& out, std::string& errJson) { if (!args.isMember(key)) return true; int v = 0; if (!GetIntArg(args, key, v)) { errJson = BuildError(id, kErrInvalid, std::string("Invalid parameter '") + key + "': expected an integer"); return false; } if (v < minV || v > maxV) { errJson = BuildError(id, kErrInvalid, std::string("Parameter '") + key + "' out of range [" + std::to_string(minV) + ", " + std::to_string(maxV) + "]"); return false; } out = v; return true; } // ========== schema 辅助 ========== Json::Value PropStr(const char* desc = nullptr) { Json::Value p(Json::objectValue); p["type"] = "string"; if (desc) p["description"] = desc; return p; } Json::Value PropInt(const char* desc = nullptr) { Json::Value p(Json::objectValue); p["type"] = "integer"; if (desc) p["description"] = desc; return p; } Json::Value PropBool(const char* desc = nullptr) { Json::Value p(Json::objectValue); p["type"] = "boolean"; if (desc) p["description"] = desc; return p; } Json::Value MakeSchema(const Json::Value& props, std::initializer_list required) { Json::Value schema(Json::objectValue); schema["type"] = "object"; schema["properties"] = props; if (required.size() > 0) { Json::Value req(Json::arrayValue); for (const char* r : required) req.append(r); schema["required"] = req; } return schema; } Json::Value OutputSchema(Json::Value props, std::initializer_list required) { return MakeSchema(props, required); } Json::Value HostIdProp() { return PropStr(u8"主机 id,取 list_online_hosts / search_hosts 返回的 id 字段"); } Json::Value SessionIdProp() { return PropStr(u8"chat_open 返回的 session_id(32 位 hex)"); } void AddTool(Json::Value& tools, const char* name, const char* desc, const Json::Value& inputSchema, const Json::Value& outputSchema) { Json::Value tool(Json::objectValue); tool["name"] = name; tool["description"] = desc; // 调用方传 u8 字面量,保证 JSON 里是 UTF-8 tool["inputSchema"] = inputSchema; tool["outputSchema"] = outputSchema; tools.append(tool); } std::string BuildStructuredResult(const Json::Value& id, const Json::Value& sc, const std::string& text) { Json::Value result(Json::objectValue); result["structuredContent"] = sc; Json::Value content(Json::arrayValue); Json::Value item(Json::objectValue); item["type"] = "text"; item["text"] = text; content.append(item); result["content"] = content; result["isError"] = false; return BuildResult(id, result); } // ========== 入参校验 ========== // 纯空白(space/tab/CR/LF);控制字符由调用方另行拒绝。 bool IsBlankMessage(const std::string& s) { return Trim(s).empty(); } // 拒绝 NUL、CR/LF 与所有 C0 控制字符。UTF-8 多字节字节均 >= 0x80,不会误判。 bool HasControlChar(const std::string& s) { for (unsigned char c : s) { if (c < 0x20 || c == 0x7F) return true; } return false; } } // namespace namespace McpChat { bool Handles(const std::string& toolName) { static const char* kTools[] = { "chat_open", "chat_send", "chat_wait", "chat_close", }; for (const char* t : kTools) { if (toolName == t) return true; } return false; } void AppendTools(Json::Value& tools) { CMcpServer& mcp = McpServer(); // 独立开关(默认关):与 McpReadonly 无关,关闭时工具完全不可见。 if (!mcp.IsChatEnabled()) return; // ---- chat_open ---- { Json::Value props(Json::objectValue); props["id"] = HostIdProp(); props["timeout_ms"] = PropInt(u8"等待客户端弹出聊天窗口的超时,默认 15000,范围 2000-60000"); AddTool(tools, "chat_open", u8"在指定在线 Windows 主机上弹出「远程交谈」窗口,建立一次多轮文字对话。返回 session_id,后续 chat_send / chat_wait / chat_close 都要带上。客户端使用者会直接看到聊天窗口(无确认框)。同一主机同一时刻只允许一个会话;会话在客户端断开或空闲 10 分钟后自动失效。会话建立后应先用 chat_send 发送问候,再用 chat_wait 等待对方回复。", MakeSchema(props, { "id" }), OutputSchema([] { Json::Value p(Json::objectValue); p["session_id"] = PropStr(); p["status"] = PropStr(u8"固定为 open"); p["message"] = PropStr(); return p; }(), { "session_id", "status" })); } // ---- chat_send ---- { Json::Value props(Json::objectValue); props["id"] = HostIdProp(); props["session_id"] = SessionIdProp(); props["message"] = PropStr(u8"要发给客户端使用者的纯文本(UTF-8)。不能为空、不能含换行/控制字符、不能超过 4000 个 GBK 字节(约 2000 个汉字);emoji 等 GBK 无法表示的字符会被拒绝"); AddTool(tools, "chat_send", u8"向会话对端发送一条文本。消息以客户端本地代码页(简体中文为 GBK/CP936)显示;客户端不保证非中文系统能正确显示。发送成功只代表已写入子连接,不代表对方已读。请勿在同一条消息里塞入多段对话;需要等待回复时调用 chat_wait。", MakeSchema(props, { "id", "session_id", "message" }), OutputSchema([] { Json::Value p(Json::objectValue); p["session_id"] = PropStr(); p["sent"] = PropBool(); p["bytes"] = PropInt(u8"实际下发的 GBK 字节数"); p["message"] = PropStr(); return p; }(), { "session_id", "sent" })); } // ---- chat_wait ---- { Json::Value props(Json::objectValue); props["id"] = HostIdProp(); props["session_id"] = SessionIdProp(); props["timeout_ms"] = PropInt(u8"最长等待时间,默认 30000,范围 0-55000;0 = 立即返回当前已收到的消息"); props["max_messages"] = PropInt(u8"单次最多返回条数,默认 20,范围 1-50;其余留到下一次"); AddTool(tools, "chat_wait", u8"等待客户端使用者发来的消息。有消息立即返回;没有消息则阻塞到 timeout_ms 或会话关闭。返回 messages(按时间顺序,text 为 UTF-8)、timed_out、closed、dropped(因缓冲区溢出丢弃的最旧消息数)。timed_out=true 只表示这段时间对方没说话,会话仍然有效;closed=true 表示客户端已断开,需要重新 chat_open。一般回复后立刻调用 chat_wait 形成对话循环。", MakeSchema(props, { "id", "session_id" }), OutputSchema([] { Json::Value p(Json::objectValue); p["session_id"] = PropStr(); Json::Value arr(Json::objectValue); arr["type"] = "array"; Json::Value items(Json::objectValue); items["type"] = "object"; Json::Value ip(Json::objectValue); ip["id"] = PropInt(u8"会话内单调递增序号"); ip["at"] = PropInt(u8"服务端收到的 Unix 秒"); ip["text"] = PropStr(u8"UTF-8 文本"); items["properties"] = ip; arr["items"] = items; p["messages"] = arr; p["timed_out"] = PropBool(); p["closed"] = PropBool(); p["dropped"] = PropInt(); return p; }(), { "session_id", "messages", "timed_out", "closed" })); } // ---- chat_close ---- { Json::Value props(Json::objectValue); props["id"] = HostIdProp(); props["session_id"] = SessionIdProp(); AddTool(tools, "chat_close", u8"结束会话并关闭客户端上的聊天窗口。主机在线且会话已不存在时按幂等处理为成功;主机已离线时返回 -32002(此时会话已随断线失效)。关闭后旧的 session_id 不能再用;可立即重新 chat_open。", MakeSchema(props, { "id", "session_id" }), OutputSchema([] { Json::Value p(Json::objectValue); p["session_id"] = PropStr(); p["closed"] = PropBool(); p["message"] = PropStr(); return p; }(), { "session_id", "closed" })); } } // ========== 工具实现 ========== static std::string BuildChatOpen(const Json::Value& id, const Json::Value& args, CMy2015RemoteDlg* parent) { HostTool ht; std::string errJson; if (!ResolveHost(id, args, parent, ht, /*requireWindows_=*/true, errJson)) return errJson; int timeoutMs = kOpenDefaultMs; if (!GetOptIntArg(id, args, "timeout_ms", kOpenMinMs, kOpenMaxMs, timeoutMs, errJson)) return errJson; CMcpServer& mcp = McpServer(); mcp.SweepIdleChats(600); const std::string sessionId = GenerateRandomToken(); if (!mcp.BeginChatOpen(ht.dev, ht.mainCtx, sessionId)) { return BuildError(id, kErrBusy, "This host already has an active chat session; close it first " "(chat_close) or wait for it to expire"); } BYTE cmd = (BYTE)COMMAND_CHAT; if (!ht.mainCtx->Send2Client(&cmd, 1)) { context* sub = nullptr; bool had = false; mcp.AbortChatOpen(ht.dev, sub, had); if (sub) sub->CancelIO(); return BuildError(id, kErrSend, "Failed to send the chat command to the host"); } CMcpServer::ChatReadyResult r = mcp.WaitChatReady(ht.dev, sessionId, timeoutMs); if (r != CMcpServer::ChatReadyResult::Ok) { context* sub = nullptr; bool had = false; mcp.AbortChatOpen(ht.dev, sub, had); if (sub) sub->CancelIO(); switch (r) { case CMcpServer::ChatReadyResult::Closed: return BuildError(id, kErrOffline, "The client closed the chat sub-connection during setup"); case CMcpServer::ChatReadyResult::NotFound: return BuildError(id, kErrOffline, "The chat session disappeared during setup"); case CMcpServer::ChatReadyResult::Timeout: default: return BuildError(id, kErrTimeout, "Timeout waiting for the client to open the chat window " "(the host may be offline, the client may be too old, or it may " "not be running in an interactive desktop session)"); } } AuditChatAction(parent, ht.dev, "chat_open", "session=" + sessionId); Json::Value sc(Json::objectValue); sc["session_id"] = sessionId; sc["status"] = "open"; sc["message"] = u8"聊天窗口已打开,可以开始发送消息。"; std::string text = u8"已与主机 " + std::to_string(ht.dev) + u8" 建立远程交谈会话;客户端上的聊天窗口已弹出。session_id=" + sessionId; return BuildStructuredResult(id, sc, text); } static std::string BuildChatSend(const Json::Value& id, const Json::Value& args, CMy2015RemoteDlg* parent) { HostTool ht; std::string errJson; if (!ResolveHost(id, args, parent, ht, /*requireWindows_=*/true, errJson)) return errJson; const std::string sessionId = OptStr(args, "session_id"); if (sessionId.empty()) return BuildError(id, kErrInvalid, "Missing required parameter: session_id"); // 注意:message 不做 Trim(首尾空格是用户输入的一部分),只做空白判定。 const std::string message = GetStringArg(args, "message"); if (message.empty()) return BuildError(id, kErrInvalid, "Parameter 'message' must not be empty"); if (IsBlankMessage(message)) return BuildError(id, kErrInvalid, "Parameter 'message' must not be blank"); if (HasControlChar(message)) return BuildError(id, kErrInvalid, "Parameter 'message' must not contain control characters " "(newlines, tabs, NUL, ... )"); if (!IsValidUtf8(message.c_str(), message.size())) return BuildError(id, kErrInvalid, "Parameter 'message' is not valid UTF-8"); // 严格转 CP936:emoji / GBK 外字符直接拒绝,绝不静默变成 '?'。 std::string gbk; if (!Utf8ToCpExact(message, 936, gbk) || gbk.empty()) { return BuildError(id, kErrInvalid, "Parameter 'message' contains characters that cannot be represented " "in GBK (the client's local code page); emoji are not supported"); } if (gbk.size() > kMaxSendBytes) { return BuildError(id, kErrInvalid, "Parameter 'message' is too long: " + std::to_string(gbk.size()) + " GBK bytes (limit " + std::to_string(kMaxSendBytes) + ")"); } // 客户端在长度判断之前丢弃首字节为 TOKEN_HEARTBEAT(101, 'e') 的包。 // 前置一个空格规避,显示上无感(2026-10 客户端 ChatManager.cpp:281)。 if ((BYTE)gbk[0] == (BYTE)TOKEN_HEARTBEAT) gbk.insert(gbk.begin(), ' '); std::vector packet(gbk.begin(), gbk.end()); packet.push_back('\0'); CMcpServer& mcp = McpServer(); context* sub = nullptr; int br = mcp.BeginChatSend(ht.dev, sessionId, sub); if (br == 1) { return BuildError(id, kErrOffline, "Chat session not found, not ready, or already closed; " "call chat_open again if needed"); } if (br == 2) { return BuildError(id, kErrBusy, "Another message is still being sent; retry shortly"); } bool ok = sub && sub->Send2Client((PBYTE)packet.data(), (ULONG)packet.size()); if (ok) sub->WaitSendIdle(2000); mcp.FinishChatSend(ht.dev, sessionId); if (!ok) return BuildError(id, kErrSend, "Failed to send the message to the client"); AuditChatAction(parent, ht.dev, "chat_send", "session=" + sessionId + " bytes=" + std::to_string(gbk.size())); Json::Value sc(Json::objectValue); sc["session_id"] = sessionId; sc["sent"] = true; sc["bytes"] = (Json::Int64)gbk.size(); sc["message"] = u8"消息已发送(客户端无回执,不代表对方已读)。"; std::string text = u8"已发送 " + std::to_string(gbk.size()) + u8" GBK 字节。"; return BuildStructuredResult(id, sc, text); } static std::string BuildChatWait(const Json::Value& id, const Json::Value& args, CMy2015RemoteDlg* parent) { HostTool ht; std::string errJson; if (!ResolveHost(id, args, parent, ht, /*requireWindows_=*/true, errJson)) return errJson; const std::string sessionId = OptStr(args, "session_id"); if (sessionId.empty()) return BuildError(id, kErrInvalid, "Missing required parameter: session_id"); int timeoutMs = kWaitDefaultMs; if (!GetOptIntArg(id, args, "timeout_ms", 0, kWaitMaxMs, timeoutMs, errJson)) return errJson; int maxMessages = kWaitDefaultMessages; if (!GetOptIntArg(id, args, "max_messages", 1, kWaitMaxMessages, maxMessages, errJson)) return errJson; CMcpServer& mcp = McpServer(); mcp.SweepIdleChats(600); std::vector msgs; bool timedOut = false, closed = false; int dropped = 0; CMcpServer::ChatWaitResult r = mcp.WaitChatMessages( ht.dev, sessionId, timeoutMs, (size_t)maxMessages, msgs, timedOut, closed, dropped); switch (r) { case CMcpServer::ChatWaitResult::NotFound: return BuildError(id, kErrOffline, "Chat session not found; it may have been closed or expired. " "Call chat_open to start a new one"); case CMcpServer::ChatWaitResult::Mismatch: return BuildError(id, kErrOffline, "session_id does not match the active chat session for this host"); case CMcpServer::ChatWaitResult::Busy: return BuildError(id, kErrBusy, "Another chat_wait is already pending for this session"); case CMcpServer::ChatWaitResult::Closed: case CMcpServer::ChatWaitResult::Timeout: case CMcpServer::ChatWaitResult::Ok: default: break; } Json::Value arr(Json::arrayValue); for (const ChatMessage& m : msgs) { Json::Value item(Json::objectValue); item["id"] = (Json::UInt64)m.id; item["at"] = (Json::Int64)m.at; item["text"] = m.text; arr.append(item); } Json::Value sc(Json::objectValue); sc["session_id"] = sessionId; sc["messages"] = arr; sc["timed_out"] = timedOut; sc["closed"] = closed; sc["dropped"] = dropped; std::string text; if (!msgs.empty()) { text = u8"收到 " + std::to_string(msgs.size()) + u8" 条消息"; if (timedOut) text += u8"(等待超时,但已有消息返回)"; } else if (closed) { text = u8"会话已关闭:客户端断开或窗口已结束。"; } else { text = u8"等待超时:对方在这段时间内没有发言,会话仍然有效。"; } if (dropped > 0) text += u8"(缓冲区溢出丢弃了 " + std::to_string(dropped) + u8" 条最旧消息)"; return BuildStructuredResult(id, sc, text); } static std::string BuildChatClose(const Json::Value& id, const Json::Value& args, CMy2015RemoteDlg* parent) { HostTool ht; std::string errJson; if (!ResolveHost(id, args, parent, ht, /*requireWindows_=*/true, errJson)) return errJson; const std::string sessionId = OptStr(args, "session_id"); if (sessionId.empty()) return BuildError(id, kErrInvalid, "Missing required parameter: session_id"); CMcpServer& mcp = McpServer(); context* sub = nullptr; bool hadSession = false; int cr = mcp.CloseChatSession(ht.dev, sessionId, sub, hadSession); if (cr == 2) { return BuildError(id, kErrOffline, "session_id does not match the active chat session for this host"); } if (cr == 0 && sub) { // 先通知客户端销毁窗口,再断子连接;客户端 ChatManager 析构时也会销毁窗口作兜底。 BYTE closeCmd = (BYTE)COMMAND_CHAT_CLOSE; sub->Send2Client(&closeCmd, 1); sub->WaitSendIdle(1000); sub->CancelIO(); } if (cr == 0) AuditChatAction(parent, ht.dev, "chat_close", "session=" + sessionId); Json::Value sc(Json::objectValue); sc["session_id"] = sessionId; sc["closed"] = true; sc["message"] = hadSession ? u8"会话已关闭。" : u8"会话不存在,按幂等处理为已关闭。"; std::string text = hadSession ? u8"远程交谈已结束。" : u8"会话不存在,已是关闭状态。"; return BuildStructuredResult(id, sc, text); } std::string Call(const Json::Value& id, const std::string& toolName, const Json::Value& args, CMy2015RemoteDlg* parent) { CMcpServer& mcp = McpServer(); if (!mcp.IsChatEnabled()) { return BuildError(id, kErrDisabled, "Remote chat tools are disabled (set McpChat=1 to enable)"); } if (!args.isObject()) { return BuildError(id, kErrInvalid, "Invalid params: arguments must be an object"); } if (toolName == "chat_open") return BuildChatOpen(id, args, parent); if (toolName == "chat_send") return BuildChatSend(id, args, parent); if (toolName == "chat_wait") return BuildChatWait(id, args, parent); if (toolName == "chat_close") return BuildChatClose(id, args, parent); return BuildError(id, -32601, "Unknown tool: " + toolName); } } // namespace McpChat // ============================================================================ // CMcpServer:聊天会话状态机(UI 线程与工具线程共享) // ============================================================================ using std::chrono::steady_clock; void CMcpServer::RefreshChatRoutesLocked() { m_ChatRoutes.store((int)(m_ChatContextToDevice.size() + m_ChatAbsorbedSubCtx.size()), std::memory_order_relaxed); } void CMcpServer::PruneChatOrphansLocked(steady_clock::time_point now) { for (auto it = m_ChatOrphanByDevice.begin(); it != m_ChatOrphanByDevice.end(); ) { if (it->second.expiry <= now) it = m_ChatOrphanByDevice.erase(it); else ++it; } for (auto it = m_ChatOrphanByPeer.begin(); it != m_ChatOrphanByPeer.end(); ) { if (it->second.expiry <= now) it = m_ChatOrphanByPeer.erase(it); else ++it; } } bool CMcpServer::OnChatSubPacket(context* subCtx, const BYTE* data, ULONG len) { if (!subCtx || !data || len == 0) return false; const BYTE token = data[0]; const bool readyToken = (token == (BYTE)TOKEN_CHAT_START); // 快速路径:没有已绑定/已吸收的子连接,且不是就绪包 → 与聊天无关。 // MessageHandle 每个包都会调用本函数(含音频/屏幕等高频子连接),不能逐包抢锁。 // 注意:未绑定的 TOKEN_CONN_AUTH(246) 也走这里返回 false,交给鉴权分支;已绑定的 // 聊天文本首字节可能是 0xF6(GBK 前导字节),必须由情形 1 消费,不能提前放行。 if (!readyToken && m_ChatRoutes.load(std::memory_order_relaxed) == 0) return false; // FindMainContext 要取 parent->m_cs,必须在持 m_ChatMutex 之前完成, // 维持锁序 m_cs → m_ChatMutex(与 P2/P7/P8 的调用顺序一致)。 const uint64_t cid = subCtx->GetClientID(); const bool mainIsSelf = readyToken && cid != 0 && FindMainContext(m_parent, cid) == subCtx; const std::string peer = subCtx->GetPeerName(); std::unique_lock lk(m_ChatMutex); const steady_clock::time_point now = steady_clock::now(); PruneChatOrphansLocked(now); // ---- 情形 1:已绑定的子连接 → 该连接上的包全部由 MCP 消费 ---- auto bindIt = m_ChatContextToDevice.find(subCtx); if (bindIt != m_ChatContextToDevice.end()) { const uint64_t dev = bindIt->second; auto sit = m_ChatSessions.find(dev); if (sit == m_ChatSessions.end() || sit->second->subCtx != subCtx || sit->second->closed) { // 会话已结束但包还在路上:吸收并断开 m_ChatContextToDevice.erase(bindIt); m_ChatAbsorbedSubCtx.insert(subCtx); RefreshChatRoutesLocked(); lk.unlock(); subCtx->CancelIO(); return true; } ChatSessionPtr s = sit->second; s->lastActiveAt = time(nullptr); // 单字节包是客户端心跳/控制包,不是文本(客户端发送文本时至少带尾部 NUL)。 if (len <= 1) return true; if ((size_t)len > kMaxIncomingBytes) { ++s->dropped; Mprintf("[MCP-Chat] drop oversized inbound packet dev=%llu len=%lu\n", (unsigned long long)dev, (unsigned long)len); return true; } const size_t textLen = BoundedStrlen((const char*)data, (size_t)len); if (textLen == 0) return true; const std::string raw((const char*)data, textLen); ChatMessage msg; msg.id = s->nextMsgId++; msg.at = (int64_t)time(nullptr); msg.text = ToUtf8(raw.c_str(), 936); if (msg.text.empty()) return true; s->inbox.push_back(msg); s->inboxBytes += msg.text.size(); while (s->inbox.size() > kInboxMaxMessages || s->inboxBytes > kInboxMaxBytes) { s->inboxBytes -= s->inbox.front().text.size(); s->inbox.pop_front(); ++s->dropped; } m_ChatCv.notify_all(); return true; } // ---- 情形 2:已吸收(CancelIO 后仍到达的尾包)→ 丢弃 ---- if (m_ChatAbsorbedSubCtx.find(subCtx) != m_ChatAbsorbedSubCtx.end()) { lk.unlock(); subCtx->CancelIO(); return true; } // ---- 情形 3:未绑定且不是就绪包 → 交回 MFC ---- if (!readyToken) return false; if (mainIsSelf) return false; // 主连接不会发就绪包 // 3a. 按 clientID 精确匹配(经 TOKEN_CONN_AUTH 已钉 ID),优先匹配「尚未绑定」的会话。 uint64_t boundDev = 0; if (cid != 0) { auto it = m_ChatSessions.find(cid); if (it != m_ChatSessions.end()) { const ChatSessionPtr& s = it->second; if (!s->closed && !s->started && s->subCtx == nullptr) boundDev = cid; } } // 3b. 老客户端子连接 clientID 可能为 0:按 peer 兜底,仅在同 peer 唯一候选时绑定。 if (boundDev == 0 && cid == 0 && !peer.empty()) { uint64_t candidate = 0; int matches = 0; for (auto& kv : m_ChatSessions) { const ChatSessionPtr& s = kv.second; if (s->closed || s->started || s->subCtx != nullptr) continue; if (s->peer == peer) { candidate = kv.first; ++matches; } } if (matches == 1) boundDev = candidate; } // 3c. 已有 started 会话(重复子连接):吸收,避免弹出第二个 MFC 聊天窗口。 bool absorbDuplicate = false; if (boundDev == 0) { if (cid != 0) { auto it = m_ChatSessions.find(cid); if (it != m_ChatSessions.end() && it->second->started && !it->second->closed) absorbDuplicate = true; } else if (!peer.empty()) { for (auto& kv : m_ChatSessions) { if (kv.second->started && !kv.second->closed && kv.second->peer == peer) { absorbDuplicate = true; break; } } } } // 3d. 无等待中的会话:5s 墓碑内吸收迟到的 62(chat_open 超时/关闭后的尾巴), // 避免弹出 MFC 对话框。 bool absorbOrphan = false; if (boundDev == 0 && !absorbDuplicate) { if (cid != 0) { auto o = m_ChatOrphanByDevice.find(cid); if (o != m_ChatOrphanByDevice.end() && o->second.expiry > now) absorbOrphan = true; } else if (!peer.empty()) { auto o = m_ChatOrphanByPeer.find(peer); if (o != m_ChatOrphanByPeer.end() && o->second.expiry > now) absorbOrphan = true; } } if (boundDev == 0 && (absorbDuplicate || absorbOrphan)) { m_ChatAbsorbedSubCtx.insert(subCtx); RefreshChatRoutesLocked(); lk.unlock(); Mprintf("[MCP-Chat] absorb late/duplicate TOKEN_CHAT_START peer=%s cid=%llu dup=%d orphan=%d\n", peer.c_str(), (unsigned long long)cid, absorbDuplicate ? 1 : 0, absorbOrphan ? 1 : 0); subCtx->CancelIO(); return true; } if (boundDev == 0) return false; // 用户手动「远程交谈」→ 保持 MFC 行为 auto sit = m_ChatSessions.find(boundDev); if (sit == m_ChatSessions.end()) return false; ChatSessionPtr s = sit->second; // ---- 绑定:回 COMMAND_NEXT_CHAT(0),客户端据此弹窗 ---- s->subCtx = subCtx; s->started = true; s->lastActiveAt = time(nullptr); s->readyAt = now + std::chrono::milliseconds(kChatOpenSettleMs); m_ChatContextToDevice[subCtx] = boundDev; RefreshChatRoutesLocked(); lk.unlock(); BYTE next = (BYTE)COMMAND_NEXT_CHAT; subCtx->Send2Client(&next, 1); subCtx->WaitSendIdle(1000); m_ChatCv.notify_all(); Mprintf("[MCP-Chat] bind dev=%llu sub=%p peer=%s\n", (unsigned long long)boundDev, (void*)subCtx, peer.c_str()); return true; } void CMcpServer::OnChatContextClosed(context* ctx) { if (!ctx) return; std::vector toCancel; { std::lock_guard lk(m_ChatMutex); m_ChatAbsorbedSubCtx.erase(ctx); // 子连接断开:会话标记 closed、清路由,保留在表中让 chat_wait 报告 closed。 auto it = m_ChatContextToDevice.find(ctx); if (it != m_ChatContextToDevice.end()) { const uint64_t dev = it->second; m_ChatContextToDevice.erase(it); auto sit = m_ChatSessions.find(dev); if (sit != m_ChatSessions.end() && sit->second->subCtx == ctx) { sit->second->subCtx = nullptr; sit->second->closed = true; } } // 主连接断开:关闭该主连接上的所有会话,并锁外断其子连接。 for (auto& kv : m_ChatSessions) { ChatSessionPtr s = kv.second; if (s->mainCtx != ctx || s->closed) continue; s->closed = true; s->mainCtx = nullptr; if (s->subCtx) { context* sub = s->subCtx; s->subCtx = nullptr; m_ChatContextToDevice.erase(sub); m_ChatAbsorbedSubCtx.insert(sub); toCancel.push_back(sub); } } RefreshChatRoutesLocked(); m_ChatCv.notify_all(); } for (context* c : toCancel) c->CancelIO(); } bool CMcpServer::BeginChatOpen(uint64_t device_id, context* mainCtx, const std::string& sessionId) { if (!mainCtx) return false; context* staleSub = nullptr; { std::lock_guard lk(m_ChatMutex); auto it = m_ChatSessions.find(device_id); if (it != m_ChatSessions.end()) { if (!it->second->closed) return false; // 同设备已有活跃会话 // 清理已关闭的旧会话(路由/吸收表可能还留着指针) staleSub = it->second->subCtx; if (staleSub) { m_ChatContextToDevice.erase(staleSub); m_ChatAbsorbedSubCtx.insert(staleSub); } m_ChatSessions.erase(it); } m_ChatOrphanByDevice.erase(device_id); ChatSessionPtr s = std::make_shared(); s->sessionId = sessionId; s->mainCtx = mainCtx; s->mainClientId = device_id; s->peer = mainCtx->GetPeerName(); s->lastActiveAt = time(nullptr); m_ChatSessions[device_id] = s; RefreshChatRoutesLocked(); } if (staleSub) staleSub->CancelIO(); return true; } CMcpServer::ChatReadyResult CMcpServer::WaitChatReady(uint64_t device_id, const std::string& sessionId, int timeoutMs) { std::unique_lock lk(m_ChatMutex); auto it = m_ChatSessions.find(device_id); if (it == m_ChatSessions.end()) return ChatReadyResult::NotFound; ChatSessionPtr s = it->second; if (s->sessionId != sessionId) return ChatReadyResult::NotFound; const steady_clock::time_point deadline = steady_clock::now() + std::chrono::milliseconds(timeoutMs); while (true) { const steady_clock::time_point now = steady_clock::now(); if (s->closed) return ChatReadyResult::Closed; if (s->started && now >= s->readyAt) return ChatReadyResult::Ok; if (now >= deadline) return ChatReadyResult::Timeout; // 绑定通知到达后只需等到 readyAt,不要一直睡到整个 open 超时。 steady_clock::time_point wake = deadline; if (s->started && s->readyAt < wake) wake = s->readyAt; m_ChatCv.wait_until(lk, wake); } } int CMcpServer::BeginChatSend(uint64_t device_id, const std::string& sessionId, context*& subCtx) { subCtx = nullptr; std::lock_guard lk(m_ChatMutex); auto it = m_ChatSessions.find(device_id); if (it == m_ChatSessions.end()) return 1; ChatSessionPtr s = it->second; if (s->sessionId != sessionId) return 1; if (s->closed || !s->started || s->subCtx == nullptr) return 1; if (steady_clock::now() < s->readyAt) return 1; // 客户端窗口尚未创建完成 if (s->sendBusy) return 2; s->sendBusy = true; s->lastActiveAt = time(nullptr); subCtx = s->subCtx; return 0; } void CMcpServer::FinishChatSend(uint64_t device_id, const std::string& sessionId) { std::lock_guard lk(m_ChatMutex); auto it = m_ChatSessions.find(device_id); if (it == m_ChatSessions.end()) return; if (it->second->sessionId != sessionId) return; // 旧会话已被新会话替换 it->second->sendBusy = false; } CMcpServer::ChatWaitResult CMcpServer::WaitChatMessages( uint64_t device_id, const std::string& sessionId, int timeoutMs, size_t maxMessages, std::vector& out, bool& timedOut, bool& closed, int& dropped) { out.clear(); timedOut = false; closed = false; dropped = 0; std::unique_lock lk(m_ChatMutex); auto it = m_ChatSessions.find(device_id); if (it == m_ChatSessions.end()) return ChatWaitResult::NotFound; ChatSessionPtr s = it->second; if (s->sessionId != sessionId) return ChatWaitResult::Mismatch; if (s->waitBusy) return ChatWaitResult::Busy; s->waitBusy = true; const steady_clock::time_point deadline = steady_clock::now() + std::chrono::milliseconds(timeoutMs); while (s->inbox.empty() && !s->closed) { if (timeoutMs <= 0 || steady_clock::now() >= deadline) break; m_ChatCv.wait_until(lk, deadline); } while (!s->inbox.empty() && out.size() < maxMessages) { out.push_back(s->inbox.front()); s->inboxBytes -= s->inbox.front().text.size(); s->inbox.pop_front(); } timedOut = s->inbox.empty() && !s->closed; closed = s->closed; dropped = s->dropped; s->dropped = 0; s->waitBusy = false; if (!s->closed) s->lastActiveAt = time(nullptr); return timedOut ? ChatWaitResult::Timeout : ChatWaitResult::Ok; } int CMcpServer::CloseChatSession(uint64_t device_id, const std::string& sessionId, context*& subCtxOut, bool& hadSession) { subCtxOut = nullptr; hadSession = false; std::lock_guard lk(m_ChatMutex); auto it = m_ChatSessions.find(device_id); if (it == m_ChatSessions.end()) return 1; // 幂等 ChatSessionPtr s = it->second; if (s->sessionId != sessionId) return 2; hadSession = true; s->closed = true; s->mainCtx = nullptr; subCtxOut = s->subCtx; if (subCtxOut) { m_ChatContextToDevice.erase(subCtxOut); m_ChatAbsorbedSubCtx.insert(subCtxOut); s->subCtx = nullptr; } m_ChatSessions.erase(it); m_ChatOrphanByDevice.erase(device_id); RefreshChatRoutesLocked(); m_ChatCv.notify_all(); return 0; } void CMcpServer::AbortChatOpen(uint64_t device_id, context*& subCtxOut, bool& hadSession) { subCtxOut = nullptr; hadSession = false; std::lock_guard lk(m_ChatMutex); auto it = m_ChatSessions.find(device_id); if (it == m_ChatSessions.end()) return; hadSession = true; ChatSessionPtr s = it->second; s->closed = true; s->mainCtx = nullptr; subCtxOut = s->subCtx; if (subCtxOut) { m_ChatContextToDevice.erase(subCtxOut); m_ChatAbsorbedSubCtx.insert(subCtxOut); s->subCtx = nullptr; } else { // 子连接还没到:打 5s 墓碑,吸收迟到的 TOKEN_CHAT_START,避免弹出 MFC 对话框。 ChatOrphanMark mk; mk.expiry = steady_clock::now() + std::chrono::milliseconds(kChatOrphanMs); m_ChatOrphanByDevice[device_id] = mk; if (!s->peer.empty()) m_ChatOrphanByPeer[s->peer] = mk; } m_ChatSessions.erase(it); RefreshChatRoutesLocked(); m_ChatCv.notify_all(); } int CMcpServer::SweepIdleChats(time_t idleTimeoutSec) { std::vector toCancel; int swept = 0; { std::lock_guard lk(m_ChatMutex); const time_t now = time(nullptr); for (auto it = m_ChatSessions.begin(); it != m_ChatSessions.end(); ) { ChatSessionPtr s = it->second; if (s->waitBusy || s->sendBusy || now - s->lastActiveAt <= idleTimeoutSec) { ++it; continue; } s->closed = true; s->mainCtx = nullptr; if (s->subCtx) { context* sub = s->subCtx; s->subCtx = nullptr; m_ChatContextToDevice.erase(sub); m_ChatAbsorbedSubCtx.insert(sub); toCancel.push_back(sub); } it = m_ChatSessions.erase(it); ++swept; Mprintf("[MCP-Chat] sweep idle session dev=%llu\n", (unsigned long long)s->mainClientId); } if (swept > 0) { RefreshChatRoutesLocked(); m_ChatCv.notify_all(); } } for (context* c : toCancel) c->CancelIO(); return swept; }