first commit

CMakeLists.txt

@@ -20,8 +20,8 @@ set(CMAKE_RUNTIME_OUTPUT_DIRECTORY ${CMAKE_BINARY_DIR}/bin)
 
 set(EDGE_AGENT_SOURCES
     src/main.cpp
-    src/fake_driver.cpp
+    src/pipeline/EdgeStreamPipeline.cpp
-    src/fake_driver.hpp
+    src/pipeline/EdgeStreamPipeline.h
     src/device_bus/monitor/DeviceEvent.h
     src/device_bus/monitor/IDeviceMonitor.h
     src/device_bus/monitor/UdevDeviceMonitor.cpp
@@ -62,18 +62,42 @@ endif()
 
 qt_add_executable(softbus_edge_agent ${EDGE_AGENT_SOURCES})
 
+set(SOFTBUS_DAEMON_ROOT "${CMAKE_CURRENT_SOURCE_DIR}/../softbus_daemon")
+target_sources(softbus_edge_agent PRIVATE
+    ${SOFTBUS_DAEMON_ROOT}/src/core/plugin_system/PluginHost.cpp
+    ${SOFTBUS_DAEMON_ROOT}/src/core/plugin_system/PluginManager.cpp
+    ${SOFTBUS_DAEMON_ROOT}/src/message_bus/pipeline/stages/1_framers/PassthroughFramer.cpp
+)
+
 target_include_directories(softbus_edge_agent
     PRIVATE
         ${CMAKE_SOURCE_DIR}/src
+        ${SOFTBUS_DAEMON_ROOT}/src
+        ${SOFTBUS_DAEMON_ROOT}
 )
 
 target_link_libraries(softbus_edge_agent PRIVATE Qt6::Core Qt6::Network Qt6::SerialPort)
+if(UNIX AND NOT APPLE)
+    target_link_libraries(softbus_edge_agent PRIVATE dl)
+endif()
 if(SOFTBUS_HAVE_CAN)
     target_link_libraries(softbus_edge_agent PRIVATE Qt6::SerialBus)
 endif()
 
 target_compile_definitions(softbus_edge_agent PRIVATE SOFTBUS_HAVE_CAN=${SOFTBUS_HAVE_CAN})
 
+# 与 daemon 侧 ASan 编译的协议 .so 对齐：Debug/RelWithDebInfo 下主程序也链 ASan，否则 dlopen 会报
+# "ASan runtime does not come first" 并退出。Release 不受影响。可用 -DSOFTBUS_EDGE_AGENT_DEBUG_ASAN=OFF 关闭。
+option(SOFTBUS_EDGE_AGENT_DEBUG_ASAN "Link AddressSanitizer in Debug/RelWithDebInfo (non-Apple Unix)" ON)
+if(SOFTBUS_EDGE_AGENT_DEBUG_ASAN AND UNIX AND NOT APPLE)
+    if(CMAKE_CXX_COMPILER_ID MATCHES "GNU|Clang")
+        target_compile_options(softbus_edge_agent PRIVATE
+            $<$<OR:$<CONFIG:Debug>,$<CONFIG:RelWithDebInfo>>:-fsanitize=address>)
+        target_link_options(softbus_edge_agent PRIVATE
+            $<$<OR:$<CONFIG:Debug>,$<CONFIG:RelWithDebInfo>>:-fsanitize=address>)
+    endif()
+endif()
+
 if(UNIX AND NOT APPLE)
     find_package(PkgConfig QUIET)
     if(PkgConfig_FOUND)

CMakeLists.txt.user

@@ -1,6 +1,6 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <!DOCTYPE QtCreatorProject>
-<!-- Written by QtCreator 17.0.1, 2026-05-13T14:08:44. -->
+<!-- Written by QtCreator 17.0.1, 2026-05-14T16:06:22. -->
 <qtcreator>
  <data>
   <variable>EnvironmentId</variable>
@@ -93,29 +93,29 @@
   <valuemap type="QVariantMap">
    <value type="QString" key="DeviceType">Desktop</value>
    <value type="bool" key="HasPerBcDcs">true</value>
-   <value type="QString" key="ProjectExplorer.ProjectConfiguration.DefaultDisplayName">Qt 6.9.2</value>
+   <value type="QString" key="ProjectExplorer.ProjectConfiguration.DefaultDisplayName">Desktop Qt 6.9.2</value>
-   <value type="QString" key="ProjectExplorer.ProjectConfiguration.DisplayName">Qt 6.9.2</value>
+   <value type="QString" key="ProjectExplorer.ProjectConfiguration.DisplayName">Desktop Qt 6.9.2</value>
-   <value type="QString" key="ProjectExplorer.ProjectConfiguration.Id">{ac766883-1e6c-4832-9072-86a8b976b89e}</value>
+   <value type="QString" key="ProjectExplorer.ProjectConfiguration.Id">qt.qt6.692.linux_gcc_64_kit</value>
    <value type="qlonglong" key="ProjectExplorer.Target.ActiveBuildConfiguration">0</value>
    <value type="qlonglong" key="ProjectExplorer.Target.ActiveDeployConfiguration">0</value>
    <value type="qlonglong" key="ProjectExplorer.Target.ActiveRunConfiguration">0</value>
    <valuemap type="QVariantMap" key="ProjectExplorer.Target.BuildConfiguration.0">
-    <value type="QString" key="CMake.Build.Type">Build</value>
+    <value type="QString" key="CMake.Build.Type">Debug</value>
     <value type="int" key="CMake.Configure.BaseEnvironment">2</value>
     <value type="bool" key="CMake.Configure.ClearSystemEnvironment">false</value>
     <valuelist type="QVariantList" key="CMake.Configure.UserEnvironmentChanges"/>
-    <value type="QString" key="CMake.Initial.Parameters">-DCMAKE_PREFIX_PATH:PATH=%{Qt:QT_INSTALL_PREFIX}
+    <value type="QString" key="CMake.Initial.Parameters">-DCMAKE_GENERATOR:STRING=Ninja
--DCMAKE_BUILD_TYPE:STRING=Build
 -DCMAKE_CXX_COMPILER:FILEPATH=%{Compiler:Executable:Cxx}
--DCMAKE_GENERATOR:STRING=Unix Makefiles
--DCMAKE_CXX_FLAGS_INIT:STRING=%{Qt:QML_DEBUG_FLAG}
--DQT_MAINTENANCE_TOOL:FILEPATH=/home/dt/app/qt/MaintenanceTool
--DQT_QMAKE_EXECUTABLE:FILEPATH=%{Qt:qmakeExecutable}
--DCMAKE_PROJECT_INCLUDE_BEFORE:FILEPATH=%{BuildConfig:BuildDirectory:NativeFilePath}/.qtc/package-manager/auto-setup.cmake
 -DCMAKE_C_COMPILER:FILEPATH=%{Compiler:Executable:C}
--DCMAKE_COLOR_DIAGNOSTICS:BOOL=ON</value>
+-DCMAKE_COLOR_DIAGNOSTICS:BOOL=ON
-    <value type="QString" key="CMake.Source.Directory">/home/dt/myCode/bus/soft_bus/edge_agent</value>
+-DCMAKE_CXX_FLAGS_INIT:STRING=%{Qt:QML_DEBUG_FLAG}
-    <value type="QString" key="ProjectExplorer.BuildConfiguration.BuildDirectory">/home/dt/myCode/bus/soft_bus/edge_agent/build</value>
+-DCMAKE_PREFIX_PATH:PATH=%{Qt:QT_INSTALL_PREFIX}
+-DCMAKE_PROJECT_INCLUDE_BEFORE:FILEPATH=%{BuildConfig:BuildDirectory:NativeFilePath}/.qtc/package-manager/auto-setup.cmake
+-DQT_QMAKE_EXECUTABLE:FILEPATH=%{Qt:qmakeExecutable}
+-DCMAKE_BUILD_TYPE:STRING=Debug
+-DQT_MAINTENANCE_TOOL:FILEPATH=/home/dt/app/qt/MaintenanceTool</value>
+    <value type="int" key="EnableQmlDebugging">0</value>
+    <value type="QString" key="ProjectExplorer.BuildConfiguration.BuildDirectory">/home/dt/myCode/bus/soft_bus/edge_agent/build/Desktop_Qt_6_9_2-Debug</value>
     <valuemap type="QVariantMap" key="ProjectExplorer.BuildConfiguration.BuildStepList.0">
      <valuemap type="QVariantMap" key="ProjectExplorer.BuildStepList.Step.0">
       <value type="QString" key="CMakeProjectManager.MakeStep.BuildPreset"></value>
@@ -155,7 +155,7 @@
     <valuelist type="QVariantList" key="ProjectExplorer.BuildConfiguration.CustomParsers"/>
     <value type="bool" key="ProjectExplorer.BuildConfiguration.ParseStandardOutput">false</value>
     <valuelist type="QVariantList" key="ProjectExplorer.BuildConfiguration.UserEnvironmentChanges"/>
-    <value type="QString" key="ProjectExplorer.ProjectConfiguration.DisplayName">构建 (imported)</value>
+    <value type="QString" key="ProjectExplorer.ProjectConfiguration.DisplayName">Debug</value>
     <value type="QString" key="ProjectExplorer.ProjectConfiguration.Id">CMakeProjectManager.CMakeBuildConfiguration</value>
     <value type="qlonglong" key="ProjectExplorer.Target.ActiveDeployConfiguration">0</value>
     <value type="qlonglong" key="ProjectExplorer.Target.ActiveRunConfiguration">0</value>

README.md

@@ -1,14 +1,14 @@
 # softbus_edge_agent
 
-独立边缘进程：连接核心 `softbus_daemon` 的 **edgeIngress** TCP 端口，发送 **HELLO**、**DATA_RAWBUS**，以及通过极简 **本地 HTTP**（`GET /`、`GET /ui`、`POST /propose` 等）触发 **TREE_PROPOSE** 反向同步。
+独立边缘进程：连接核心 `softbus_daemon` 的 **edgeIngress** TCP 端口，发送 **HELLO**、**DEBUG_TUNNEL**（含周期性边缘心跳）、**DATA_RAWBUS**（真实总线数据），以及通过极简 **本地 HTTP**（`GET /`、`GET /ui`、`POST /propose` 等）触发 **TREE_PROPOSE** 反向同步。
 
 ## 目录布局（与 `softbus_daemon` 的 `src/` 对齐）
 
 | 路径 | 说明 |
 |------|------|
 | `src/device_bus/tree/` | 与核心同源的 `DeviceTreeModel`（本地子树、`TREE_FULL` 落盘加载） |
-| `src/device_bus/monitor/` | **udev 串口发现**、`EdgeDiscoveryCoordinator`（防抖后组 `TREE_PROPOSE` / `upsert` 向核心注册） |
+| `src/device_bus/monitor/` | **udev 串口发现**、`EdgeDiscoveryCoordinator`（防抖后：有核心则 `TREE_PROPOSE`，无核心则写本地树并 reconcile） |
-| `src/device_bus/manager/` | `SerialDeviceManager`、`CanDeviceManager`（无 Registry / Pipeline，仅 `IIngressPort` + 内存池） |
+| `src/device_bus/manager/` | `SerialDeviceManager`、`CanDeviceManager`；串口可选 **`EdgeStreamPipeline`**（daemon 同源 `PluginHost` + framer/parser/mapper） |
 | `src/device_bus/EdgeDeviceBus.*` | 边缘总线编排：加载本地树、**reconcile** 串口与 **SocketCAN** 实例 |
 | `src/devices/`、`src/hardware/` | 与核心同源的 `SerialDevice`、`SerialQtDriver` 等 |
 | `src/message_bus/ingress/` | `EdgeSession`、`EdgeUplinkWire`、`EdgeTcpUplinkPort`（`RawBusMessage` → SBUP 上行） |
@@ -22,6 +22,9 @@
 ```bash
 cmake -S . -B build -DCMAKE_PREFIX_PATH=/path/to/Qt/6.x/gcc_64
 cmake --build build
+# 无参：使用 $HOME/softbus/edge/edge_manifest.json（首次启动会自动创建目录并生成默认 manifest 与 local_tree.json）
+./build/bin/softbus_edge_agent
+# 或显式指定 manifest（例如仓库内示例；示例中 localTreePath 使用 $HOME/ 前缀便于各用户通用）：
 ./build/bin/softbus_edge_agent config/edge_manifest.example.json
 ```
 
@@ -29,7 +32,7 @@ cmake --build build
 
 ## 配置 `edge_manifest.json`（可选）
 
-若默认路径 **`config/edge_manifest.json`** 不存在、或传入的 manifest 路径无法打开、或 JSON 非法，进程会使用**内置默认值**启动（等价于示例 manifest 的字段），并在日志中提示。
+无命令行参数时，默认 manifest 为 **`$HOME/softbus/edge/edge_manifest.json`**（启动前会 `mkpath` 该目录；若缺少 `edge_manifest.json` / `local_tree.json` 则写入首次启动模板）。若该文件仍无法打开、或 JSON 非法、或你传入的其它 manifest 路径无效，进程会使用**内置默认值**合并/回退，并在日志中提示。
 
 仍可通过命令行指定路径覆盖默认文件位置：
 
@@ -41,16 +44,26 @@ cmake --build build
 
 要点：
 
-- **`coreHost` / `corePort`**：与 `daemon_config.json` 中 `edgeIngress` 一致。
+- **`coreHost` / `corePort`**：与 `daemon_config.json` 中 `edgeIngress` 一致。核心**晚于**边缘启动时，进程会**自动每约 3 秒重试 TCP**，连上后自动发送 **HELLO**（`currentTreeRevision` 为「核心侧已知 revision」与「本地设备树 revision」的**较大值**），之后 `DATA_RAWBUS` / `TREE_PROPOSE` 才会真正发到核心。
 - **`edgeId`**：HELLO 与会话绑定；`TREE_PROPOSE` 必须相同。
-- **`edgeDiscoveryEnabled`**（bool，默认 `true`）：在 TCP 连接且 **HELLO 成功** 后启动 udev 发现协调器，将串口以 `op: upsert` 写入核心设备树（需 Linux + `libudev`）。设为 `false` 可仅依赖手工 `TREE_PROPOSE` 或核心下发树。
+- **`edgeHeartbeatIntervalMs`**（int，默认 `5000`，`0` 关闭）：已连接核心 TCP 时，周期性经 **`DEBUG_TUNNEL`** 上送 **`kind: edge_heartbeat`** JSON（含 **`treeRevision`**、**`nodeCount`**、**`treeFingerprint`**（设备树 `tree` 紧凑 JSON 的 `qHash`）、**`coreTreeRevisionKnown`**、时间戳），核心在 `EdgeTcpIngressService` 中打日志便于对账。未设置时可回退读取旧字段 **`fakeIntervalMs`**。
-- **`endpoint`**：用于计算 `endpointHash`（与核心一致：`qHash(endpoint)` 即 Qt 默认字符串哈希）。
+- **`edgeDiscoveryEnabled`**（bool，默认 `true`）：启动 udev 串口发现。已连接核心且收到有效 **`treeRevision`** 后，将串口以 `op: upsert` 经 **`TREE_PROPOSE`** 同步到核心；**核心不可达或 HELLO 失败**时仍运行发现，并把结果写入**本地** `localTreePath`（与核心上线后的双向同步兼容）。设为 `false` 可仅依赖手工 `TREE_PROPOSE` 或核心下发树。需 Linux + `libudev`。
-- **`fakeDeviceId`**：组装 `routingKey = (endpointHash << 32) | deviceId` 与核心 `makeRoutingKey` 一致。
+- **真实串口**：由设备树节点驱动（`type`/`endpoint` 可识别为 Serial，且 `params.online` 为 true）。节点 `params.deviceId` 会写入上行 `RawBusMessage::deviceId`（未设则 `-1`）；**`endpoint` / `protocol`** 等字段写在设备树节点 `params` 中（与核心约定一致），不再在 manifest 顶层使用假数据字段。
-- **`fakeIntervalMs`**：设为 `0` 可关闭假数据定时器；**真实串口**由核心下发的设备树节点驱动（`type`/`endpoint` 可识别为 Serial，且 `params.online` 为 true）。节点 `params.deviceId` 会写入上行 `RawBusMessage::deviceId`（未设则 `-1`）。
+
+## 协议插件与流式解析（与 daemon Pipeline 对齐）
+
+当 manifest 中 **`plugins.protocols`** 非空时，进程会加载与 `softbus_daemon` **相同 ABI** 的 `.so`（如 `libmodbus_rtu.so`），并在串口 `read` 路径上启用：
+
+1. **Framer**：优先使用插件内的 `IProtocolFramerSession`（如 Modbus RTU 粘包）；否则退回 **PassthroughFramer**（每块 read 视为一帧，与未启用插件时行为一致）。
+2. **分帧线程与有界队列**：与 daemon `EndpointFramingState` 对齐——按 **`endpointHash`** 各起一条消费线程，串口读回调仅 **`tryEnqueue`** 到 **SPSC `LockFreeQueue`**（容量见 manifest **`edgeSerialChunkQueueCapacity`**，默认 `4096`；**`edgeUseEndpointFramerThreads`** 默认 `true`，`false` 时在同线程同步分帧供调试）。队列满时上报 **`edge stream chunk queue full`**。
+3. **Parser + Mapper**：与核心 `PipelineEngine::stage2Parser` 一致；mapper 在边缘使用 **sniffingEnabled=true** 且无 `mappingLookup`，故无映射规则时仍会产出带 `unmapped` 标记的 DOM 样本（仅 **日志** `EdgeStreamPipeline: parsed DOM ...`，**不上送 DOM 到核心**）。
+4. **上行**：仍只对 **完整帧** 发送 **`DATA_RAWBUS`**（与核心解码路径兼容）。
+
+**`plugins.protocols` 路径**：可为绝对路径，或与 manifest 文件同目录的相对路径。请确保边缘运行时能 `dlopen` 到该文件（通常与核心构建产物 `bin/plugins/*.so` 一致）。
 
 ## 运维 HTTP
 
-若 `opsListenPort > 0`，监听 `127.0.0.1:opsListenPort`：
+若 `opsListenPort > 0`，在 TCP 连接核心**之前**即启动监听（便于核心未启动时仍能打开本页排查配置）。默认 **`opsListenAllInterfaces` 为 false**：仅绑定 **`127.0.0.1`**，必须在**运行 edge_agent 的本机**浏览器打开。若需从局域网其他电脑访问，在 manifest 中设 **`"opsListenAllInterfaces": true`**（绑定 `0.0.0.0`，请自行评估内网安全）。
 
 - `GET /`：返回 `localTreePath` 文件内容（无文件则返回占位 HTML，含指向配置页的链接）。
 - **`GET /ui`**：**阶段 5** 简易配置页——从 **`GET /discovered`** 同源列表选择已发现串口节点，填写 `params.protocol`（如 `modbus_rtu`）、波特率等，提交后由边缘组装 **`op: upsert`** 的 `TREE_PROPOSE` 发往核心（使用当前会话的 **`baseTreeRevision`**，须已 HELLO 且收到过 `TREE_RESULT`）。
@@ -74,12 +87,19 @@ curl -s -X POST http://127.0.0.1:9080/propose \
 
 （`entries` 中节点 id 必须满足核心 `DeviceTreeModel::nodeBelongsToEdge` 规则。）
 
+### 打不开 `/ui` 时排查
+
+1. **`opsListenPort` 是否为 0**：为 0 则完全不监听；示例见 `config/edge_manifest.example.json`（如 `9080`）。
+2. **是否用错地址**：默认只能 **`http://127.0.0.1:<端口>/ui`** 在**边缘本机**打开；在别的电脑浏览器里填边缘的 `127.0.0.1` 是无效的，需 **`opsListenAllInterfaces: true`** 后用 **`http://<边缘局域网IP>:<端口>/ui`**，或 SSH 端口转发。
+3. **看进程日志**：启动成功会打印 `ops UI http://127.0.0.1:...` 或 `http://0.0.0.0:...`；若出现 `OpsUiHttp: listen failed`，多为端口被占用。
+4. **核心连不上**：自本版本起，只要 **`opsListenPort > 0`**，即使核心 TCP/HELLO 失败进程也会继续运行并保留运维 HTTP（日志中会提示 `continuing with ops UI only`）；此前版本会在启动 Ops 前退出，表现为「完全打不开」。
+
 ## 本地树文件
 
-`TREE_FULL` 落盘格式为 `{ "treeRevision", "edgeId", "tree": [ ... ] }`。**`EdgeDeviceBus`** 会将其中的 `tree` 当作 `device_tree` 加载，并对 **Serial** 与 **CAN**（`endpoint` 形如 `can:vcan0`，`type`/`endpoint` 推断为 CAN）执行 **reconcile**：打开设备后读到的数据经 `EdgeTcpUplinkPort` 编码为 `DATA_RAWBUS` 发往核心。与核心真源同步仍通过 **TREE_PROPOSE**（含边缘 **`upsert`** 自动注册）、核心侧 D-Bus `patch_device_tree_node` 或运维 **`push_edge_tree`**。
+`TREE_FULL` 落盘格式为 `{ "treeRevision", "edgeId", "tree": [ ... ] }`。**`EdgeDeviceBus`** 会将其中的 `tree` 当作 `device_tree` 加载，并对 **Serial** 与 **CAN**（`endpoint` 形如 `can:vcan0`，`type`/`endpoint` 推断为 CAN）执行 **reconcile**：打开设备后读到的数据经 `EdgeTcpUplinkPort` 编码为 `DATA_RAWBUS` 发往核心。串口节点可在 **`params`** 中设置 **`serialReadCoalesceMs`**（0～500，默认 **0**）：读字节先入驱动内缓冲，静音该毫秒数后合并再交给上层；启用流式 pipeline 时主路径为 **每 endpoint 分帧线程 + 有界队列**，一般保持 **0** 即可，仅在仍需驱动侧攒包时再调大。可选 **`serialReadBufferCapBytes`** 限制缓冲上限。
 
 ### 发现 → 注册流程（简要）
 
 1. 边缘连接核心并 **HELLO**；核心下发 **`TREE_FULL`**，本地落盘并 reconcile 已有节点。  
-2. `edgeDiscoveryEnabled` 为真时，`EdgeDiscoveryCoordinator` 扫描/监听 udev 串口 **add/remove**，在拿到有效 **`baseTreeRevision`**（来自 `TREE_RESULT`）后发送 **`TREE_PROPOSE`**：`op: upsert` 注册 `edges/<edgeId>/discovered/<stableKey>`，remove 时 **`patch`** `params.online: false`。  
+2. `edgeDiscoveryEnabled` 为真时，`EdgeDiscoveryCoordinator` 扫描/监听 udev 串口 **add/remove**：若已拿到核心 **`baseTreeRevision`** 则发送 **`TREE_PROPOSE`**；若**无核心**（或 HELLO 未成功）则把 **`op: upsert` / `patch`** 应用到**本地** `DeviceTreeModel` 并落盘 `local_tree.json`，再 **reconcile** 串口。核心恢复后仍通过 **`TREE_PROPOSE`** 与 **`TREE_FULL`** 与真源对齐。  
 3. 核心 `applyEdgePropose` 成功后再次 **`TREE_FULL`**；边缘 reload 树文件后对 **Serial/CAN** 执行 reconcile，**raw 仅上行**。

config/edge_manifest.example.json

@@ -5,10 +5,13 @@
   "edgeId": "edge-dev-1",
   "authToken": "",
   "opsListenPort": 9080,
-  "localTreePath": "/tmp/edge_local_tree.json",
+  "opsListenAllInterfaces": false,
-  "fakeIntervalMs": 5000,
+  "localTreePath": "$HOME/softbus/edge/local_tree.json",
-  "fakeDeviceId": 1,
+  "edgeHeartbeatIntervalMs": 5000,
-  "endpoint": "edge:fake",
+  "edgeDiscoveryEnabled": true,
-  "protocol": 0,
+  "plugins": {
-  "edgeDiscoveryEnabled": true
+    "protocols": [
+      { "path": "../../softbus_daemon/bin/plugins/libmodbus_rtu.so" }
+    ]
+  }
 }

