From d38d1af68d53d965b5c66ac3a80cc6b90fb64b87 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=BB=9D=E5=B0=98?= <237809796@qq.com> Date: Mon, 22 Jun 2026 02:45:36 +0800 Subject: [PATCH] =?UTF-8?q?=E9=87=8D=E6=9E=84:=20df-tunnel=20Phase3?= =?UTF-8?q?=E9=98=B6=E6=AE=B51=E7=BA=AF=E9=80=8F=E4=BC=A0(D1=3DA:CommandHa?= =?UTF-8?q?ndler=E6=94=B6Value+payload=E9=80=8F=E4=BC=A0=E4=B8=8D=E8=A7=A3?= =?UTF-8?q?=E6=9E=90TunnelCommand,D5=E4=BF=9D=E7=95=99=E5=BC=B1=E6=A0=A1?= =?UTF-8?q?=E9=AA=8C)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- crates/df-tunnel/src/tunnel.rs | 83 ++++++++++++++++++++-------------- docs/todo.md | 2 +- 2 files changed, 49 insertions(+), 36 deletions(-) diff --git a/crates/df-tunnel/src/tunnel.rs b/crates/df-tunnel/src/tunnel.rs index 5c193d9..52fd30b 100644 --- a/crates/df-tunnel/src/tunnel.rs +++ b/crates/df-tunnel/src/tunnel.rs @@ -9,7 +9,7 @@ //! `{"kind":"control","error":"..."}` + Close,成功则静默进入收发循环(无显式 ack) //! - 握手成功信号:**未收到 error 帧 且 socket 保持打开**(connect 用 timeout 探测首帧) //! - 入站消息:relay 把对端 Miniapp 的文本帧包成 BroadcastMessage 转发,桌面端收到的 -//! 是完整 BroadcastMessage JSON,payload 才是真正的 TunnelCommand +//! 是完整 BroadcastMessage JSON,payload 是 miniapp 指令原样 JSON(Phase3 纯透传,不解析) //! - 出站消息:桌面端发 TunnelEvent 的 JSON(relay 把它当 payload 包成 BroadcastMessage) //! //! ## AiChatEvent 透传原则 @@ -18,7 +18,8 @@ //! //! ## 并发模型 //! 内部 spawn 收发循环 task,通过 mpsc 解耦 send_event(调用方)与 socket 写入。 -//! 收到 TunnelCommand 走 on_command 回调(调用方在 connect 时注册),不阻塞收发循环。 +//! 收到 command 方向 payload(`serde_json::Value`)走 on_command 回调(调用方在 connect 时注册),不阻塞收发循环。 +//! Phase3 协议统一(D1=A):tunnel 纯透传,业务协议解析在 device 端桥接层(src-tauri)。 //! `&self` + `Arc>` 持连接状态,trait 方法全部 async + 不持 stream 借用。 use std::sync::atomic::{AtomicBool, Ordering}; @@ -33,13 +34,15 @@ use tokio_tungstenite::tungstenite::protocol::CloseFrame; use tokio_tungstenite::tungstenite::Message as WsMessage; use crate::error::{Result, TunnelError}; -use crate::events::{Hello, HelloKind, RelayControlError, TunnelCommand, TunnelEvent}; +use crate::events::{Hello, HelloKind, RelayControlError, TunnelEvent}; -/// 收到 TunnelCommand 时的回调类型(Boxed Future,在收发循环 task 内 await) +/// 收到 miniapp→device 指令 payload 时的回调类型(Boxed Future,在收发循环 task 内 await) /// +/// Phase3 协议统一(D1=A):tunnel 纯透传,不再反序列化 TunnelCommand 强类型, +/// payload 以 `serde_json::Value` 上抛,业务协议解析在 device 端桥接层(src-tauri)。 /// 设计为 async 便于调用方执行 Tauri command 异步路由。回调内 panic 会中断收发循环, /// 调用方应自行处理错误(回调返回值忽略,失败由调用方记录)。 -pub type CommandHandler = Arc futures_util::future::BoxFuture<'static, ()> + Send + Sync>; +pub type CommandHandler = Arc futures_util::future::BoxFuture<'static, ()> + Send + Sync>; /// 隧道客户端抽象 /// @@ -58,7 +61,8 @@ pub trait TunnelClient: Send + Sync { /// /// `url` 形如 `wss://host/ws/device`(query 不带 token,token 走 Hello 帧)。 /// 成功后进入「已连接」状态并 spawn 收发循环。失败返回 TunnelError::Connect/Auth。 - /// `on_command` 收到云后端转发的 TunnelCommand 时被调用(在收发 task 内异步执行)。 + /// `on_command` 收到云后端转发的 command 方向 payload Value 时被调用(在收发 task 内异步执行)。 + /// Phase3 纯透传(D1=A):payload 不在 tunnel 解析,由 device 端桥接层 match cmd 路由。 async fn connect( &self, url: &str, @@ -294,7 +298,7 @@ impl WsTunnelClient { /// 收发循环主体(spawn 后独立运行) /// /// 职责: -/// 1. socket 入帧 → 解析 BroadcastMessage → 取 payload 反序列化 TunnelCommand → on_command 回调 +/// 1. socket 入帧 → 解析 BroadcastMessage → 取 payload Value 纯透传 → on_command 回调(协议解析在桥接层) /// 2. mpsc 出帧 → socket 写入(Event 序列化 / Ping 心跳 / Close 关闭) /// 3. 心跳定时器:每 HEARTBEAT_INTERVAL 发一次 Ping /// 4. 任一端断开 → 退出 task,置 connected=false @@ -400,16 +404,12 @@ async fn run_loop( async fn handle_inbound(msg: &WsMessage, on_command: &CommandHandler) -> bool { match msg { WsMessage::Text(text) => { - // relay 转发的是完整 BroadcastMessage JSON - // (含 device_id/kind/source/from/payload/ts,payload 才是 TunnelCommand) - // 先尝试按 BroadcastMessage 镜像解析取 payload;失败则尝试直接解析为 TunnelCommand - // (兼容 relay 未来直发 payload 的变更) - let cmd_opt = parse_command_from_broadcast(text) - .or_else(|| serde_json::from_str::(text).ok()); - - if let Some(cmd) = cmd_opt { - // 回调内执行指令路由(Tauri command);回调失败不影响收发循环 - let fut = on_command(cmd); + // Phase3 纯透传(D1=A):relay 转发完整 BroadcastMessage, + // payload 是 miniapp 指令原样 JSON。tunnel 只提取 payload Value 上抛, + // 不解析业务协议(协议解析在 device 端桥接层 src-tauri)。 + if let Some(payload) = parse_payload_from_broadcast(text) { + // 回调内执行指令路由(桥接层 match cmd → Tauri command);回调失败不影响收发循环 + let fut = on_command(payload); fut.await; } else { // 非 Command 消息(Control/Event 回环/未知):忽略,不中断 @@ -431,12 +431,12 @@ async fn handle_inbound(msg: &WsMessage, on_command: &CommandHandler) -> bool { } } -/// 从 BroadcastMessage JSON 提取 TunnelCommand +/// 从 BroadcastMessage JSON 提取 command 方向的 payload Value(纯透传,不解析业务协议) /// -/// df-relay 入站包成 `BroadcastMessage { ..., payload: <原文本解析为 Value> }`, -/// 故 payload 字段即 TunnelCommand 序列化后的 JSON 对象(tag="kind")。 -/// 此处只镜像解析 payload 字段(避免依赖 df-relay crate),payload 用 Value 透传不耦合。 -fn parse_command_from_broadcast(raw: &str) -> Option { +/// Phase3 协议统一(D1=A):df-relay 入站包成 `BroadcastMessage { ..., payload: Value }`, +/// tunnel 只提取 payload 字段以 Value 上抛 on_command,不反序列化为 TunnelCommand 强类型 +/// (业务协议解析在 device 端桥接层)。仅 kind=="command" 时提取(Event/Control 不提,避免回环噪音)。 +fn parse_payload_from_broadcast(raw: &str) -> Option { // 仅取 payload 字段,避整结构强类型耦合(device_id/source 等字段本客户端不关心) #[derive(serde::Deserialize)] struct BroadcastLike { @@ -446,12 +446,11 @@ fn parse_command_from_broadcast(raw: &str) -> Option { kind: Option, } let parsed: BroadcastLike = serde_json::from_str(raw).ok()?; - // 仅 kind == "command" 时才尝试提 TunnelCommand(Event/Control 不提) + // 仅 kind == "command" 时才提 payload(Event/Control 不提) if parsed.kind.as_deref() != Some("command") { return None; } - let payload = parsed.payload?; - serde_json::from_value::(payload).ok() + parsed.payload } /// 判断是否为 relay 控制面错误帧 @@ -486,10 +485,30 @@ mod tests { } #[test] - fn parse_command_from_broadcast_extracts_payload() { - // 模拟 relay 转发的 BroadcastMessage - let raw = r#"{"device_id":"dev-1","kind":"command","source":{"0":42},"from":"miniapp","payload":{"kind":"send","conv_id":"c1","content":"hi"},"ts":1700000000000}"#; - let cmd = parse_command_from_broadcast(raw).expect("应提取出 TunnelCommand"); + fn parse_payload_from_broadcast_extracts_value() { + // 模拟 relay 转发的 BroadcastMessage(miniapp 发的 MiniCommand {cmd, args} 在 payload) + let raw = r#"{"device_id":"dev-1","kind":"command","source":{"0":42},"from":"miniapp","payload":{"cmd":"send_message","args":{"message":"hi","conversation_id":"c1"}},"ts":1700000000000}"#; + let payload = parse_payload_from_broadcast(raw).expect("应提取出 payload Value"); + // 纯透传:payload 原样上抛,字段保持 miniapp 端 {cmd, args} 结构(未做协议转换) + assert_eq!(payload["cmd"], "send_message"); + assert_eq!(payload["args"]["message"], "hi"); + assert_eq!(payload["args"]["conversation_id"], "c1"); + } + + #[test] + fn parse_payload_ignores_event_kind() { + // Event 方向(device→miniapp)不提取,避免回环噪音 + let raw = r#"{"device_id":"dev-1","kind":"event","source":{"0":1},"from":"device","payload":{"type":"text_delta","conversation_id":"c1","delta":"x"},"ts":1}"#; + assert!(parse_payload_from_broadcast(raw).is_none()); + } + + /// D5 弱校验:payload Value 仍可反序列化为 TunnelCommand(保留强类型作桥接层可选校验) + #[test] + fn payload_compatible_with_tunnel_command() { + use crate::events::TunnelCommand; + let raw = r#"{"device_id":"dev-1","kind":"command","source":{"0":42},"from":"miniapp","payload":{"kind":"send","conv_id":"c1","content":"hi"},"ts":1}"#; + let payload = parse_payload_from_broadcast(raw).expect("应提取 payload"); + let cmd: TunnelCommand = serde_json::from_value(payload).expect("payload 应兼容 TunnelCommand"); match cmd { TunnelCommand::Send { conv_id, content } => { assert_eq!(conv_id, "c1"); @@ -499,12 +518,6 @@ mod tests { } } - #[test] - fn parse_command_ignores_event_kind() { - let raw = r#"{"device_id":"dev-1","kind":"event","source":{"0":1},"from":"device","payload":{"kind":"text_delta","conv_id":"c1","delta":"x"},"ts":1}"#; - assert!(parse_command_from_broadcast(raw).is_none()); - } - #[test] fn parse_relay_error_detects_auth_failure() { let msg = WsMessage::Text(r#"{"kind":"control","error":"auth_failed"}"#.into()); diff --git a/docs/todo.md b/docs/todo.md index 2637008..8f79f0d 100644 --- a/docs/todo.md +++ b/docs/todo.md @@ -756,7 +756,7 @@ **状态**:✅ Phase2 三层全落地(2026-06-22:df-relay `2b8b30e` Hello握手+鉴权+ConnRegistry配对路由+BroadcastMessage透传 / df-tunnel `25d6565` WS客户端出站穿NAT+心跳25s+指数退避+TunnelCommand 5变体 / df-miniapp `280baea` uni-app WS连relay透传+18变体AiChatEvent镜像+useAiChat分派;双 crate 独立消息骨架不依赖 src-tauri/df-types 避跨 crate 强耦合)。📐 **Phase3 联调设计已出** [F-260622-01](../docs/02-架构设计/已编号方案/F-260622-01-跨端AIChat-Phase3联调设计-2026-06-22.md)(协议统一方案A纯透传 + AiSession桥接风险 + 4阶段路线),待实施。 **Phase3 联调待办**(D1-D6 决策点推荐:A纯透传 / 全19变体透传 / EventBus汇聚 / switch不处理 / 强类型保留 / 桥接层R1兜底): -- [ ] F-260622-01-阶段1:tunnel CommandHandler 签名放宽为 `serde_json::Value`(D1定A后)+ parse_command_from_broadcast 改返回 Value +- [x] ✅(2026-06-22)F-260622-01-阶段1(D1=A):tunnel `CommandHandler` 收 `serde_json::Value` + `parse_payload_from_broadcast` 返回 Value(入站纯透传,cargo check + 5 测试过,D5 保留 TunnelCommand 弱校验)。**出站 `send_raw_event` 移阶段2**(无阶段1 验证场景,阶段2 EventBus 接入同步加) - [ ] F-260622-01-阶段2:EventBus 接入 55 处 emit 点(批2)+ tunnel 注册 subscriber + send_raw_event - [ ] F-260622-01-阶段3:src-tauri 桥接模块(MiniCommand→Tauri command 路由 + R1 generating 兜底)+ miniapp 补齐 approve/authorize_dir/continue/stop_loop - [ ] F-260622-01-阶段4:真机联调 + 多会话并发验证矩阵(6 场景含 F-09 跨端并发)