config/edge_manifest.schema.json

@@ -8,15 +8,53 @@
     "corePort": { "type": "integer", "description": "daemon_config edgeIngress.port" },
     "edgeId": { "type": "string", "description": "HELLO / PROPOSE 与会话绑定的边缘 ID" },
     "authToken": { "type": "string", "description": "若核心配置了 edgeIngress.authToken，则此处必须一致" },
-    "opsListenPort": { "type": "integer", "minimum": 0, "description": "0 关闭本地运维 HTTP；否则监听 127.0.0.1:port" },
+    "opsListenPort": { "type": "integer", "minimum": 0, "description": "0 关闭运维 HTTP；否则监听端口。默认仅 127.0.0.1；设 opsListenAllInterfaces 为 true 时监听 0.0.0.0（局域网可访问，注意安全）" },
-    "localTreePath": { "type": "string", "description": "GET / 返回的本地树文件路径（展示用）" },
+    "opsListenAllInterfaces": {
-    "fakeIntervalMs": { "type": "integer", "minimum": 100, "description": "假数据 DATA_RAWBUS 发送周期（毫秒）" },
+      "type": "boolean",
-    "fakeDeviceId": { "type": "integer", "description": "RawBusMessage.deviceId" },
+      "description": "false（默认）时运维 HTTP 仅绑定本机回环；true 时绑定所有网卡，便于从局域网浏览器访问"
-    "endpoint": { "type": "string", "description": "用于 qHash(endpoint) 得到 endpointHash，须与核心设备树约定一致" },
+    },
-    "protocol": { "type": "integer", "description": "ProtocolType 枚举数值（与核心 RawBusMessage 一致）" },
+    "localTreePath": {
+      "type": "string",
+      "description": "本地设备树 JSON 路径（Ops GET / 展示、TREE_FULL 落盘）。未配置时内置默认为 $HOME/softbus/edge/local_tree.json。字符串若以 $HOME/、${HOME}/ 或 ~/ 开头，会在加载时展开为当前用户家目录"
+    },
+    "edgeHeartbeatIntervalMs": {
+      "type": "integer",
+      "minimum": 0,
+      "description": "经 DEBUG_TUNNEL 上送边缘心跳的周期（毫秒）。载荷 JSON 含 treeRevision、nodeCount、treeFingerprint（设备树紧凑 JSON 的 qHash）。0 关闭"
+    },
     "edgeDiscoveryEnabled": {
       "type": "boolean",
-      "description": "HELLO 成功后是否启动 udev 串口发现并向核心发送 TREE_PROPOSE（upsert/patch）；需 Linux libudev"
+      "description": "为 true 时启动 udev 串口发现：已连接核心且拿到 treeRevision 后发送 TREE_PROPOSE；无核心或 HELLO 失败时仍发现，并将 upsert/patch 写入本地 local_tree.json 与 reconcile。需 Linux libudev"
+    },
+    "edgeSerialChunkQueueCapacity": {
+      "type": "integer",
+      "minimum": 64,
+      "description": "启用流式 pipeline 时，每个 endpointHash 一条分帧线程对应的有界 SPSC 队列容量（与 daemon EndpointFramingState::chunkQ 语义对齐）。默认 4096"
+    },
+    "edgeUseEndpointFramerThreads": {
+      "type": "boolean",
+      "description": "为 true（默认）时串口读回调仅 tryEnqueue 到分帧线程；为 false 时在同线程同步分帧（调试用）。主路径推荐 true"
+    },
+    "plugins": {
+      "type": "object",
+      "description": "可选；与 softbus_daemon 的 daemon_config.plugins 形态一致。非空时启用边缘流式 framer/parser/mapper（见 README）",
+      "properties": {
+        "protocols": {
+          "type": "array",
+          "items": {
+            "oneOf": [
+              { "type": "string", "description": "协议插件 .so 路径（与 manifest 同目录的相对路径可解析）" },
+              {
+                "type": "object",
+                "required": ["path"],
+                "properties": {
+                  "path": { "type": "string", "description": "dlopen 路径，通常指向 libmodbus_rtu.so 等" }
+                }
+              }
+            ]
+          }
+        }
+      }
     }
   }
 }

src/device_bus/EdgeDeviceBus.cpp

@@ -8,6 +8,7 @@
 #include <QSet>
 
 #include "device_bus/manager/SerialDeviceManager.h"
+#include "pipeline/EdgeStreamPipeline.h"
 #if SOFTBUS_HAVE_CAN
 #include "device_bus/manager/CanDeviceManager.h"
 #endif
@@ -19,7 +20,8 @@ namespace softbus::device_bus
 
 EdgeDeviceBus::EdgeDeviceBus()
     : m_tree(std::make_shared<DeviceTreeModel>())
-    , m_pool(std::make_shared<softbus::core::memory::MemoryPool>())
+    // 多串口并发读 + 分帧再分配：默认 1024 块在 11 路串口下易耗尽，导致 Modbus 分帧静默不出帧
+    , m_pool(std::make_shared<softbus::core::memory::MemoryPool>(4096, 16384))
 {
 }
 
@@ -36,6 +38,20 @@ void EdgeDeviceBus::setIngressPort(softbus::message_bus::ingress::IIngressPort*
     m_ingress = port;
 }
 
+void EdgeDeviceBus::setStreamPipeline(std::shared_ptr<softbus_edge::EdgeStreamPipeline> pipeline)
+{
+    m_streamPipeline = std::move(pipeline);
+    if (m_streamPipeline) {
+        m_streamPipeline->setTreeModel(m_tree.get());
+        m_streamPipeline->setFramedSink([this](softbus::core::models::RawBusMessage msg) {
+            if (!m_ingress) {
+                return false;
+            }
+            return m_ingress->push(std::move(msg));
+        });
+    }
+}
+
 bool EdgeDeviceBus::loadLocalEdgeTreeFile(const QString& path, QString* errorMessage)
 {
     QFile f(path);
@@ -56,13 +72,18 @@ bool EdgeDeviceBus::loadLocalEdgeTreeFile(const QString& path, QString* errorMes
     const QJsonObject rootObj = doc.object();
     const QJsonArray treeArr = rootObj.value(QStringLiteral("tree")).toArray();
     if (treeArr.isEmpty()) {
-        if (errorMessage) {
+        QJsonObject wrapped;
-            *errorMessage = QStringLiteral("missing or empty 'tree' array");
+        wrapped.insert(QStringLiteral("device_tree"), QJsonArray());
+        if (rootObj.contains(QStringLiteral("treeRevision"))) {
+            wrapped.insert(QStringLiteral("device_tree_revision"), rootObj.value(QStringLiteral("treeRevision")));
         }
-        return false;
+        return m_tree->loadFromDaemonConfig(wrapped, errorMessage);
     }
     QJsonObject wrapped;
     wrapped.insert(QStringLiteral("device_tree"), treeArr);
+    if (rootObj.contains(QStringLiteral("treeRevision"))) {
+        wrapped.insert(QStringLiteral("device_tree_revision"), rootObj.value(QStringLiteral("treeRevision")));
+    }
     return m_tree->loadFromDaemonConfig(wrapped, errorMessage);
 }
 
@@ -130,11 +151,16 @@ void EdgeDeviceBus::reconcileSerialFromTree()
         }
     }
 
+    if (m_streamPipeline) {
+        m_streamPipeline->setTreeModel(m_tree.get());
+    }
+
     softbus::device_bus::manager::SerialInitDeps deps;
     deps.treeModel = m_tree.get();
     deps.dirtyCb = []() {};
     deps.ingressPort = m_ingress;
     deps.memoryPool = m_pool;
+    deps.streamPipeline = m_streamPipeline;
     deps.serialDevicesByEndpoint = &m_serialByEndpoint;
 
     for (const QString& ep : desired) {

src/device_bus/EdgeDeviceBus.h

@@ -14,6 +14,11 @@
 #endif
 #include "message_bus/ingress/IIngressPort.h"
 
+namespace softbus_edge
+{
+class EdgeStreamPipeline;
+}
+
 namespace softbus::device_bus
 {
 
@@ -29,6 +34,9 @@ public:
 
     void setIngressPort(softbus::message_bus::ingress::IIngressPort* port);
 
+    /** 可选：加载与 daemon 相同的协议 .so 后启用 framer+parser+mapper（见 manifest plugins.protocols）。 */
+    void setStreamPipeline(std::shared_ptr<softbus_edge::EdgeStreamPipeline> pipeline);
+
     /** 读取边缘落盘：`{ "tree": [...], "treeRevision", "edgeId" }`，包装为 `loadFromDaemonConfig`。 */
     bool loadLocalEdgeTreeFile(const QString& path, QString* errorMessage = nullptr);
 
@@ -51,6 +59,7 @@ private:
 
     std::shared_ptr<DeviceTreeModel> m_tree;
     std::shared_ptr<softbus::core::memory::MemoryPool> m_pool;
+    std::shared_ptr<softbus_edge::EdgeStreamPipeline> m_streamPipeline;
     softbus::message_bus::ingress::IIngressPort* m_ingress{nullptr};
     QHash<QString, QSharedPointer<softbus::devices::physical::SerialDevice>> m_serialByEndpoint;
 #if SOFTBUS_HAVE_CAN

src/device_bus/manager/SerialDeviceManager.cpp

@@ -1,12 +1,16 @@
 #include "device_bus/manager/SerialDeviceManager.h"
 
+#include <algorithm>
+
 #include <QJsonDocument>
 #include <QSerialPort>
 
 #include "device_bus/tree/DeviceTreeModel.h"
+#include "core/models/MessageRoutingUtils.h"
 #include "devices/DeviceTypes.h"
 #include "hardware/drivers/serial/SerialCapabilities.h"
 #include "hardware/drivers/serial/SerialQtDriver.h"
+#include "pipeline/EdgeStreamPipeline.h"
 #include "utils/logs/logging.h"
 
 using softbus::hardware::drivers::serial::SerialQtDriver;
@@ -15,6 +19,27 @@ namespace serial_cap = softbus::hardware::drivers::serial::capabilities;
 namespace softbus::device_bus::manager
 {
 
+/** 发现节点常无 params.deviceId；生成稳定正整数供 DATA_RAWBUS / 路由使用。 */
+static int edgeResolveRuntimeDeviceId(const QJsonObject& params, const QString& endpoint)
+{
+    const int explicitId = params.value(QStringLiteral("deviceId")).toInt(-1);
+    if (explicitId >= 0) {
+        return explicitId;
+    }
+    const QString sk = params.value(QStringLiteral("stableKey")).toString();
+    if (!sk.isEmpty()) {
+        bool ok = false;
+        const quint64 u = sk.toULongLong(&ok);
+        if (ok && u > 0) {
+            return static_cast<int>((u % 2147483646ULL) + 1ULL);
+        }
+        const quint32 h = qHash(sk);
+        return static_cast<int>(h % 2147483646U) + 1;
+    }
+    const quint32 eh = softbus::core::models::makeEndpointHash(endpoint);
+    return static_cast<int>(eh % 2147483646U) + 1;
+}
+
 static bool setIfChanged(QJsonObject& params, const QString& key, const QJsonValue& value)
 {
     const auto it = params.find(key);
@@ -121,8 +146,15 @@ bool initializeSerialDeviceWithDeps(const QString& endpoint, const QJsonObject&
     }
     driver->SetFlowControl(flow);
 
+    driver->SetReadCoalesceMs(
+        std::min(500, std::max(0, params.value(QStringLiteral("serialReadCoalesceMs")).toInt(8))));
+    const int capParam = params.value(QStringLiteral("serialReadBufferCapBytes")).toInt(0);
+    if (capParam > 0) {
+        driver->SetReadBufferCapBytes(capParam);
+    }
+
     softbus::devices::DeviceInfo info;
-    info.id = params.value(QStringLiteral("deviceId")).toInt(-1);
+    info.id = edgeResolveRuntimeDeviceId(params, endpoint);
     info.endpoint = endpoint;
     info.type = QStringLiteral("serial");
     info.kind = softbus::devices::DeviceKind::Serial;
@@ -139,6 +171,9 @@ bool initializeSerialDeviceWithDeps(const QString& endpoint, const QJsonObject&
     QSharedPointer<softbus::devices::physical::SerialDevice> device(
         new softbus::devices::physical::SerialDevice(std::move(driver), cfg, info));
     device->bindIngress(deps.ingressPort, deps.memoryPool);
+    if (deps.streamPipeline) {
+        device->setStreamPipeline(deps.streamPipeline);
+    }
 
     if (!device->start()) {
         LOG_ERROR() << "Edge SerialDeviceManager: failed to start serial device for" << endpoint;

src/device_bus/manager/SerialDeviceManager.h

@@ -13,6 +13,11 @@
 #include "devices/physical/SerialDevice.h"
 #include "message_bus/ingress/IIngressPort.h"
 
+namespace softbus_edge
+{
+class EdgeStreamPipeline;
+}
+
 class DeviceTreeModel;
 
 namespace softbus::device_bus::manager
@@ -24,6 +29,8 @@ struct SerialInitDeps
     std::function<void()> dirtyCb;
     softbus::message_bus::ingress::IIngressPort* ingressPort{nullptr};
     std::shared_ptr<softbus::core::memory::MemoryPool> memoryPool;
+    /** 可选：与 daemon 对齐的 framer/parser；非空时 SerialDevice 会走流式解析再上行。 */
+    std::shared_ptr<softbus_edge::EdgeStreamPipeline> streamPipeline;
     QHash<QString, QSharedPointer<softbus::devices::physical::SerialDevice>>* serialDevicesByEndpoint{nullptr};
 };
 

src/device_bus/monitor/EdgeDiscoveryCoordinator.cpp

@@ -5,8 +5,11 @@
 #include <QJsonDocument>
 #include <QJsonObject>
 #include <QMutexLocker>
+#include <QSaveFile>
 #include <QTimer>
 
+#include "device_bus/EdgeDeviceBus.h"
+#include "device_bus/tree/DeviceTreeModel.h"
 #include "device_bus/monitor/UdevDeviceMonitor.h"
 #include "message_bus/ingress/EdgeSession.h"
 #include "utils/logs/logging.h"
@@ -39,6 +42,18 @@ void EdgeDiscoveryCoordinator::setCoreTreeRevisionGetter(std::function<qint64()>
     m_getCoreRevision = std::move(fn);
 }
 
+void EdgeDiscoveryCoordinator::setLocalEdgePersistence(softbus::device_bus::EdgeDeviceBus* bus,
+                                                         const QString& localTreePath)
+{
+    m_edgeBus = bus;
+    m_localTreePath = localTreePath;
+}
+
+void EdgeDiscoveryCoordinator::setStandaloneLocalOnly(bool standalone)
+{
+    m_standaloneLocalOnly = standalone;
+}
+
 QString EdgeDiscoveryCoordinator::stableKeyForSerial(const softbus::device_bus::monitor::DeviceEvent& ev) const
 {
     const QString syspath = ev.props.value(QStringLiteral("SYSPATH"));
@@ -63,7 +78,7 @@ void EdgeDiscoveryCoordinator::enqueueUpsert(const softbus::device_bus::monitor:
     params.insert(QStringLiteral("edgeId"), m_edgeId);
     params.insert(QStringLiteral("stableKey"), sk);
     params.insert(QStringLiteral("online"), true);
-    params.insert(QStringLiteral("baudRate"), 115200);
+    params.insert(QStringLiteral("baudRate"), 9600);
     params.insert(QStringLiteral("protocol"), QStringLiteral("modbus_rtu"));
     params.insert(QStringLiteral("discoveredVia"), QStringLiteral("udev"));
     if (ev.props.contains(QStringLiteral("SYSPATH"))) {
@@ -161,17 +176,12 @@ void EdgeDiscoveryCoordinator::scheduleFlush()
 
 void EdgeDiscoveryCoordinator::flushPending()
 {
-    if (!m_session || m_edgeId.isEmpty()) {
+    if (m_edgeId.isEmpty()) {
         return;
     }
     if (m_proposeInFlight) {
         return;
     }
-    const qint64 base = m_getCoreRevision ? m_getCoreRevision() : 0;
-    if (base <= 0) {
-        LOG_INFO() << "EdgeDiscoveryCoordinator: skip flush, core treeRevision not ready yet";
-        return;
-    }
     if (m_pendingUpsert.isEmpty() && m_pendingPatchEntries.isEmpty()) {
         return;
     }
@@ -192,18 +202,88 @@ void EdgeDiscoveryCoordinator::flushPending()
         return;
     }
 
-    QJsonObject body;
+    const qint64 coreRev = m_getCoreRevision ? m_getCoreRevision() : 0;
-    body.insert(QStringLiteral("edgeId"), m_edgeId);
+    const bool useCorePropose = !m_standaloneLocalOnly && coreRev > 0 && m_session;
-    body.insert(QStringLiteral("baseTreeRevision"), base);
-    body.insert(QStringLiteral("entries"), entries);
 
-    if (!m_session->sendTreePropose(body)) {
+    if (useCorePropose) {
-        LOG_WARNING() << "EdgeDiscoveryCoordinator: sendTreePropose failed";
+        QJsonObject body;
+        body.insert(QStringLiteral("edgeId"), m_edgeId);
+        body.insert(QStringLiteral("baseTreeRevision"), coreRev);
+        body.insert(QStringLiteral("entries"), entries);
+
+        if (!m_session->sendTreePropose(body)) {
+            LOG_WARNING() << "EdgeDiscoveryCoordinator: sendTreePropose failed";
+            return;
+        }
+
+        m_proposeInFlight = true;
+        LOG_INFO() << "EdgeDiscoveryCoordinator: sent TREE_PROPOSE" << entries.size() << "entries baseRev" << coreRev;
         return;
     }
 
-    m_proposeInFlight = true;
+    if (!m_standaloneLocalOnly && coreRev <= 0) {
-    LOG_INFO() << "EdgeDiscoveryCoordinator: sent TREE_PROPOSE" << entries.size() << "entries baseRev" << base;
+        LOG_INFO() << "EdgeDiscoveryCoordinator: skip flush, waiting for core treeRevision (HELLO / TREE_RESULT)";
+        return;
+    }
+
+    if (!m_edgeBus) {
+        LOG_WARNING() << "EdgeDiscoveryCoordinator: standalone/local mode but EdgeDeviceBus not set, cannot persist";
+        return;
+    }
+
+    DeviceTreeModel& tm = m_edgeBus->treeModel();
+    const qint64 localBase = tm.treeRevision();
+    const DeviceTreeModel::EdgeProposeResult res = tm.applyEdgePropose(m_edgeId, localBase, entries);
+    if (!res.ok) {
+        LOG_WARNING() << "EdgeDiscoveryCoordinator: local applyEdgePropose failed" << res.code << res.message;
+        if (res.code == QStringLiteral("CONFLICT")) {
+            scheduleFlush();
+        }
+        return;
+    }
+
+    QString perr;
+    if (!persistLocalEdgeTree(&perr)) {
+        LOG_WARNING() << "EdgeDiscoveryCoordinator: persist local tree failed" << perr;
+    }
+    m_edgeBus->reconcileSerialFromTree();
+    m_edgeBus->reconcileCanFromTree();
+
+    m_pendingUpsert.clear();
+    m_pendingPatchEntries.clear();
+    LOG_INFO() << "EdgeDiscoveryCoordinator: applied discovery to local tree revision" << res.newRevision << "nodes"
+               << tm.nodes().size();
+}
+
+bool EdgeDiscoveryCoordinator::persistLocalEdgeTree(QString* errOut)
+{
+    if (!m_edgeBus || m_localTreePath.isEmpty()) {
+        if (errOut) {
+            *errOut = QStringLiteral("not configured");
+        }
+        return false;
+    }
+    const DeviceTreeModel& tm = m_edgeBus->treeModel();
+    QJsonObject doc;
+    doc.insert(QStringLiteral("treeRevision"), tm.treeRevision());
+    doc.insert(QStringLiteral("edgeId"), m_edgeId);
+    doc.insert(QStringLiteral("tree"), tm.toDeviceTreeJson());
+
+    QSaveFile out(m_localTreePath);
+    if (!out.open(QIODevice::WriteOnly | QIODevice::Truncate)) {
+        if (errOut) {
+            *errOut = QStringLiteral("open failed");
+        }
+        return false;
+    }
+    out.write(QJsonDocument(doc).toJson(QJsonDocument::Indented));
+    if (!out.commit()) {
+        if (errOut) {
+            *errOut = QStringLiteral("commit failed");
+        }
+        return false;
+    }
+    return true;
 }
 
 void EdgeDiscoveryCoordinator::onTreeResultFromCore(const QJsonObject& o)
@@ -283,4 +363,9 @@ void EdgeDiscoveryCoordinator::stop()
     m_devnodeToStableKey.clear();
 }
 
+void EdgeDiscoveryCoordinator::requestFlush()
+{
+    scheduleFlush();
+}
+
 } // namespace softbus_edge

src/device_bus/monitor/EdgeDiscoveryCoordinator.h

@@ -25,10 +25,15 @@ namespace softbus::message_bus::ingress
 class EdgeSession;
 }
 
+namespace softbus::device_bus
+{
+class EdgeDeviceBus;
+}
+
 namespace softbus_edge
 {
 
-/** udev 串口发现 → 组装 `TREE_PROPOSE`（含 `op: upsert`）向核心注册设备树节点。 */
+/** udev 串口发现 → 组装 `TREE_PROPOSE` 向核心同步；无核心时写入本地 `DeviceTreeModel` 并落盘 `localTreePath`。 */
 class EdgeDiscoveryCoordinator final : public QObject
 {
 public:
@@ -39,9 +44,18 @@ public:
     void setEdgeId(const QString& edgeId);
     void setCoreTreeRevisionGetter(std::function<qint64()> fn);
 
+    /** 无核心或尚未拿到 core treeRevision 时，将发现结果写入本地 `DeviceTreeModel` 并落盘到 `localTreePath`。 */
+    void setLocalEdgePersistence(softbus::device_bus::EdgeDeviceBus* bus, const QString& localTreePath);
+
+    /** 由 main 设置：TCP/HELLO 未就绪、仅以边缘为真源运行（udev → 本地树）时为 true。 */
+    void setStandaloneLocalOnly(bool standalone);
+
     bool start();
     void stop();
 
+    /** 请求在防抖后尽快执行一次 `flushPending`（例如核心链路刚恢复时）。 */
+    void requestFlush();
+
     void onTreeResultFromCore(const QJsonObject& o);
 
     QJsonArray recentDiscoveriesJson() const;
@@ -54,10 +68,14 @@ private:
     void enqueueOfflinePatch(const QString& devnode);
     void scheduleFlush();
     void flushPending();
+    bool persistLocalEdgeTree(QString* errOut);
 
     softbus::message_bus::ingress::EdgeSession* m_session{nullptr};
     QString m_edgeId;
     std::function<qint64()> m_getCoreRevision;
+    softbus::device_bus::EdgeDeviceBus* m_edgeBus{nullptr};
+    QString m_localTreePath;
+    bool m_standaloneLocalOnly{false};
 
     std::unique_ptr<softbus::device_bus::monitor::UdevDeviceMonitor> m_udev;
 

src/devices/physical/SerialDevice.cpp

@@ -11,6 +11,7 @@
 #include "core/models/RawBusMessage.h"
 #include "hardware/drivers/serial/SerialQtDriver.h"
 #include "message_bus/ingress/IIngressPort.h"
+#include "pipeline/EdgeStreamPipeline.h"
 #include "utils/logs/logging.h"
 
 using softbus::hardware::drivers::serial::SerialQtDriver;
@@ -78,6 +79,11 @@ void SerialDevice::bindIngress(softbus::message_bus::ingress::IIngressPort* ingr
     m_pool = std::move(pool);
 }
 
+void SerialDevice::setStreamPipeline(std::shared_ptr<softbus_edge::EdgeStreamPipeline> pipeline)
+{
+    m_streamPipeline = std::move(pipeline);
+}
+
 // 串口写入数据回调 不定长数据
 bool SerialDevice::writePayload(const std::vector<std::uint8_t>& data)
 {
@@ -111,24 +117,28 @@ void SerialDevice::onRead(const std::vector<std::uint8_t>& data)
 
     std::copy(data.begin(), data.end(), block->data);
     block->size = data.size();
-    // @test code
     LOG_INFO() << "SerialDevice: onRead data size" << data.size() << "endpoint" << m_config.endpoint;
-    // 将数据推送到 ingress 队列 包含端口名、协议、数据、时间戳、traceId
-    softbus::core::models::RawBusMessage ctx;
-    ctx.direction = softbus::core::models::BusDirection::Upstream;
-    ctx.endpointHash = softbus::core::models::makeEndpointHash(m_config.endpoint);
-    ctx.protocol = softbus::core::models::protocolTypeFromHint(m_info.protocol);
-    ctx.logicalAddress = 0;
-    ctx.routingKey = softbus::core::models::makeRoutingKey(ctx.endpointHash, m_info.id);
-    ctx.timestampMs = QDateTime::currentMSecsSinceEpoch();
-    ctx.deviceId = m_info.id;
-    ctx.payload.block = std::move(block);
-    ctx.payload.length = data.size();
-    ctx.traceId = QStringLiteral("serial:%1:%2")
-                      .arg(m_config.endpoint)
-                      .arg(ctx.timestampMs);
 
-    if (!m_ingress->push(std::move(ctx))) {
+    softbus::core::models::RawBusMessage chunk;
+    chunk.direction = softbus::core::models::BusDirection::Upstream;
+    chunk.endpointHash = softbus::core::models::makeEndpointHash(m_config.endpoint);
+    chunk.protocol = softbus::core::models::protocolTypeFromHint(m_info.protocol);
+    chunk.logicalAddress = 0;
+    chunk.routingKey = softbus::core::models::makeRoutingKey(chunk.endpointHash, m_info.id);
+    chunk.timestampMs = QDateTime::currentMSecsSinceEpoch();
+    chunk.deviceId = m_info.id;
+    chunk.payload.block = std::move(block);
+    chunk.payload.length = data.size();
+    chunk.traceId = QStringLiteral("serial:%1:%2").arg(m_config.endpoint).arg(chunk.timestampMs);
+
+    if (m_streamPipeline && m_streamPipeline->ok()) {
+        if (!m_streamPipeline->tryEnqueueStreamChunk(std::move(chunk))) {
+            m_ingress->reportError(m_config.endpoint, QStringLiteral("edge stream chunk queue full"));
+        }
+        return;
+    }
+
+    if (!m_ingress->push(std::move(chunk))) {
         m_ingress->reportError(m_config.endpoint, QStringLiteral("ingress queue full"));
     }
 }

src/devices/physical/SerialDevice.h

@@ -10,6 +10,7 @@
 namespace softbus::hardware::drivers::serial { class SerialQtDriver; }
 namespace softbus::message_bus::ingress { class IIngressPort; }
 namespace softbus::core::memory { class MemoryPool; }
+namespace softbus_edge { class EdgeStreamPipeline; }
 
 namespace softbus::devices::physical
 {
@@ -42,6 +43,8 @@ public:
     softbus::devices::DeviceKind   kind() const;
     void bindIngress(softbus::message_bus::ingress::IIngressPort* ingress,
                      std::shared_ptr<softbus::core::memory::MemoryPool> pool);
+    /** 可选：与 daemon Pipeline 对齐的 framer+parser+mapper；未设置则保持每 read 直接上行。 */
+    void setStreamPipeline(std::shared_ptr<softbus_edge::EdgeStreamPipeline> pipeline);
     bool writePayload(const std::vector<std::uint8_t>& data);
     bool writePayload(const std::uint8_t* data, std::size_t size);
 
@@ -55,6 +58,7 @@ private:
     softbus::devices::DeviceInfo                                         m_info;
     softbus::message_bus::ingress::IIngressPort*                         m_ingress{nullptr};
     std::shared_ptr<softbus::core::memory::MemoryPool>                   m_pool;
+    std::shared_ptr<softbus_edge::EdgeStreamPipeline>                    m_streamPipeline;
 };
 
 } // namespace softbus::devices::physical

src/fake_driver.cpp

@@ -1,20 +0,0 @@
-#include "fake_driver.hpp"
-
-FakeDriver::FakeDriver(QObject* parent)
-    : QObject(parent)
-{
-    connect(&m_timer, &QTimer::timeout, this, &FakeDriver::onTimeout);
-}
-
-void FakeDriver::start(int intervalMs, std::function<void()> onTick)
-{
-    m_onTick = std::move(onTick);
-    m_timer.start(intervalMs);
-}
-
-void FakeDriver::onTimeout()
-{
-    if (m_onTick) {
-        m_onTick();
-    }
-}

src/fake_driver.hpp

@@ -1,22 +0,0 @@
-#pragma once
-
-#include <functional>
-
-#include <QObject>
-#include <QTimer>
-
-class FakeDriver final : public QObject
-{
-    Q_OBJECT
-public:
-    explicit FakeDriver(QObject* parent = nullptr);
-
-    void start(int intervalMs, std::function<void()> onTick);
-
-private slots:
-    void onTimeout();
-
-private:
-    QTimer m_timer;
-    std::function<void()> m_onTick;
-};

src/hardware/drivers/serial/SerialQtDriver.cpp

@@ -1,7 +1,12 @@
 #include "hardware/drivers/serial/SerialQtDriver.h"
 
+#include <algorithm>
+#include <cstring>
+
 #include <QByteArray>
 #include <QMetaObject>
+#include <QSerialPort>
+#include <QTimer>
 #include <QThread>
 
 #include "utils/logs/logging.h"
@@ -13,6 +18,10 @@ SerialQtDriver::SerialQtDriver(QObject* parent)
     : QObject(parent),
       m_port(this)
 {
+    m_flushReadCoalesce = new QTimer(this);
+    m_flushReadCoalesce->setSingleShot(true);
+    m_flushReadCoalesce->setTimerType(Qt::PreciseTimer);
+    connect(m_flushReadCoalesce, &QTimer::timeout, this, &SerialQtDriver::flushPendingRead);
     connect(&m_port, &QSerialPort::readyRead, this, &SerialQtDriver::handleReadyRead);
     connect(&m_port, &QSerialPort::errorOccurred, this, &SerialQtDriver::handleError);
 }
@@ -106,23 +115,53 @@ void SerialQtDriver::SetFlowControl(QSerialPort::FlowControl flow)
     m_port.setFlowControl(flow);
 }
 
+void SerialQtDriver::SetReadCoalesceMs(int ms)
+{
+    m_readCoalesceMs = std::max(0, ms);
+}
+
+void SerialQtDriver::SetReadBufferCapBytes(int cap)
+{
+    m_readBufferCap = std::max(1024, cap);
+}
+
 void SerialQtDriver::handleReadyRead()
 {
+    const QByteArray data = m_port.readAll();
+    if (data.isEmpty()) {
+        return;
+    }
     if (!m_readCallback) {
-        const QByteArray data = m_port.readAll();
         LOG_DEBUG() << "SerialQtDriver: dropped" << data.size() << "bytes on" << m_port.portName()
                     << "because no read callback is set";
         return;
     }
 
-    const QByteArray data = m_port.readAll();
+    m_pendingReadBuf.append(data);
-    if (data.isEmpty()) {
+    if (m_pendingReadBuf.size() >= m_readBufferCap) {
+        flushPendingRead();
         return;
     }
 
-    std::vector<std::uint8_t> buffer(reinterpret_cast<const std::uint8_t*>(data.constData()),
+    if (m_readCoalesceMs <= 0) {
-                                     reinterpret_cast<const std::uint8_t*>(data.constData()) + data.size());
+        flushPendingRead();
-    m_readCallback(buffer);
+        return;
+    }
+
+    m_flushReadCoalesce->start(m_readCoalesceMs);
+}
+
+void SerialQtDriver::flushPendingRead()
+{
+    if (!m_readCallback || m_pendingReadBuf.isEmpty()) {
+        m_pendingReadBuf.clear();
+        return;
+    }
+    const int n = m_pendingReadBuf.size();
+    std::vector<std::uint8_t> buffer(static_cast<std::size_t>(n));
+    std::memcpy(buffer.data(), m_pendingReadBuf.constData(), static_cast<std::size_t>(n));
+    m_pendingReadBuf.clear();
+    m_readCallback(std::move(buffer));
 }
 
 void SerialQtDriver::handleError(QSerialPort::SerialPortError error)
@@ -170,12 +209,27 @@ bool SerialQtDriver::openOnOwnerThread(const QString& endpoint)
         return false;
     }
 
+    // 显式接收缓冲：避免默认过小导致高频小 read；具体上限仍受 OS 驱动约束。
+    m_port.setReadBufferSize(256 * 1024);
+    m_pendingReadBuf.clear();
+    if (m_flushReadCoalesce) {
+        m_flushReadCoalesce->stop();
+    }
+
     LOG_DEBUG() << "SerialQtDriver: opened port" << m_port.portName();
     return true;
 }
 
 void SerialQtDriver::closeOnOwnerThread()
 {
+    if (m_flushReadCoalesce) {
+        m_flushReadCoalesce->stop();
+    }
+    if (!m_pendingReadBuf.isEmpty()) {
+        flushPendingRead();
+    } else {
+        m_pendingReadBuf.clear();
+    }
     if (m_port.isOpen()) {
         const QString name = m_port.portName();
         m_port.close();

src/hardware/drivers/serial/SerialQtDriver.h

@@ -1,10 +1,12 @@
 #pragma once
 
+#include <QByteArray>
 #include <QObject>
 #include <QSerialPort>
 #include <QString>
 #include <cstddef>
-#include <QByteArray>
+
+class QTimer;
 
 #include "hardware/interfaces/IStreamDriver.h"
 
@@ -74,19 +76,32 @@ public:
     void SetStopBits(QSerialPort::StopBits stopBits);
     void SetFlowControl(QSerialPort::FlowControl flow);
 
+    /**
+     * 读侧字节合并（对齐 daemon 侧 Pipeline 入队前的「攒包」思路，减轻高频 1~2 字节 readyRead 对上层/framer 的冲击）。
+     * 0：每次 `readAll()` 后立即回调；>0：数据先入内部缓冲，连续 `readyRead` 静默达到 ms 后合并为一次回调。默认 0ms。
+     */
+    void SetReadCoalesceMs(int ms);
+    /** 内部读缓冲上限（字节），达到后强制立刻 flush，避免无限增长。默认 256KiB。 */
+    void SetReadBufferCapBytes(int cap);
+
 private slots:
     void handleReadyRead();
     void handleError(QSerialPort::SerialPortError error);
+    void flushPendingRead();
 
 private:
     bool openOnOwnerThread(const QString& endpoint);
     void closeOnOwnerThread();
     bool writeOnOwnerThread(const QByteArray& bytes);
 
-private:
     QSerialPort   m_port;
     ReadCallback  m_readCallback;
     ErrorCallback m_errorCallback;
+
+    QTimer*   m_flushReadCoalesce{nullptr};
+    QByteArray m_pendingReadBuf;
+    int        m_readCoalesceMs{0};
+    int        m_readBufferCap{256 * 1024};
 };
 
 } // namespace softbus::hardware::drivers::serial

src/main.cpp

@@ -1,22 +1,56 @@
 #include <QCoreApplication>
 #include <QDateTime>
 #include <QDebug>
+#include <QDir>
 #include <QFile>
+#include <QFileInfo>
 #include <QJsonArray>
 #include <QJsonDocument>
 #include <QJsonObject>
 #include <QJsonParseError>
 #include <QSaveFile>
+#include <QTimer>
+
+#include <algorithm>
 
 #include "device_bus/EdgeDeviceBus.h"
 #include "device_bus/monitor/EdgeDiscoveryCoordinator.h"
 #include "message_bus/ingress/EdgeSession.h"
 #include "message_bus/ingress/EdgeTcpUplinkPort.h"
-#include "message_bus/ingress/EdgeUplinkWire.h"
+#include "pipeline/EdgeStreamPipeline.h"
 
-#include "fake_driver.hpp"
 #include "ops_ui_http.hpp"
 
+/** 默认数据目录：`$HOME/softbus/edge`（自动创建）。 */
+static QString edgeDefaultDataDir()
+{
+    return QDir::cleanPath(QDir::homePath() + QLatin1String("/softbus/edge"));
+}
+
+static QString edgeDefaultManifestPath()
+{
+    return edgeDefaultDataDir() + QLatin1String("/edge_manifest.json");
+}
+
+static QString edgeDefaultLocalTreePath()
+{
+    return edgeDefaultDataDir() + QLatin1String("/local_tree.json");
+}
+
+/** manifest 中 `localTreePath` 若以 `$HOME/`、`${HOME}/` 或 `~/` 开头，则展开为当前用户家目录。 */
+static QString expandManifestHomePath(QString p)
+{
+    const QString home = QDir::homePath();
+    if (p.startsWith(QLatin1String("$HOME/"))) {
+        p.replace(0, 6, home);
+    } else if (p.startsWith(QLatin1String("${HOME}/"))) {
+        p.replace(0, 7, home);
+    } else if (p.startsWith(QLatin1String("~/"))) {
+        p.replace(0, 1, home);
+    }
+    return QDir::cleanPath(p);
+}
+
 /** 无配置文件或与核心同目录运行时使用的默认值（可被 manifest 覆盖）。 */
 static QJsonObject defaultManifest()
 {
@@ -26,12 +60,13 @@ static QJsonObject defaultManifest()
     m.insert(QStringLiteral("edgeId"), QStringLiteral("edge-dev-1"));
     m.insert(QStringLiteral("authToken"), QStringLiteral(""));
     m.insert(QStringLiteral("opsListenPort"), 0);
-    m.insert(QStringLiteral("localTreePath"), QStringLiteral("/tmp/edge_local_tree.json"));
+    m.insert(QStringLiteral("opsListenAllInterfaces"), false);
-    m.insert(QStringLiteral("fakeIntervalMs"), 5000);
+    m.insert(QStringLiteral("localTreePath"), edgeDefaultLocalTreePath());
-    m.insert(QStringLiteral("fakeDeviceId"), 1);
+    m.insert(QStringLiteral("edgeHeartbeatIntervalMs"), 5000);
-    m.insert(QStringLiteral("endpoint"), QStringLiteral("edge:fake"));
-    m.insert(QStringLiteral("protocol"), 0);
     m.insert(QStringLiteral("edgeDiscoveryEnabled"), true);
+    QJsonObject plugins;
+    plugins.insert(QStringLiteral("protocols"), QJsonArray());
+    m.insert(QStringLiteral("plugins"), plugins);
     return m;
 }
 
@@ -57,6 +92,170 @@ static QJsonObject loadManifestOrDefaults(const QString& path)
     return out;
 }
 
+/** 创建 `$HOME/softbus/edge`，并写入首次启动的 `edge_manifest.json` / `local_tree.json`（若不存在）。 */
+static bool ensureEdgeDataLayout(QString* errOut)
+{
+    const QString root = edgeDefaultDataDir();
+    QDir dir;
+    if (!dir.mkpath(root)) {
+        if (errOut) {
+            *errOut = QStringLiteral("mkpath failed: %1").arg(root);
+        }
+        return false;
+    }
+    const QString treePath = edgeDefaultLocalTreePath();
+    if (!QFile::exists(treePath)) {
+        QJsonObject doc;
+        doc.insert(QStringLiteral("treeRevision"), 0);
+        doc.insert(QStringLiteral("edgeId"), QStringLiteral("edge-dev-1"));
+        doc.insert(QStringLiteral("tree"), QJsonArray());
+        QFile f(treePath);
+        if (!f.open(QIODevice::WriteOnly | QIODevice::Truncate)) {
+            if (errOut) {
+                *errOut = QStringLiteral("cannot write %1").arg(treePath);
+            }
+            return false;
+        }
+        f.write(QJsonDocument(doc).toJson(QJsonDocument::Indented));
+    }
+    const QString manPath = edgeDefaultManifestPath();
+    if (!QFile::exists(manPath)) {
+        QFile f(manPath);
+        if (!f.open(QIODevice::WriteOnly | QIODevice::Truncate)) {
+            if (errOut) {
+                *errOut = QStringLiteral("cannot write %1").arg(manPath);
+            }
+            return false;
+        }
+        f.write(QJsonDocument(defaultManifest()).toJson(QJsonDocument::Indented));
+    }
+    return true;
+}
+
+static QStringList protocolPluginPathsFromManifest(const QJsonObject& manifest, const QString& manifestPath)
+{
+    QStringList out;
+    const QJsonArray arr =
+        manifest.value(QStringLiteral("plugins")).toObject().value(QStringLiteral("protocols")).toArray();
+    const QFileInfo mf(manifestPath);
+    const QDir baseDir = mf.absoluteDir();
+    for (const QJsonValue& v : arr) {
+        QString p;
+        if (v.isObject()) {
+            p = v.toObject().value(QStringLiteral("path")).toString();
+        } else if (v.isString()) {
+            p = v.toString();
+        }
+        if (p.isEmpty()) {
+            continue;
+        }
+        if (!QFileInfo(p).isAbsolute()) {
+            p = baseDir.filePath(p);
+        }
+        out.append(p);
+    }
+    return out;
+}
+
+static bool appendPluginPathIfFileExists(QStringList& paths, const QString& candidate)
+{
+    const QFileInfo fi(candidate);
+    if (!fi.exists() || !fi.isFile()) {
+        return false;
+    }
+    paths.append(fi.absoluteFilePath());
+    return true;
+}
+
+/** 从 `startDir` 起向父目录最多爬 `maxLevels` 层，查找 `softbus_daemon/bin/plugins/libmodbus_rtu.so`。 */
+static bool appendDefaultModbusPluginByWalkingUp(QStringList& paths, const QString& startDir, int maxLevels = 10)
+{
+    QDir d(QFileInfo(startDir).absoluteFilePath());
+    if (!d.exists()) {
+        return false;
+    }
+    for (int i = 0; i < maxLevels; ++i) {
+        const QString candidate =
+            d.absoluteFilePath(QStringLiteral("softbus_daemon/bin/plugins/libmodbus_rtu.so"));
+        if (appendPluginPathIfFileExists(paths, candidate)) {
+            qInfo() << "Edge: manifest has no plugins.protocols; auto-loaded" << paths.constLast();
+            return true;
+        }
+        if (!d.cdUp()) {
+            break;
+        }
+    }
+    return false;
+}
+
+/** manifest 未配置 plugins 时，尝试从常见相对路径加载 modbus_rtu（与仓库内 edge_manifest.example 一致）。 */
+static void appendDefaultProtocolPluginsIfManifestEmpty(QStringList& paths, const QString& manifestPath)
+{
+    if (!paths.isEmpty()) {
+        return;
+    }
+    const QByteArray envPath = qgetenv("EDGE_MODBUS_RTU_SO");
+    if (!envPath.isEmpty()) {
+        const QString p = QString::fromLocal8Bit(envPath.trimmed());
+        if (appendPluginPathIfFileExists(paths, p)) {
+            qInfo() << "Edge: loaded protocol plugin from EDGE_MODBUS_RTU_SO" << paths.constLast();
+            return;
+        }
+        qWarning() << "Edge: EDGE_MODBUS_RTU_SO is set but file not found or not a file:" << p;
+    }
+
+    // 与数据目录约定一致：$HOME/softbus/bin/plugins 放置协议插件（用户态安装）
+    const QString homeSoftbusPlugin =
+        QDir(QDir::homePath()).filePath(QStringLiteral("softbus/bin/plugins/libmodbus_rtu.so"));
+    if (appendPluginPathIfFileExists(paths, homeSoftbusPlugin)) {
+        qInfo() << "Edge: manifest has no plugins.protocols; auto-loaded" << paths.constLast();
+        return;
+    }
+
+    const QString appBin = QCoreApplication::applicationDirPath();
+    const QString manDir = QFileInfo(manifestPath).absolutePath();
+    const QString exeFile = QCoreApplication::applicationFilePath();
+
+    if (!exeFile.isEmpty() && appendDefaultModbusPluginByWalkingUp(paths, QFileInfo(exeFile).absolutePath())) {
+        return;
+    }
+    if (appendDefaultModbusPluginByWalkingUp(paths, appBin)) {
+        return;
+    }
+    if (appendDefaultModbusPluginByWalkingUp(paths, manDir)) {
+        return;
+    }
+
+    const QStringList legacyCandidates = {
+        QDir(appBin).filePath(QStringLiteral("plugins/libmodbus_rtu.so")),
+        QDir(appBin).filePath(QStringLiteral("../../../softbus_daemon/bin/plugins/libmodbus_rtu.so")),
+        QDir(manDir).filePath(QStringLiteral("../../softbus_daemon/bin/plugins/libmodbus_rtu.so")),
+        QDir(manDir).filePath(QStringLiteral("../softbus_daemon/bin/plugins/libmodbus_rtu.so")),
+    };
+    for (const QString& p : legacyCandidates) {
+        if (appendPluginPathIfFileExists(paths, p)) {
+            qInfo() << "Edge: manifest has no plugins.protocols; auto-loaded" << paths.constLast();
+            return;
+        }
+    }
+
+    const QString home = QDir::homePath();
+    const QStringList homeDevCandidates = {
+        QDir(home).filePath(QStringLiteral("myCode/bus/soft_bus/softbus_daemon/bin/plugins/libmodbus_rtu.so")),
+        QDir(home).filePath(QStringLiteral("myCode/bus/soft_bus/softbus_daemon/build/plugins/libmodbus_rtu.so")),
+    };
+    for (const QString& p : homeDevCandidates) {
+        if (appendPluginPathIfFileExists(paths, p)) {
+            qInfo() << "Edge: manifest has no plugins.protocols; auto-loaded (under $HOME)" << paths.constLast();
+            return;
+        }
+    }
+
+    qWarning() << "Edge: plugins.protocols is empty and libmodbus_rtu.so not found. Expected"
+                  " $HOME/softbus/bin/plugins/libmodbus_rtu.so (or set EDGE_MODBUS_RTU_SO / add plugins to"
+                  " edge_manifest.json).";
+}
+
 static void tryReloadTreeFromDisk(softbus::device_bus::EdgeDeviceBus& bus, const QString& localTreePath)
 {
     QString err;
@@ -69,11 +268,47 @@ static void tryReloadTreeFromDisk(softbus::device_bus::EdgeDeviceBus& bus, const
     }
 }
 
+/** 兼容旧 manifest 的 `fakeIntervalMs`。 */
+static int edgeHeartbeatIntervalMsFromManifest(const QJsonObject& m)
+{
+    if (m.contains(QStringLiteral("edgeHeartbeatIntervalMs"))) {
+        return m.value(QStringLiteral("edgeHeartbeatIntervalMs")).toInt(5000);
+    }
+    return m.value(QStringLiteral("fakeIntervalMs")).toInt(5000);
+}
+
+/** 经 `DEBUG_TUNNEL` 上送：设备树节点数、revision、紧凑 JSON 的 qHash 指纹（便于对账）。 */
+static QJsonObject makeEdgeHeartbeatPayload(const QString& edgeId,
+                                            qint64 coreTreeRevisionKnown,
+                                            const softbus::device_bus::EdgeDeviceBus& bus)
+{
+    const DeviceTreeModel& tm = bus.treeModel();
+    const QJsonArray treeJson = tm.toDeviceTreeJson();
+    const QByteArray canon = QJsonDocument(treeJson).toJson(QJsonDocument::Compact);
+    const quint32 treeFp = static_cast<quint32>(qHash(canon));
+
+    QJsonObject o;
+    o.insert(QStringLiteral("kind"), QStringLiteral("edge_heartbeat"));
+    o.insert(QStringLiteral("edgeId"), edgeId);
+    o.insert(QStringLiteral("tsMs"), QDateTime::currentMSecsSinceEpoch());
+    o.insert(QStringLiteral("treeRevision"), static_cast<qint64>(tm.treeRevision()));
+    o.insert(QStringLiteral("nodeCount"), treeJson.size());
+    o.insert(QStringLiteral("treeFingerprint"), static_cast<qint64>(treeFp));
+    o.insert(QStringLiteral("coreTreeRevisionKnown"), coreTreeRevisionKnown);
+    return o;
+}
+
 int main(int argc, char* argv[])
 {
     QCoreApplication app(argc, argv);
-    const QString manifestPath = argc > 1 ? QString::fromLocal8Bit(argv[1])
+
-                                            : QStringLiteral("config/edge_manifest.json");
+    QString layoutErr;
+    if (!ensureEdgeDataLayout(&layoutErr)) {
+        qWarning() << "ensureEdgeDataLayout:" << layoutErr;
+    }
+
+    const QString manifestPath =
+        argc > 1 ? QString::fromLocal8Bit(argv[1]) : edgeDefaultManifestPath();
     const QJsonObject m = loadManifestOrDefaults(manifestPath);
 
     const QString coreHost = m.value(QStringLiteral("coreHost")).toString(QStringLiteral("127.0.0.1"));
@@ -81,11 +316,10 @@ int main(int argc, char* argv[])
     const QString edgeId = m.value(QStringLiteral("edgeId")).toString(QStringLiteral("edge-dev-1"));
     const QString authToken = m.value(QStringLiteral("authToken")).toString();
     const int opsPort = m.value(QStringLiteral("opsListenPort")).toInt(0);
-    const QString localTreePath = m.value(QStringLiteral("localTreePath")).toString(QStringLiteral("/tmp/edge_local_tree.json"));
+    const bool opsListenAll = m.value(QStringLiteral("opsListenAllInterfaces")).toBool(false);
-    const int fakeIntervalMs = m.value(QStringLiteral("fakeIntervalMs")).toInt(5000);
+    const QString localTreePath =
-    const int deviceId = m.value(QStringLiteral("fakeDeviceId")).toInt(1);
+        expandManifestHomePath(m.value(QStringLiteral("localTreePath")).toString(edgeDefaultLocalTreePath()));
-    const QString endpoint = m.value(QStringLiteral("endpoint")).toString(QStringLiteral("edge:fake"));
+    const int heartbeatIntervalMs = edgeHeartbeatIntervalMsFromManifest(m);
-    const int protocol = m.value(QStringLiteral("protocol")).toInt(0);
     const bool edgeDiscoveryEnabled = m.value(QStringLiteral("edgeDiscoveryEnabled")).toBool(true);
 
     softbus::device_bus::EdgeDeviceBus edgeBus;
@@ -96,11 +330,35 @@ int main(int argc, char* argv[])
     discovery.setEdgeSession(&session);
     discovery.setEdgeId(edgeId);
     discovery.setCoreTreeRevisionGetter([&coreTreeRevision]() { return coreTreeRevision; });
+    discovery.setLocalEdgePersistence(&edgeBus, localTreePath);
 
     auto tcpUplink = std::make_unique<softbus::message_bus::ingress::EdgeTcpUplinkPort>(
         [&session](const QByteArray& fr) { return session.writeSbupFrame(fr); });
     edgeBus.setIngressPort(tcpUplink.get());
 
+    QStringList pluginPaths = protocolPluginPathsFromManifest(m, manifestPath);
+    appendDefaultProtocolPluginsIfManifestEmpty(pluginPaths, manifestPath);
+    {
+        softbus_edge::EdgeStreamPipeline::Config pcfg;
+        pcfg.pool = edgeBus.memoryPool();
+        pcfg.protocolPluginPaths = pluginPaths;
+        pcfg.treeModel = &edgeBus.treeModel();
+        pcfg.edgeSerialChunkQueueCapacity =
+            static_cast<std::size_t>(std::max(64, m.value(QStringLiteral("edgeSerialChunkQueueCapacity")).toInt(4096)));
+        pcfg.edgeUseEndpointFramerThreads = m.value(QStringLiteral("edgeUseEndpointFramerThreads")).toBool(true);
+        auto pipe = std::make_shared<softbus_edge::EdgeStreamPipeline>(pcfg);
+        if (!pipe->ok()) {
+            qWarning() << "EdgeStreamPipeline: disabled:" << pipe->lastError();
+        } else {
+            edgeBus.setStreamPipeline(pipe);
+            if (!pluginPaths.isEmpty()) {
+                qInfo() << "EdgeStreamPipeline: loaded protocol plugins" << pluginPaths;
+            } else {
+                qInfo() << "EdgeStreamPipeline: active (passthrough; manifest has no protocol plugins)";
+            }
+        }
+    }
+
     if (QFile::exists(localTreePath)) {
         tryReloadTreeFromDisk(edgeBus, localTreePath);
     }
@@ -146,19 +404,10 @@ int main(int argc, char* argv[])
                          tryReloadTreeFromDisk(edgeBus, localTreePath);
                      });
 
-    if (!session.connectToCore(coreHost, corePort)) {
+    QObject::connect(&session, &softbus::message_bus::ingress::EdgeSession::receivedDebugTunnel, &app,
-        qWarning() << "cannot reach core; start softbus_daemon with edgeIngress.enabled and matching port,"
+                     [&](const QJsonObject& envelope) {
-                   << "or set coreHost/corePort in edge_manifest.json";
+                         qInfo() << "DebugTunnel from host" << envelope;
-        return 3;
+                     });
-    }
-    if (!session.sendHello(edgeId, authToken, 0)) {
-        qWarning() << "hello send failed";
-        return 4;
-    }
-
-    if (edgeDiscoveryEnabled) {
-        (void)discovery.start();
-    }
 
     OpsUiHttp ops;
     if (opsPort > 0) {
@@ -171,42 +420,62 @@ int main(int argc, char* argv[])
                                qWarning() << "propose send failed";
                            }
                        },
-                       [&discovery]() { return discovery.recentDiscoveriesJson(); })) {
+                       [&discovery]() { return discovery.recentDiscoveriesJson(); },
+                       [&](const QJsonObject& dbg) { return session.sendDebugToCore(dbg); },
+                       [&session]() { return session.recentDebugFromHost(); },
+                       opsListenAll)) {
             qWarning() << "ops ui listen failed";
         } else {
-            qInfo() << "ops UI http://127.0.0.1:" << opsPort
+            qInfo() << "ops UI"
-                    << " GET / GET /ui GET /discovered POST /propose POST /register_discovered";
+                    << (opsListenAll ? QStringLiteral("http://0.0.0.0:") : QStringLiteral("http://127.0.0.1:"))
+                    << opsPort << "GET / GET /ui GET /discovered GET /debug GET /debug_host POST /propose"
+                                  " POST /register_discovered POST /debug_edge";
         }
     }
 
-    FakeDriver fake;
+    discovery.setStandaloneLocalOnly(true);
-    if (fakeIntervalMs > 0) {
+    session.configureUplink(
-        fake.start(fakeIntervalMs, [&]() {
+        coreHost,
-            const std::uint32_t epHash = static_cast<std::uint32_t>(qHash(endpoint));
+        corePort,
-            const std::uint64_t routing = (static_cast<std::uint64_t>(epHash) << 32)
+        edgeId,
-                                            | static_cast<std::uint32_t>(deviceId);
+        authToken,
-            const QByteArray inner = softbus::message_bus::ingress::edge_uplink::encodeDataRawBusPayload(
+        [&]() { return std::max(coreTreeRevision, edgeBus.treeModel().treeRevision()); });
-                deviceId,
+    QObject::connect(&session, &softbus::message_bus::ingress::EdgeSession::uplinkReady, &app, [&]() {
-                epHash,
+        discovery.setStandaloneLocalOnly(false);
-                static_cast<std::uint16_t>(protocol),
+        discovery.requestFlush();
-                0,
+    });
-                routing,
+    QObject::connect(&session, &softbus::message_bus::ingress::EdgeSession::uplinkLost, &app, [&]() {
-                0,
+        discovery.setStandaloneLocalOnly(true);
-                QDateTime::currentMSecsSinceEpoch(),
+    });
-                1,
+
-                QStringLiteral("edge:fake"),
+    if (edgeDiscoveryEnabled) {
-                QByteArray(1, '\x01'));
+        if (!discovery.start()) {
-            if (inner.isEmpty()) {
+            qWarning() << "EdgeDiscoveryCoordinator: udev discovery did not start";
-                qWarning() << "encode data failed";
+        }
-                return;
-            }
-            if (!session.sendDataRawBusInner(inner)) {
-                qWarning() << "send data failed";
-            }
-        });
     }
 
-    QObject::connect(&app, &QCoreApplication::aboutToQuit, &discovery, [&discovery]() { discovery.stop(); });
+    session.beginUplinkLoop();
+
+    QTimer heartbeatTimer;
+    if (heartbeatIntervalMs > 0) {
+        heartbeatTimer.setInterval(std::clamp(heartbeatIntervalMs, 100, 24 * 3600 * 1000));
+        heartbeatTimer.setTimerType(Qt::CoarseTimer);
+        QObject::connect(&heartbeatTimer, &QTimer::timeout, &app, [&session, &edgeBus, edgeId, &coreTreeRevision]() {
+            if (!session.isCoreTcpConnected()) {
+                return;
+            }
+            const QJsonObject body = makeEdgeHeartbeatPayload(edgeId, coreTreeRevision, edgeBus);
+            if (!session.sendDebugToCore(body)) {
+                qWarning() << "edge heartbeat (DEBUG_TUNNEL) send failed";
+            }
+        });
+        heartbeatTimer.start();
+    }
+
+    QObject::connect(&app, &QCoreApplication::aboutToQuit, &app, [&discovery, &session]() {
+        session.shutdownUplink();
+        discovery.stop();
+    });
 
     (void)tcpUplink;
     return app.exec();

src/message_bus/ingress/EdgeSession.cpp

@@ -1,7 +1,9 @@
 #include "message_bus/ingress/EdgeSession.h"
 
+#include <QDateTime>
 #include <QJsonDocument>
 #include <QJsonParseError>
+#include <QTimer>
 
 #include "message_bus/ingress/EdgeUplinkWire.h"
 
@@ -18,6 +20,123 @@ EdgeSession::EdgeSession(QObject* parent)
     connect(&m_sock, &QTcpSocket::readyRead, this, &EdgeSession::onSocketReadyRead);
 }
 
+void EdgeSession::configureUplink(const QString& host,
+                                  int port,
+                                  const QString& edgeId,
+                                  const QString& authToken,
+                                  std::function<qint64()> currentTreeRevision)
+{
+    m_uplinkHost = host;
+    m_uplinkPort = port;
+    m_uplinkEdgeId = edgeId;
+    m_uplinkAuthToken = authToken;
+    m_uplinkTreeRevision = std::move(currentTreeRevision);
+
+    if (!m_reconnectTimer) {
+        m_reconnectTimer = new QTimer(this);
+        m_reconnectTimer->setSingleShot(true);
+        connect(m_reconnectTimer, &QTimer::timeout, this, &EdgeSession::kickConnectInternal);
+    }
+
+    if (!m_uplinkHooksInstalled) {
+        m_uplinkHooksInstalled = true;
+        connect(&m_sock, &QTcpSocket::connected, this, &EdgeSession::onTcpConnected);
+        connect(&m_sock, &QTcpSocket::disconnected, this, &EdgeSession::onTcpDisconnected);
+        connect(&m_sock, &QAbstractSocket::errorOccurred, this, &EdgeSession::onTcpError);
+    }
+}
+
+void EdgeSession::beginUplinkLoop()
+{
+    m_uplinkShutdown = false;
+    kickConnectInternal();
+}
+
+void EdgeSession::shutdownUplink()
+{
+    m_uplinkShutdown = true;
+    if (m_reconnectTimer) {
+        m_reconnectTimer->stop();
+    }
+    m_sock.abort();
+}
+
+bool EdgeSession::isCoreTcpConnected() const
+{
+    return m_sock.state() == QAbstractSocket::ConnectedState;
+}
+
+void EdgeSession::kickConnectInternal()
+{
+    if (m_uplinkShutdown || m_uplinkHost.isEmpty() || m_uplinkPort <= 0) {
+        return;
+    }
+    if (m_sock.state() == QAbstractSocket::ConnectedState
+        || m_sock.state() == QAbstractSocket::ConnectingState) {
+        return;
+    }
+    if (m_sock.state() != QAbstractSocket::UnconnectedState) {
+        m_sock.abort();
+    }
+    qInfo() << "EdgeSession: connecting to core" << m_uplinkHost << m_uplinkPort;
+    m_sock.connectToHost(m_uplinkHost, static_cast<quint16>(m_uplinkPort));
+}
+
+void EdgeSession::planReconnect()
+{
+    if (m_uplinkShutdown || !m_reconnectTimer) {
+        return;
+    }
+    if (!m_reconnectTimer->isActive()) {
+        qInfo() << "EdgeSession: scheduling core reconnect in 3s";
+        m_reconnectTimer->start(3000);
+    }
+}
+
+void EdgeSession::onTcpConnected()
+{
+    if (m_uplinkShutdown) {
+        return;
+    }
+    if (m_reconnectTimer) {
+        m_reconnectTimer->stop();
+    }
+    const qint64 tr = m_uplinkTreeRevision ? m_uplinkTreeRevision() : 0;
+    if (!sendHello(m_uplinkEdgeId, m_uplinkAuthToken, tr)) {
+        qWarning() << "EdgeSession: HELLO write failed";
+        emit uplinkLost();
+        m_sock.abort();
+        planReconnect();
+        return;
+    }
+    qInfo() << "EdgeSession: TCP up, HELLO sent currentTreeRevision" << tr;
+    flushPendingSbupFrames();
+    emit uplinkReady();
+}
+
+void EdgeSession::onTcpDisconnected()
+{
+    if (m_uplinkShutdown) {
+        return;
+    }
+    qWarning() << "EdgeSession: disconnected from core";
+    emit uplinkLost();
+    planReconnect();
+}
+
+void EdgeSession::onTcpError(QAbstractSocket::SocketError)
+{
+    if (m_uplinkShutdown) {
+        return;
+    }
+    qWarning() << "EdgeSession: socket error" << m_sock.errorString();
+    emit uplinkLost();
+    if (m_sock.state() != QAbstractSocket::UnconnectedState) {
+        m_sock.abort();
+    }
+    planReconnect();
+}
+
 bool EdgeSession::connectToCore(const QString& host, int port)
 {
     m_sock.connectToHost(host, static_cast<quint16>(port));
@@ -35,6 +154,9 @@ void EdgeSession::disconnectCore()
 
 bool EdgeSession::sendHello(const QString& edgeId, const QString& authToken, qint64 currentTreeRevision)
 {
+    if (m_sock.state() != QAbstractSocket::ConnectedState) {
+        return false;
+    }
     QJsonObject o;
     o.insert(QStringLiteral("edgeId"), edgeId);
     if (!authToken.isEmpty()) {
@@ -48,20 +170,89 @@ bool EdgeSession::sendHello(const QString& edgeId, const QString& authToken, qin
 
 bool EdgeSession::sendDataRawBusInner(const QByteArray& innerPayload)
 {
+    if (m_sock.state() != QAbstractSocket::ConnectedState) {
+        return false;
+    }
     const QByteArray frame = encodeOuterFrame(FrameKind::DataRawBus, innerPayload);
     return m_sock.write(frame) == frame.size();
 }
 
 bool EdgeSession::sendTreePropose(const QJsonObject& body)
 {
+    if (m_sock.state() != QAbstractSocket::ConnectedState) {
+        return false;
+    }
     const QByteArray pl = QJsonDocument(body).toJson(QJsonDocument::Compact);
     const QByteArray frame = encodeOuterFrame(FrameKind::TreePropose, pl);
     return m_sock.write(frame) == frame.size();
 }
 
+bool EdgeSession::sendDebugToCore(const QJsonObject& body)
+{
+    QJsonObject envelope;
+    envelope.insert(QStringLiteral("v"), 1);
+    envelope.insert(QStringLiteral("from"), QStringLiteral("edge"));
+    envelope.insert(QStringLiteral("tsMs"), QDateTime::currentMSecsSinceEpoch());
+    envelope.insert(QStringLiteral("body"), body);
+    const QByteArray pl = QJsonDocument(envelope).toJson(QJsonDocument::Compact);
+    const QByteArray frame = encodeOuterFrame(FrameKind::DebugTunnel, pl);
+    if (m_sock.state() != QAbstractSocket::ConnectedState) {
+        return false;
+    }
+    return m_sock.write(frame) == frame.size();
+}
+
+void EdgeSession::flushPendingSbupFrames()
+{
+    if (m_sock.state() != QAbstractSocket::ConnectedState) {
+        return;
+    }
+    while (!m_pendingSbupFrames.isEmpty()) {
+        const QByteArray frame = m_pendingSbupFrames.takeFirst();
+        if (m_sock.write(frame) != frame.size()) {
+            m_pendingSbupFrames.prepend(frame);
+            break;
+        }
+    }
+}
+
 bool EdgeSession::writeSbupFrame(const QByteArray& frame)
 {
-    return m_sock.write(frame) == frame.size();
+    if (frame.isEmpty()) {
+        return false;
+    }
+    if (m_sock.state() != QAbstractSocket::ConnectedState) {
+        if (m_pendingSbupFrames.size() < kMaxPendingSbupFrames) {
+            m_pendingSbupFrames.push_back(frame);
+            return true;
+        }
+        return false;
+    }
+    if (m_sock.write(frame) != frame.size()) {
+        if (m_pendingSbupFrames.size() < kMaxPendingSbupFrames) {
+            m_pendingSbupFrames.push_back(frame);
+        }
+        return false;
+    }
+    flushPendingSbupFrames();
+    return true;
+}
+
+void EdgeSession::appendDebugFromHost(const QJsonObject& envelope)
+{
+    m_debugFromHostRing.push_back(envelope);
+    if (m_debugFromHostRing.size() > kMaxDebugRing) {
+        m_debugFromHostRing.remove(0, m_debugFromHostRing.size() - kMaxDebugRing);
+    }
+}
+
+QJsonArray EdgeSession::recentDebugFromHost() const
+{
+    QJsonArray arr;
+    for (const QJsonObject& o : m_debugFromHostRing) {
+        arr.append(o);
+    }
+    return arr;
 }
 
 void EdgeSession::onSocketReadyRead()
@@ -84,6 +275,14 @@ void EdgeSession::onSocketReadyRead()
             if (pe.error == QJsonParseError::NoError && doc.isObject()) {
                 emit receivedTreeFull(doc.object());
             }
+        } else if (popped->kind == FrameKind::DebugTunnel) {
+            QJsonParseError pe;
+            const QJsonDocument doc = QJsonDocument::fromJson(popped->payload, &pe);
+            if (pe.error == QJsonParseError::NoError && doc.isObject()) {
+                const QJsonObject o = doc.object();
+                appendDebugFromHost(o);
+                emit receivedDebugTunnel(o);
+            }
         }
     }
 }

src/message_bus/ingress/EdgeSession.h

@@ -1,9 +1,17 @@
 #pragma once
 
+#include <functional>
+
+#include <QAbstractSocket>
 #include <QByteArray>
+#include <QJsonArray>
 #include <QJsonObject>
 #include <QObject>
+#include <QString>
 #include <QTcpSocket>
+#include <QVector>
+
+class QTimer;
 
 namespace softbus::message_bus::ingress
 {
@@ -17,25 +25,69 @@ public:
     bool connectToCore(const QString& host, int port);
     void disconnectCore();
 
+    /**
+     * 配置核心 TCP 与 HELLO 参数，并启用自动重连（核心晚于边缘启动时仍可连上）。
+     * 调用 `beginUplinkLoop()` 后开始尝试连接；断线或拒绝连接后约 3s 重试。
+     */
+    void configureUplink(const QString& host,
+                         int port,
+                         const QString& edgeId,
+                         const QString& authToken,
+                         std::function<qint64()> currentTreeRevision);
+    void beginUplinkLoop();
+    void shutdownUplink();
+    bool isCoreTcpConnected() const;
+
     bool sendHello(const QString& edgeId, const QString& authToken, qint64 currentTreeRevision);
     bool sendDataRawBusInner(const QByteArray& innerPayload);
     bool sendTreePropose(const QJsonObject& body);
+    /** 经 SBUP `DebugTunnel` 向核心发送 JSON（外层载荷为 UTF-8 JSON）。 */
+    bool sendDebugToCore(const QJsonObject& body);
     /** 已带 SBUP 外层的任意帧（如完整 `DATA_RAWBUS`）。 */
     bool writeSbupFrame(const QByteArray& frame);
 
     void setMaxPayloadBytes(int m) { m_maxPayload = m; }
 
+    /** 最近从主机收到的 `DebugTunnel` 记录（环形，用于 Ops `/debug_host`）。 */
+    QJsonArray recentDebugFromHost() const;
+
 signals:
     void receivedTreeResult(const QJsonObject& obj);
     void receivedTreeFull(const QJsonObject& obj);
+    void receivedDebugTunnel(const QJsonObject& obj);
+    /** TCP 已连上且 HELLO 帧已写入套接字（之后由核心下发 TREE_RESULT 等）。 */
+    void uplinkReady();
+    /** 与核心 TCP 断开或连接失败进入待机重连。 */
+    void uplinkLost();
 
 private slots:
     void onSocketReadyRead();
+    void onTcpConnected();
+    void onTcpDisconnected();
+    void onTcpError(QAbstractSocket::SocketError socketError);
 
 private:
+    void appendDebugFromHost(const QJsonObject& envelope);
+    void kickConnectInternal();
+    void planReconnect();
+    void flushPendingSbupFrames();
+
     QTcpSocket m_sock;
     QByteArray m_inbuf;
     int m_maxPayload{4 * 1024 * 1024};
+    QVector<QJsonObject> m_debugFromHostRing;
+    static constexpr int kMaxDebugRing = 64;
+
+    QString m_uplinkHost;
+    int m_uplinkPort{0};
+    QString m_uplinkEdgeId;
+    QString m_uplinkAuthToken;
+    std::function<qint64()> m_uplinkTreeRevision;
+    QTimer* m_reconnectTimer{nullptr};
+    bool m_uplinkHooksInstalled{false};
+    bool m_uplinkShutdown{false};
+    QVector<QByteArray> m_pendingSbupFrames;
+    static constexpr int kMaxPendingSbupFrames = 512;
 };
 
 } // namespace softbus::message_bus::ingress

src/message_bus/ingress/EdgeTcpUplinkPort.cpp

@@ -6,6 +6,7 @@
 #include <QDebug>
 
 #include "message_bus/ingress/EdgeUplinkWire.h"
+#include "utils/AgentDebugNdjson5448.h"
 
 namespace softbus::message_bus::ingress
 {
@@ -42,7 +43,19 @@ bool EdgeTcpUplinkPort::push(softbus::core::models::RawBusMessage ctx)
     }
     const QByteArray outer =
         edge_uplink::encodeOuterFrame(edge_uplink::FrameKind::DataRawBus, inner);
-    return m_writer(outer);
+    const bool ok = m_writer(outer);
+    // #region agent log
+    if (ok || busPayload.size() >= 4) {
+        QJsonObject d;
+        d.insert(QStringLiteral("ok"), ok);
+        d.insert(QStringLiteral("deviceId"), ctx.deviceId);
+        d.insert(QStringLiteral("protocol"), static_cast<int>(ctx.protocol));
+        d.insert(QStringLiteral("busPayloadLen"), busPayload.size());
+        d.insert(QStringLiteral("outerLen"), outer.size());
+        agent_debug_5448::log("H-uplink", "EdgeTcpUplinkPort.cpp:push", "encode_data_rawbus", d);
+    }
+    // #endregion
+    return ok;
 }
 
 void EdgeTcpUplinkPort::reportError(const QString& endpoint, const QString& error)

src/message_bus/ingress/EdgeUplinkWire.h

@@ -27,6 +27,8 @@ enum class FrameKind : std::uint16_t
     TreePropose = 6,
     TreeResult = 7,
     Report = 8,
+    /** 双向 UTF-8 JSON 调试/监控（主机↔边缘，需已 HELLO）。见 softbus_daemon/docs/EDGE_UPLINK_PROTOCOL_v1.md §10。 */
+    DebugTunnel = 9,
 };
 
 inline constexpr std::size_t kOuterHeaderBytes = 16;

src/ops_ui_http.cpp

@@ -1,5 +1,6 @@
 #include "ops_ui_http.hpp"
 
+#include <QDebug>
 #include <QFile>
 #include <QHostAddress>
 #include <QJsonArray>
@@ -79,7 +80,7 @@ QByteArray buildUiPage(const QString& edgeId, const QJsonArray& discovered)
         "<label for=name>显示名称（可选）</label>"
         "<input id=name name=name type=text placeholder=\"例如 电表\">"
         "<button type=submit>发送 upsert PROPOSE</button></form>"
-        "<p><a href=/>本地树 JSON</a> · <a href=/discovered>发现 JSON</a> · <code>POST /propose</code> 仍可提交完整 JSON。</p>"
+        "<p><a href=/>本地树 JSON</a> · <a href=/discovered>发现 JSON</a> · <a href=/debug>调试通道</a> · <code>POST /propose</code> 仍可提交完整 JSON。</p>"
         "</body></html>")
                .arg(htmlEscape(edgeId), options);
     return html.toUtf8();
@@ -130,16 +131,23 @@ bool OpsUiHttp::start(quint16 port,
                       const QString& edgeId,
                       const std::function<qint64()>& getBaseTreeRevision,
                       const std::function<void(const QJsonObject&)>& onPropose,
-                      const std::function<QJsonArray()>& getDiscovered)
+                      const std::function<QJsonArray()>& getDiscovered,
+                      const std::function<bool(const QJsonObject&)>& sendDebugToCore,
+                      const std::function<QJsonArray()>& getDebugFromHostInbox,
+                      bool listenOnAllInterfaces)
 {
     m_treePath = localTreePath;
     m_edgeId = edgeId;
     m_getBaseTreeRevision = getBaseTreeRevision;
     m_onPropose = onPropose;
     m_getDiscovered = getDiscovered;
+    m_sendDebugToCore = sendDebugToCore;
+    m_getDebugFromHostInbox = getDebugFromHostInbox;
     m_tcp = new QTcpServer(this);
     connect(m_tcp, &QTcpServer::newConnection, this, &OpsUiHttp::onNewConnection);
-    if (!m_tcp->listen(QHostAddress::LocalHost, port)) {
+    const QHostAddress bindAddr = listenOnAllInterfaces ? QHostAddress::Any : QHostAddress::LocalHost;
+    if (!m_tcp->listen(bindAddr, port)) {
+        qWarning() << "OpsUiHttp: listen failed" << bindAddr << port << m_tcp->errorString();
         return false;
     }
     return true;
@@ -235,6 +243,50 @@ void OpsUiHttp::handleSocket(QTcpSocket* sock, const QByteArray& data)
         return;
     }
 
+    if (method == "GET" && path == "/debug_host") {
+        if (!m_getDebugFromHostInbox) {
+            const QJsonObject err{{QStringLiteral("error"), QStringLiteral("debug inbox not wired")}};
+            writeHttp(sock, 503, QJsonDocument(err).toJson(QJsonDocument::Compact), "application/json; charset=utf-8");
+            return;
+        }
+        const QJsonArray items = m_getDebugFromHostInbox();
+        const QByteArray out =
+            QJsonDocument(QJsonObject{{QStringLiteral("items"), items}}).toJson(QJsonDocument::Indented);
+        writeHttp(sock, 200, out, "application/json; charset=utf-8");
+        return;
+    }
+
+    if (method == "POST" && path == "/debug_edge") {
+        if (!m_sendDebugToCore) {
+            const QJsonObject err{{QStringLiteral("error"), QStringLiteral("debug uplink not wired")}};
+            writeHttp(sock, 503, QJsonDocument(err).toJson(QJsonDocument::Compact), "application/json; charset=utf-8");
+            return;
+        }
+        QJsonParseError pe;
+        const QJsonDocument doc = QJsonDocument::fromJson(body, &pe);
+        if (pe.error != QJsonParseError::NoError || !doc.isObject()) {
+            writeHttp(sock, 400, QByteArrayLiteral("invalid json (expect object)"), "text/plain; charset=utf-8");
+            return;
+        }
+        const bool ok = m_sendDebugToCore(doc.object());
+        const QJsonObject resp{{QStringLiteral("ok"), ok}};
+        writeHttp(sock, ok ? 200 : 502, QJsonDocument(resp).toJson(QJsonDocument::Compact), "application/json; charset=utf-8");
+        return;
+    }
+
+    if (method == "GET" && path == "/debug") {
+        const QByteArray html =
+            QByteArrayLiteral("<!doctype html><meta charset=utf-8><title>调试通道</title>"
+                              "<h1>主机 ↔ 边缘 DebugTunnel</h1>"
+                              "<ul>"
+                              "<li><code>GET /debug_host</code> — 最近从主机经 SBUP 下发的调试 JSON（环形缓冲）</li>"
+                              "<li><code>POST /debug_edge</code> — body 为 JSON 对象，经 SBUP 发往主机（需已 HELLO）</li>"
+                              "</ul>"
+                              "<p><a href=/debug_host>打开 debug_host JSON</a> · <a href=/ui>配置页</a></p>");
+        writeHttp(sock, 200, html, "text/html; charset=utf-8");
+        return;
+    }
+
     if (method == "GET" && path == "/ui") {
         QJsonArray arr;
         if (m_getDiscovered) {

src/ops_ui_http.hpp

@@ -22,7 +22,10 @@ public:
                const QString& edgeId,
                const std::function<qint64()>& getBaseTreeRevision,
                const std::function<void(const QJsonObject&)>& onPropose,
-               const std::function<QJsonArray()>& getDiscovered = {});
+               const std::function<QJsonArray()>& getDiscovered = {},
+               const std::function<bool(const QJsonObject&)>& sendDebugToCore = {},
+               const std::function<QJsonArray()>& getDebugFromHostInbox = {},
+               bool listenOnAllInterfaces = false);
 
 private slots:
     void onNewConnection();
@@ -36,4 +39,6 @@ private:
     std::function<qint64()> m_getBaseTreeRevision;
     std::function<void(const QJsonObject&)> m_onPropose;
     std::function<QJsonArray()> m_getDiscovered;
+    std::function<bool(const QJsonObject&)> m_sendDebugToCore;
+    std::function<QJsonArray()> m_getDebugFromHostInbox;
 };

src/pipeline/EdgeStreamPipeline.cpp

@@ -0,0 +1,399 @@
+#include "pipeline/EdgeStreamPipeline.h"
+
+#include "device_bus/tree/DeviceTreeModel.h"
+
+#include <atomic>
+#include <chrono>
+#include <thread>
+#include <algorithm>
+
+#include <QJsonObject>
+#include <QJsonValue>
+
+#include "core/memory/LockFreeQueue.h"
+#include "core/plugin_system/IMapperDependencies.h"
+#include "utils/logs/logging.h"
+#include "utils/AgentDebugNdjson5448.h"
+
+namespace
+{
+QString edgePayloadHexPreview(const softbus::core::models::RawBusMessage& ctx, int maxBytes)
+{
+    if (!ctx.payload.valid()) {
+        return QStringLiteral("(invalid)");
+    }
+    const std::uint8_t* p = ctx.payload.bytes();
+    const std::size_t n = ctx.payload.length;
+    if (!p || n == 0) {
+        return QStringLiteral("(empty)");
+    }
+    QString s;
+    const std::size_t lim = (std::min)(n, static_cast<std::size_t>(maxBytes));
+    for (std::size_t i = 0; i < lim; ++i) {
+        s += QStringLiteral("%1 ").arg(static_cast<unsigned>(p[i]), 2, 16, QLatin1Char('0'));
+    }
+    if (lim < n) {
+        s += QStringLiteral("…");
+    }
+    return s.trimmed();
+}
+} // namespace
+
+namespace softbus_edge
+{
+
+struct EdgeFramerEndpointState
+{
+    softbus::core::memory::LockFreeQueue<softbus::core::models::RawBusMessage> chunkQ;
+    std::atomic<bool> running{true};
+    std::thread th;
+    std::uint32_t endpointHash{0};
+    EdgeStreamPipeline* pipeline{nullptr};
+
+    explicit EdgeFramerEndpointState(std::size_t cap, std::uint32_t ep, EdgeStreamPipeline* pl)
+        : chunkQ(cap)
+        , endpointHash(ep)
+        , pipeline(pl)
+    {
+    }
+};
+
+static void edgeFramerThreadMain(std::shared_ptr<EdgeFramerEndpointState> st);
+
+EdgeStreamPipeline::EdgeStreamPipeline(Config cfg)
+    : m_cfg(std::move(cfg))
+    , m_treeModel(m_cfg.treeModel)
+{
+    if (!m_cfg.pool) {
+        m_lastError = QStringLiteral("memory_pool_null");
+        return;
+    }
+    m_pluginManager = std::make_shared<softbus::core::plugin_system::PluginManager>();
+    m_pluginHost = std::make_unique<softbus::core::plugin_system::PluginHost>(m_pluginManager);
+    m_passthroughFramer =
+        std::make_unique<softbus::message_bus::pipeline::stages::framers::PassthroughFramer>(m_cfg.pool);
+
+    for (const QString& path : m_cfg.protocolPluginPaths) {
+        if (path.trimmed().isEmpty()) {
+            continue;
+        }
+        QString err;
+        if (!m_pluginHost->loadPlugin(path, &err)) {
+            m_lastError = QStringLiteral("load_plugin_failed:%1:%2").arg(path, err);
+            return;
+        }
+    }
+    if (m_pluginManager && m_cfg.pool) {
+        const softbus::core::models::ProtocolType types[] = {
+            softbus::core::models::ProtocolType::MODBUS_RTU,
+            softbus::core::models::ProtocolType::MODBUS_ASCII,
+            softbus::core::models::ProtocolType::MODBUS_TCP,
+        };
+        for (softbus::core::models::ProtocolType t : types) {
+            if (auto fr = m_pluginManager->selectProtocolFramerPlugin(t)) {
+                fr->bindMemoryPool(m_cfg.pool);
+            }
+        }
+    }
+    m_ok = true;
+}
+
+EdgeStreamPipeline::~EdgeStreamPipeline()
+{
+    shutdownFramerWorkers();
+}
+
+void EdgeStreamPipeline::setFramedSink(std::function<bool(softbus::core::models::RawBusMessage)> sink)
+{
+    m_framedSink = std::move(sink);
+}
+
+void EdgeStreamPipeline::shutdownFramerWorkers()
+{
+    std::unordered_map<std::uint32_t, std::shared_ptr<EdgeFramerEndpointState>> workers;
+    {
+        std::lock_guard<std::mutex> lk(m_framerWorkerMutex);
+        workers.swap(m_framerWorkers);
+    }
+    for (auto& kv : workers) {
+        if (kv.second) {
+            kv.second->running.store(false, std::memory_order_release);
+        }
+    }
+    for (auto& kv : workers) {
+        if (kv.second && kv.second->th.joinable()) {
+            kv.second->th.join();
+        }
+    }
+}
+
+std::shared_ptr<EdgeFramerEndpointState> EdgeStreamPipeline::ensureFramerWorker(std::uint32_t endpointHash)
+{
+    std::lock_guard<std::mutex> lk(m_framerWorkerMutex);
+    const auto it = m_framerWorkers.find(endpointHash);
+    if (it != m_framerWorkers.end()) {
+        return it->second;
+    }
+    const std::size_t cap = m_cfg.edgeSerialChunkQueueCapacity > 0 ? m_cfg.edgeSerialChunkQueueCapacity : 4096;
+    auto st = std::make_shared<EdgeFramerEndpointState>(cap, endpointHash, this);
+    st->th = std::thread(edgeFramerThreadMain, st);
+    m_framerWorkers.emplace(endpointHash, st);
+    return st;
+}
+
+bool EdgeStreamPipeline::tryEnqueueStreamChunk(softbus::core::models::RawBusMessage chunk)
+{
+    if (!m_ok) {
+        return false;
+    }
+    if (!m_cfg.edgeUseEndpointFramerThreads) {
+        processFramerIngressFromWorker(std::move(chunk));
+        return true;
+    }
+    const std::shared_ptr<EdgeFramerEndpointState> st = ensureFramerWorker(chunk.endpointHash);
+    if (!st) {
+        return false;
+    }
+    return st->chunkQ.push(std::move(chunk));
+}
+
+void EdgeStreamPipeline::processFramerIngressFromWorker(softbus::core::models::RawBusMessage&& chunkIn)
+{
+    if (!m_ok) {
+        return;
+    }
+    const qint64 inChunkLen =
+        chunkIn.payload.valid() ? static_cast<qint64>(chunkIn.payload.length) : qint64{-1};
+    QString hexPreview;
+    if (chunkIn.payload.valid() && chunkIn.payload.length > 0) {
+        hexPreview = edgePayloadHexPreview(chunkIn, 48);
+    }
+    std::vector<softbus::core::models::RawBusMessage> framedOut;
+    dispatchFramer(chunkIn, framedOut);
+    // #region agent log
+    {
+        QJsonObject d;
+        d.insert(QStringLiteral("inChunkLen"), inChunkLen);
+        d.insert(QStringLiteral("framesOut"), static_cast<int>(framedOut.size()));
+        d.insert(QStringLiteral("protocol"), static_cast<int>(chunkIn.protocol));
+        d.insert(QStringLiteral("deviceId"), chunkIn.deviceId);
+        d.insert(QStringLiteral("endpointHash"), QString::number(chunkIn.endpointHash, 16));
+        if (!chunkIn.traceId.isEmpty()) {
+            d.insert(QStringLiteral("traceId"), chunkIn.traceId);
+        }
+        if (!hexPreview.isEmpty()) {
+            d.insert(QStringLiteral("hexPreview"), hexPreview);
+        }
+        if (m_cfg.pool) {
+            d.insert(QStringLiteral("poolInUse"), static_cast<qint64>(m_cfg.pool->inUseBlocks()));
+            d.insert(QStringLiteral("poolTotal"), static_cast<qint64>(m_cfg.pool->totalBlocks()));
+        }
+        agent_debug_5448::log("H-framer", "EdgeStreamPipeline.cpp:processFramer", "after_dispatch", d);
+    }
+    // #endregion
+    for (auto& framed : framedOut) {
+        auto doms = runParserMapper(framed);
+        attachEdgeContext(framed, doms);
+        // #region agent log
+        {
+            QJsonObject d;
+            d.insert(QStringLiteral("domCount"), static_cast<int>(doms.size()));
+            d.insert(QStringLiteral("frameLen"),
+                     framed.payload.valid() ? static_cast<qint64>(framed.payload.length) : qint64{-1});
+            d.insert(QStringLiteral("frameHex"), edgePayloadHexPreview(framed, 32));
+            agent_debug_5448::log("H-parse", "EdgeStreamPipeline.cpp:runParserMapper", "dom_result", d);
+        }
+        // #endregion
+        for (const auto& d : doms) {
+            LOG_INFO() << "EdgeStreamPipeline: parsed DOM messageId=" << d.messageId << "pointId=" << d.pointId
+                       << "metadataId=" << d.metadataId;
+        }
+        if (m_framedSink) {
+            const std::uint32_t epH = framed.endpointHash;
+            const qint64 frameLen =
+                framed.payload.valid() ? static_cast<qint64>(framed.payload.length) : qint64{-1};
+            const bool pushed = m_framedSink(std::move(framed));
+            // #region agent log
+            {
+                QJsonObject d;
+                d.insert(QStringLiteral("pushed"), pushed);
+                d.insert(QStringLiteral("frameLen"), frameLen);
+                d.insert(QStringLiteral("endpointHash"), QString::number(epH, 16));
+                agent_debug_5448::log("H-uplink", "EdgeStreamPipeline.cpp:framedSink", "data_rawbus_push", d);
+            }
+            // #endregion
+            if (!pushed) {
+                LOG_WARNING() << "EdgeStreamPipeline: framedSink push failed endpointHash=0x"
+                              << QString::number(static_cast<quint32>(epH), 16);
+            }
+        } else {
+            LOG_WARNING() << "EdgeStreamPipeline: framedSink not set, dropping framed frame";
+        }
+    }
+}
+
+static void edgeFramerThreadMain(std::shared_ptr<EdgeFramerEndpointState> st)
+{
+    while (st->running.load(std::memory_order_acquire)) {
+        softbus::core::models::RawBusMessage ch;
+        if (!st->chunkQ.pop(ch)) {
+            std::this_thread::sleep_for(std::chrono::milliseconds(1));
+            continue;
+        }
+        if (st->pipeline) {
+            st->pipeline->processFramerIngressFromWorker(std::move(ch));
+        }
+    }
+}
+
+void EdgeStreamPipeline::dispatchFramer(softbus::core::models::RawBusMessage& chunkIn,
+                                        std::vector<softbus::core::models::RawBusMessage>& completeFramesOut)
+{
+    if (m_pluginManager) {
+        auto plugin = m_pluginManager->selectProtocolFramerPlugin(chunkIn.protocol);
+        if (plugin) {
+            plugin->bindMemoryPool(m_cfg.pool);
+            const PluginSessionKey key{chunkIn.endpointHash, plugin->pluginId()};
+            std::lock_guard<std::mutex> lk(m_framerSessionMutex);
+            auto it = m_framerSessions.find(key);
+            if (it == m_framerSessions.end() || !it.value()) {
+                auto s = plugin->createFramerSession(chunkIn.endpointHash, QJsonObject{});
+                m_framerSessions.insert(
+                    key,
+                    s ? std::shared_ptr<softbus::core::plugin_system::IProtocolFramerSession>(std::move(s)) : nullptr);
+                it = m_framerSessions.find(key);
+            }
+            if (it != m_framerSessions.end() && it.value()) {
+                it.value()->feedChunk(chunkIn, completeFramesOut);
+                return;
+            }
+        }
+    }
+    if (m_passthroughFramer) {
+        m_passthroughFramer->feed(chunkIn, completeFramesOut);
+    }
+}
+
+std::vector<softbus::core::models::DOMMessage> EdgeStreamPipeline::runParserMapper(
+    const softbus::core::models::RawBusMessage& ctx)
+{
+    std::vector<softbus::core::models::DOMMessage> out;
+    if (!m_pluginManager) {
+        return out;
+    }
+    auto parserPlugin = m_pluginManager->selectProtocolPlugin(ctx.protocol);
+    if (!parserPlugin) {
+        if (ctx.protocol != softbus::core::models::ProtocolType::UNKNOWN) {
+            LOG_WARNING() << "EdgeStreamPipeline: no parser plugin for protocol" << static_cast<int>(ctx.protocol)
+                          << "endpointHash=0x" << QString::number(ctx.endpointHash, 16) << "payloadLen="
+                          << (ctx.payload.valid() ? static_cast<qint64>(ctx.payload.length) : qint64{-1});
+        }
+        return out;
+    }
+    const PluginSessionKey parserKey{ctx.endpointHash, parserPlugin->pluginId()};
+    std::shared_ptr<softbus::core::plugin_system::IProtocolSession> parserSession;
+    {
+        std::lock_guard<std::mutex> lk(m_parserSessionMutex);
+        auto it = m_parserSessions.find(parserKey);
+        if (it == m_parserSessions.end() || !it.value()) {
+            auto s = parserPlugin->createSession();
+            m_parserSessions.insert(
+                parserKey,
+                s ? std::shared_ptr<softbus::core::plugin_system::IProtocolSession>(std::move(s)) : nullptr);
+            it = m_parserSessions.find(parserKey);
+        }
+        if (it == m_parserSessions.end() || !it.value()) {
+            LOG_WARNING() << "EdgeStreamPipeline: createParserSession failed for" << parserPlugin->pluginId();
+            return out;
+        }
+        parserSession = it.value();
+    }
+    softbus::message_bus::pipeline::protocol::ProtocolEnvelope envelope;
+    if (!parserSession->parse(ctx, envelope)) {
+        LOG_WARNING() << "EdgeStreamPipeline: parse failed protocol" << static_cast<int>(ctx.protocol) << "payloadLen="
+                      << (ctx.payload.valid() ? static_cast<qint64>(ctx.payload.length) : qint64{-1})
+                      << "hex=" << edgePayloadHexPreview(ctx, 24);
+        return out;
+    }
+
+    auto mapperPlugin = m_pluginManager->selectProtocolMapperPlugin(ctx.protocol);
+    if (!mapperPlugin) {
+        LOG_WARNING() << "EdgeStreamPipeline: no mapper plugin for protocol" << static_cast<int>(ctx.protocol);
+        return out;
+    }
+    const PluginSessionKey mapperKey{ctx.endpointHash, mapperPlugin->pluginId()};
+    std::shared_ptr<softbus::core::plugin_system::IProtocolMapperSession> mapperSession;
+    {
+        std::lock_guard<std::mutex> lk(m_mapperSessionMutex);
+        auto it = m_mapperSessions.find(mapperKey);
+        if (it == m_mapperSessions.end() || !it.value()) {
+            softbus::core::plugin_system::MapperSessionContext context;
+            context.mappingLookup = nullptr;
+            context.discoverySink = nullptr;
+            context.sniffingEnabled = []() { return true; };
+            auto s = mapperPlugin->createMapperSession(context);
+            m_mapperSessions.insert(
+                mapperKey,
+                s ? std::shared_ptr<softbus::core::plugin_system::IProtocolMapperSession>(std::move(s)) : nullptr);
+            it = m_mapperSessions.find(mapperKey);
+        }
+        if (it == m_mapperSessions.end() || !it.value()) {
+            LOG_WARNING() << "EdgeStreamPipeline: createMapperSession failed for" << mapperPlugin->pluginId();
+            return out;
+        }
+        mapperSession = it.value();
+    }
+    return mapperSession->map(ctx, envelope);
+}
+
+void EdgeStreamPipeline::attachEdgeContext(const softbus::core::models::RawBusMessage& raw,
+                                             std::vector<softbus::core::models::DOMMessage>& msgs) const
+{
+    if (msgs.empty()) {
+        return;
+    }
+    const DeviceTreeNode* node = nullptr;
+    if (m_treeModel && raw.deviceId >= 0) {
+        for (const auto& n : m_treeModel->nodes()) {
+            const int did = n.params.value(QStringLiteral("deviceId")).toInt(-999999);
+            if (did == raw.deviceId) {
+                node = &n;
+                break;
+            }
+        }
+    }
+    for (auto& msg : msgs) {
+        msg.attributes.insert(QStringLiteral("endpointHash"), static_cast<qint64>(raw.endpointHash));
+        if (raw.deviceId >= 0) {
+            msg.attributes.insert(QStringLiteral("runtimeDeviceId"), raw.deviceId);
+        }
+        if (node) {
+            msg.attributes.insert(QStringLiteral("deviceNodeId"), node->id);
+            msg.attributes.insert(QStringLiteral("deviceEndpoint"), node->endpoint);
+            const QJsonValue av = node->params.value(QStringLiteral("assetId"));
+            if (av.isString() && !av.toString().isEmpty()) {
+                msg.attributes.insert(QStringLiteral("assetId"), av.toString());
+                if (msg.objectType.isEmpty() || msg.objectType == QLatin1String("system")) {
+                    msg.objectType = QStringLiteral("asset");
+                    msg.objectRef = av.toString();
+                }
+            }
+        }
+        if (!msg.pointId.isEmpty()) {
+            msg.attributes.insert(QStringLiteral("pointId"), msg.pointId);
+        }
+        if (!msg.metadataId.isEmpty()) {
+            msg.attributes.insert(QStringLiteral("metadataId"), msg.metadataId);
+        }
+        if (msg.objectType.isEmpty()) {
+            msg.objectType = QStringLiteral("device");
+        }
+        if (msg.objectRef.isEmpty()) {
+            msg.objectRef = raw.deviceId >= 0 ? QString::number(raw.deviceId) : QStringLiteral("default");
+        }
+        msg.attributes.insert(QStringLiteral("objectType"), msg.objectType);
+        msg.attributes.insert(QStringLiteral("objectRef"), msg.objectRef);
+    }
+}
+
+} // namespace softbus_edge

src/pipeline/EdgeStreamPipeline.h

@@ -0,0 +1,119 @@
+#pragma once
+
+#include <cstddef>
+#include <cstdint>
+#include <functional>
+#include <memory>
+#include <mutex>
+#include <unordered_map>
+#include <vector>
+
+#include <QHash>
+#include <QString>
+#include <QStringList>
+
+#include "core/memory/MemoryPool.h"
+#include "core/models/DOMMessage.h"
+#include "core/models/RawBusMessage.h"
+#include "core/plugin_system/IProtocolFramerPlugin.h"
+#include "core/plugin_system/IProtocolMapperPlugin.h"
+#include "core/plugin_system/IProtocolPlugin.h"
+#include "core/plugin_system/PluginHost.h"
+#include "core/plugin_system/PluginManager.h"
+#include "message_bus/pipeline/stages/1_framers/PassthroughFramer.h"
+
+class DeviceTreeModel;
+
+namespace softbus_edge
+{
+
+struct EdgeFramerEndpointState;
+
+struct PluginSessionKey
+{
+    std::uint32_t endpointHash{0};
+    QString pluginId;
+    bool operator==(const PluginSessionKey& o) const
+    {
+        return endpointHash == o.endpointHash && pluginId == o.pluginId;
+    }
+};
+
+/**
+ * 与 softbus_daemon Pipeline 的 dispatchFramer + stage2Parser（解析+mapper）语义对齐；用于边缘串口流。
+ * 默认按 `endpointHash` 维护与 daemon `EndpointFramingState` 相同的 **SPSC LockFreeQueue + 独立分帧线程**，
+ * 串口回调仅入队；线程内完成 framer→parser→mapper 并经 `setFramedSink` 上行。
+ */
+class EdgeStreamPipeline final
+{
+public:
+    struct Config
+    {
+        std::shared_ptr<softbus::core::memory::MemoryPool> pool;
+        QStringList protocolPluginPaths;
+        /** 可选：用于在 DOM 上附加 deviceNodeId / assetId 等（轻量版 attachPesmRuntimeContext）。 */
+        DeviceTreeModel* treeModel{nullptr};
+        /** 与 daemon `EndpointFramingState::chunkQ` 容量一致的有界队列深度（SPSC）。 */
+        std::size_t edgeSerialChunkQueueCapacity{4096};
+        /** false 时在同线程立即处理（调试用）；true 时走队列+线程（默认）。 */
+        bool edgeUseEndpointFramerThreads{true};
+    };
+
+    explicit EdgeStreamPipeline(Config cfg);
+    ~EdgeStreamPipeline();
+
+    bool ok() const { return m_ok; }
+    QString lastError() const { return m_lastError; }
+
+    void setTreeModel(DeviceTreeModel* t) { m_treeModel = t; }
+
+    /** 完整帧经此回调上行（通常绑定 `EdgeTcpUplinkPort::push`）。须在首包入队前设置。 */
+    void setFramedSink(std::function<bool(softbus::core::models::RawBusMessage)> sink);
+
+    /**
+     * 将串口 read 组装的流式 chunk 入队（默认）或在同线程立即处理。
+     * @return false 表示队列已满（与 daemon `chunkQ.push` 失败一致）。
+     */
+    bool tryEnqueueStreamChunk(softbus::core::models::RawBusMessage chunk);
+
+    /** 分帧线程入口：framer → parser/mapper → framedSink（勿在串口回调线程直接调用）。 */
+    void processFramerIngressFromWorker(softbus::core::models::RawBusMessage&& chunkIn);
+
+private:
+    void shutdownFramerWorkers();
+    std::shared_ptr<EdgeFramerEndpointState> ensureFramerWorker(std::uint32_t endpointHash);
+
+    void dispatchFramer(softbus::core::models::RawBusMessage& chunkIn,
+                        std::vector<softbus::core::models::RawBusMessage>& completeFramesOut);
+    std::vector<softbus::core::models::DOMMessage> runParserMapper(const softbus::core::models::RawBusMessage& ctx);
+    void attachEdgeContext(const softbus::core::models::RawBusMessage& raw,
+                           std::vector<softbus::core::models::DOMMessage>& msgs) const;
+
+    bool m_ok{false};
+    QString m_lastError;
+    Config m_cfg;
+    DeviceTreeModel* m_treeModel{nullptr};
+    std::shared_ptr<softbus::core::plugin_system::PluginManager> m_pluginManager;
+    std::unique_ptr<softbus::core::plugin_system::PluginHost> m_pluginHost;
+    std::unique_ptr<softbus::message_bus::pipeline::stages::framers::PassthroughFramer> m_passthroughFramer;
+
+    std::mutex m_framerSessionMutex;
+    QHash<PluginSessionKey, std::shared_ptr<softbus::core::plugin_system::IProtocolFramerSession>> m_framerSessions;
+
+    std::mutex m_parserSessionMutex;
+    QHash<PluginSessionKey, std::shared_ptr<softbus::core::plugin_system::IProtocolSession>> m_parserSessions;
+
+    std::mutex m_mapperSessionMutex;
+    QHash<PluginSessionKey, std::shared_ptr<softbus::core::plugin_system::IProtocolMapperSession>> m_mapperSessions;
+
+    std::function<bool(softbus::core::models::RawBusMessage)> m_framedSink;
+    mutable std::mutex m_framerWorkerMutex;
+    std::unordered_map<std::uint32_t, std::shared_ptr<EdgeFramerEndpointState>> m_framerWorkers;
+};
+
+inline size_t qHash(const PluginSessionKey& key, size_t seed = 0) noexcept
+{
+    return ::qHash(key.endpointHash, ::qHash(key.pluginId, seed));
+}
+
+} // namespace softbus_edge

src/utils/AgentDebugNdjson5448.h

@@ -0,0 +1,35 @@
+#pragma once
+
+#include <QDateTime>
+#include <QDir>
+#include <QFile>
+#include <QFileInfo>
+#include <QJsonDocument>
+#include <QJsonObject>
+#include <QString>
+
+// #region agent log
+namespace agent_debug_5448
+{
+inline void log(const char* hypothesisId, const char* location, const char* message, const QJsonObject& data)
+{
+    QJsonObject root;
+    root.insert(QStringLiteral("sessionId"), QStringLiteral("5448fc"));
+    root.insert(QStringLiteral("hypothesisId"), QString::fromUtf8(hypothesisId));
+    root.insert(QStringLiteral("location"), QString::fromUtf8(location));
+    root.insert(QStringLiteral("message"), QString::fromUtf8(message));
+    root.insert(QStringLiteral("data"), data);
+    root.insert(QStringLiteral("timestamp"), QDateTime::currentMSecsSinceEpoch());
+    const QString path =
+        QStringLiteral("/home/dt/myCode/bus/soft_bus/softbus_daemon/.cursor/debug-5448fc.log");
+    const QFileInfo fi(path);
+    QDir().mkpath(fi.absolutePath());
+    QFile f(path);
+    if (!f.open(QIODevice::Append | QIODevice::Text)) {
+        return;
+    }
+    f.write(QJsonDocument(root).toJson(QJsonDocument::Compact));
+    f.write("\n", 1);
+}
+} // namespace agent_debug_5448
+// #endregion