phase 2

.gitignore

@@ -87,6 +87,11 @@ qrc_*.cpp
 __pycache__/
 .cache/
 
+# ------------------------------------------------------------------------------
+# Runtime storage (softbus_storage)
+# ------------------------------------------------------------------------------
+/data/
+
 # ------------------------------------------------------------------------------
 # Logs & temp
 # ------------------------------------------------------------------------------

CMakeLists.txt

@@ -17,12 +17,19 @@ include(SoftbusThirdParty)
 include(SoftbusPackage)
 
 add_subdirectory(src/softbus_core)
+add_subdirectory(src/softbus_registry)
+add_subdirectory(src/softbus_bus)
 add_subdirectory(src/softbus_transport)
 add_subdirectory(src/softbus_api)
+add_subdirectory(src/softbus_command)
+add_subdirectory(src/softbus_pipeline)
+add_subdirectory(src/softbus_monitor)
+add_subdirectory(src/softbus_storage)
 add_subdirectory(src/softbus_opcua)
 add_subdirectory(src/plugins/protocol_modbus)
 add_subdirectory(src/plugins/protocol_canopen)
 add_subdirectory(src/softbus_edge)
 add_subdirectory(src/softbus_daemon)
+add_subdirectory(tools)
 
 message(STATUS "Softbus workspace configured for ${SOFTBUS_PLATFORM_NAME}")

README.md

@@ -58,7 +58,7 @@ build/lib/protocol_canopen.so # CANopen 协议插件
 ```bash
 ./build/bin/softbus_daemon \
   --config config/daemon_profile.json \
-  --model config/softbus_model.json
+  --point-table config/point_table.json
 ```
 
 命令行参数：
@@ -66,7 +66,8 @@ build/lib/protocol_canopen.so # CANopen 协议插件
 | 参数 | 默认值 | 说明 |
 |------|--------|------|
 | `--config <path>` | `config/daemon_profile.json` | 守护进程运行配置 |
-| `--model <path>` | `config/softbus_model.json` | 对象模型定义 |
+| `--point-table <path>` | `config/point_table.json` | 点表（含 ValveOpening 等） |
+| `--model <path>` | 同 `--point-table` | 兼容旧参数名 |
 | `--self-test` | — | 启动后自动下发一条 ExecuteCommand 写值测试 |
 
 守护进程启动后会：
@@ -111,7 +112,7 @@ opc.tcp://localhost:4840
 }
 ```
 
-`config/softbus_model.json` 定义站点/设备/点位层级，守护进程据此构建 OPC UA 地址空间。
+`config/point_table.json` 定义设备/点位/主题/OPC UA 映射，由 `softbus_registry` 加载并驱动 OPC UA 地址空间与控制闭环。
 
 ## 典型联调流程
 

config/edge.json

@@ -0,0 +1,9 @@
+{
+  "edge": {
+    "deviceId": "EdgeDevice_01",
+    "daemonHost": "127.0.0.1",
+    "daemonPort": 9000,
+    "heartbeatPeriodMs": 1000,
+    "pointTable": "config/point_table.json"
+  }
+}

config/point_table.json

@@ -0,0 +1,46 @@
+{
+  "site": "DemoSite",
+  "system": "DemoSystem",
+  "asset": "DemoAsset",
+  "devices": [
+    {
+      "deviceId": "EdgeDevice_01",
+      "stableDeviceKey": "EdgeDevice_01",
+      "protocol": "modbus",
+      "points": [
+        {
+          "pointId": "MotorTemperature",
+          "name": "MotorTemperature",
+          "displayName": "电机温度",
+          "signalType": "AI",
+          "physicalChannel": "AI_Channel_01",
+          "register": 40001,
+          "unit": "℃",
+          "scale": 0.1,
+          "range": [0, 200],
+          "access": "read",
+          "writable": false,
+          "opcuaNodeId": "ns=1;s=DemoSite/DemoSystem/DemoAsset/EdgeDevice_01/MotorTemperature",
+          "dataTopic": "data/EdgeDevice_01/MotorTemperature"
+        },
+        {
+          "pointId": "ValveOpening",
+          "name": "ValveOpening",
+          "displayName": "阀门开度",
+          "signalType": "AO",
+          "physicalChannel": "AO_Channel_01",
+          "register": 40002,
+          "unit": "%",
+          "scale": 1.0,
+          "range": [0, 100],
+          "access": "read_write",
+          "writable": true,
+          "opcuaNodeId": "ns=1;s=DemoSite/DemoSystem/DemoAsset/EdgeDevice_01/ValveOpening",
+          "cmdTopic": "control/cmd/EdgeDevice_01/ValveOpening/set",
+          "ackTopic": "control/ack/EdgeDevice_01/ValveOpening",
+          "dataTopic": "data/EdgeDevice_01/ValveOpening"
+        }
+      ]
+    }
+  ]
+}

config/point_table.yaml

@@ -0,0 +1,29 @@
+# 点表配置（运行时加载 point_table.json；本文件为设计文档镜像）
+site: DemoSite
+system: DemoSystem
+asset: DemoAsset
+devices:
+  - deviceId: EdgeDevice_01
+    stableDeviceKey: EdgeDevice_01
+    protocol: modbus
+    points:
+      - pointId: MotorTemperature
+        displayName: 电机温度
+        signalType: AI
+        physicalChannel: AI_Channel_01
+        register: 40001
+        unit: "℃"
+        range: [0, 200]
+        access: read
+        dataTopic: data/EdgeDevice_01/MotorTemperature
+      - pointId: ValveOpening
+        displayName: 阀门开度
+        signalType: AO
+        physicalChannel: AO_Channel_01
+        register: 40002
+        unit: "%"
+        range: [0, 100]
+        access: read_write
+        cmdTopic: control/cmd/EdgeDevice_01/ValveOpening/set
+        ackTopic: control/ack/EdgeDevice_01/ValveOpening
+        dataTopic: data/EdgeDevice_01/ValveOpening

config/topic_rule.json

@@ -0,0 +1,7 @@
+{
+  "data": "data/{deviceId}/{pointId}",
+  "command": "control/cmd/{deviceId}/{pointId}/set",
+  "ack": "control/ack/{deviceId}/{pointId}",
+  "event": "event/{deviceId}/{eventType}",
+  "heartbeat": "system/heartbeat/{deviceId}"
+}

config/topic_rule.yaml

@@ -0,0 +1,6 @@
+topicRules:
+  data: "data/{deviceId}/{pointId}"
+  command: "control/cmd/{deviceId}/{pointId}/set"
+  ack: "control/ack/{deviceId}/{pointId}"
+  event: "event/{deviceId}/{eventType}"
+  heartbeat: "system/heartbeat/{deviceId}"

docs/command_flow.md

@@ -0,0 +1,29 @@
+# 控制命令闭环
+
+## 下行链路
+
+```
+OPC UA Client / --self-test
+  → UaMethodBinder::ExecuteCommand
+  → CommandService::submit()
+  → registry 点表校验 + 范围检查
+  → bus.publish(control/cmd/{device}/{point}/set)
+  → TransportBusBridge → TcpUplinkServer → edge
+  → CommandExecutor → ShadowChannelStore (mock AO)
+  → SoftbusAck → bus.publish(control/ack/...)
+  → CommandService ACK 匹配 → Succeeded
+```
+
+## 上行反馈
+
+```
+ShadowChannelStore 更新
+  → edge 构造 RawPacket 上行
+  → PipelineEngine 解析 → UaModelBinder
+  → bus.publish(data/{device}/{point})
+  → OPC UA Variable 更新
+```
+
+## 命令状态
+
+`Created → Validated → Dispatched → Succeeded / Failed / Timeout`

docs/point_table_design.md

@@ -0,0 +1,20 @@
+# 点表设计
+
+配置文件：`config/point_table.json`（YAML 镜像：`config/point_table.yaml`）
+
+## 映射关系
+
+```
+ValveOpening
+  ↔ AO_Channel_01        (physicalChannel)
+  ↔ ns=1;s=.../ValveOpening (opcuaNodeId)
+  ↔ control/cmd/EdgeDevice_01/ValveOpening/set (cmdTopic)
+  ↔ control/ack/EdgeDevice_01/ValveOpening (ackTopic)
+  ↔ data/EdgeDevice_01/ValveOpening (dataTopic)
+```
+
+由 `softbus_registry::PointRegistry` 加载并建立三向索引：`objectRef` / `topic` / `opcuaNodeId`。
+
+## 主题规则
+
+`config/topic_rule.json` 定义默认主题模板，点表可覆盖单点 `cmdTopic`/`dataTopic`。

src/softbus_api/include/softbus_api/ExecuteCommand.h

@@ -6,6 +6,7 @@
 namespace softbus::api {
 
 struct ExecuteCommand {
+    std::string requestId;
     std::string traceId;
     std::string objectRef;
     std::string command;

src/softbus_api/src/ExecuteCommand.cpp

@@ -1,10 +1,13 @@
 #include <softbus_api/ExecuteCommand.h>
 
+#include <softbus_core/models/ObjectModel.h>
+
 namespace softbus::api {
 
 ExecuteCommand ExecuteCommand::fromJson(const nlohmann::json& j)
 {
     ExecuteCommand cmd;
+    cmd.requestId = j.value("requestId", "");
     cmd.traceId = j.value("traceId", "");
     cmd.objectRef = j.value("objectRef", "");
     cmd.command = j.value("command", "");
@@ -18,6 +21,7 @@ nlohmann::json ExecuteCommand::toJson() const
 {
     return nlohmann::json{
         {"type", "executeCommand"},
+        {"requestId", requestId},
         {"traceId", traceId},
         {"objectRef", objectRef},
         {"command", command},
@@ -45,6 +49,19 @@ bool ExecuteCommand::validate(std::string* error) const
         }
         return false;
     }
+    if (!params.is_object()) {
+        if (error) {
+            *error = "params must be a JSON object";
+        }
+        return false;
+    }
+    const auto ref = softbus::core::ObjectRef::fromPath(objectRef);
+    if (!ref.hasPoint()) {
+        if (error) {
+            *error = "objectRef must be a five-layer path (Site/System/Asset/Device/Point)";
+        }
+        return false;
+    }
     return true;
 }
 

src/softbus_bus/CMakeLists.txt

@@ -0,0 +1,8 @@
+softbus_package(softbus_bus)
+
+add_library(softbus_bus STATIC
+    src/Bus.cpp
+)
+
+softbus_export_include(softbus_bus "${CMAKE_CURRENT_SOURCE_DIR}/include")
+target_link_libraries(softbus_bus PUBLIC softbus_core softbus_api)

src/softbus_bus/include/softbus_bus/Bus.h

@@ -0,0 +1,41 @@
+// 消息总线
+// 负责消息的发布和订阅
+#pragma once
+ 
+#include <softbus_bus/BusMessage.h>
+
+#include <cstdint>
+#include <functional>
+#include <mutex>
+#include <string>
+#include <vector>
+
+namespace softbus::bus {
+
+using BusHandler = std::function<void(const BusMessage&)>;
+
+class Bus {
+public:
+    using SubscriptionId = std::uint64_t;
+
+    static Bus& instance();
+
+    SubscriptionId subscribe(const std::string& topicPattern, BusHandler handler);
+    void unsubscribe(SubscriptionId id);
+    void publish(const std::string& topic, const BusMessage& message);
+
+    static bool topicMatches(const std::string& pattern, const std::string& topic);
+
+private:
+    struct Subscription {
+        SubscriptionId id{0};
+        std::string pattern;
+        BusHandler handler;
+    };
+
+    std::mutex mutex_;
+    std::vector<Subscription> subscriptions_;
+    SubscriptionId nextId_{1};
+};
+
+} // namespace softbus::bus

src/softbus_bus/include/softbus_bus/BusMessage.h

@@ -0,0 +1,31 @@
+#pragma once
+
+#include <softbus_api/ExecuteCommand.h>
+#include <softbus_core/models/RawPacket.h>
+#include <softbus_core/models/SoftbusTypes.h>
+
+#include <string>
+
+namespace softbus::bus {
+
+enum class BusPayloadType {
+    DataPoint,
+    ExecuteCommand,
+    Ack,
+    Event,
+    Heartbeat,
+    RawPacket
+};
+
+struct BusMessage {
+    BusPayloadType type{BusPayloadType::DataPoint};
+    std::string topic;
+    softbus::core::SoftbusDataPoint dataPoint;
+    softbus::api::ExecuteCommand command;
+    softbus::core::SoftbusAck ack;
+    softbus::core::SoftbusEvent event;
+    softbus::core::RawPacket rawPacket;
+    std::string deviceId;
+};
+
+} // namespace softbus::bus

src/softbus_bus/src/Bus.cpp

@@ -0,0 +1,107 @@
+#include <softbus_bus/Bus.h>
+
+#include <algorithm>
+
+namespace softbus::bus {
+
+namespace {
+
+std::vector<std::string> splitTopic(const std::string& topic)
+{
+    std::vector<std::string> parts;
+    std::string current;
+    for (char c : topic) {
+        if (c == '/') {
+            if (!current.empty()) {
+                parts.push_back(current);
+                current.clear();
+            }
+        } else {
+            current.push_back(c);
+        }
+    }
+    if (!current.empty()) {
+        parts.push_back(current);
+    }
+    return parts;
+}
+
+} // namespace
+
+Bus& Bus::instance()
+{
+    static Bus bus;
+    return bus;
+}
+
+Bus::SubscriptionId Bus::subscribe(const std::string& topicPattern, BusHandler handler)
+{
+    std::lock_guard<std::mutex> lock(mutex_);
+    const SubscriptionId id = nextId_++;
+    subscriptions_.push_back(Subscription{id, topicPattern, std::move(handler)});
+    return id;
+}
+
+void Bus::unsubscribe(SubscriptionId id)
+{
+    std::lock_guard<std::mutex> lock(mutex_);
+    subscriptions_.erase(
+        std::remove_if(subscriptions_.begin(), subscriptions_.end(),
+            [id](const Subscription& s) { return s.id == id; }),
+        subscriptions_.end());
+}
+
+void Bus::publish(const std::string& topic, const BusMessage& message)
+{
+    std::vector<BusHandler> handlers;
+    {
+        std::lock_guard<std::mutex> lock(mutex_);
+        for (const auto& sub : subscriptions_) {
+            if (topicMatches(sub.pattern, topic)) {
+                handlers.push_back(sub.handler);
+            }
+        }
+    }
+    BusMessage msg = message;
+    msg.topic = topic;
+    for (const auto& handler : handlers) {
+        handler(msg);
+    }
+}
+
+bool Bus::topicMatches(const std::string& pattern, const std::string& topic)
+{
+    if (pattern == topic) {
+        return true;
+    }
+    if (pattern == "#" || pattern == "*") {
+        return true;
+    }
+
+    const auto patternParts = splitTopic(pattern);
+    const auto topicParts = splitTopic(topic);
+    std::size_t pi = 0;
+    std::size_t ti = 0;
+    while (pi < patternParts.size() && ti < topicParts.size()) {
+        const auto& p = patternParts[pi];
+        if (p == "#") {
+            return true;
+        }
+        if (p == "*" || p == "+") {
+            ++pi;
+            ++ti;
+            continue;
+        }
+        if (p != topicParts[ti]) {
+            return false;
+        }
+        ++pi;
+        ++ti;
+    }
+    if (pi == patternParts.size() && ti == topicParts.size()) {
+        return true;
+    }
+    return pi < patternParts.size() && patternParts[pi] == "#";
+}
+
+} // namespace softbus::bus

src/softbus_command/CMakeLists.txt

@@ -0,0 +1,9 @@
+softbus_package(softbus_command)
+
+add_library(softbus_command STATIC
+    src/CommandContext.cpp
+    src/CommandService.cpp
+)
+
+softbus_export_include(softbus_command "${CMAKE_CURRENT_SOURCE_DIR}/include")
+target_link_libraries(softbus_command PUBLIC softbus_core softbus_api softbus_bus softbus_registry)

src/softbus_command/include/softbus_command/CommandContext.h

@@ -0,0 +1,34 @@
+#pragma once
+
+#include <softbus_api/ExecuteCommand.h>
+
+#include <chrono>
+#include <string>
+
+namespace softbus::command {
+
+enum class CommandState {
+    Created,
+    Validated,
+    Dispatched,
+    Delivered,
+    Executing,
+    Succeeded,
+    Failed,
+    Timeout,
+    Rejected,
+    Cancelled
+};
+
+struct CommandContext {
+    std::string requestId;
+    softbus::api::ExecuteCommand command;
+    CommandState state{CommandState::Created};
+    std::string error;
+    double executedValue{0.0};
+    std::chrono::steady_clock::time_point createdAt{std::chrono::steady_clock::now()};
+};
+
+std::string commandStateToString(CommandState state);
+
+} // namespace softbus::command

src/softbus_command/include/softbus_command/CommandService.h

@@ -0,0 +1,39 @@
+#pragma once
+
+#include <softbus_api/ExecuteCommand.h>
+#include <softbus_bus/Bus.h>
+#include <softbus_command/CommandContext.h>
+#include <softbus_registry/PointRegistry.h>
+
+#include <chrono>
+#include <condition_variable>
+#include <mutex>
+#include <optional>
+#include <string>
+#include <unordered_map>
+
+namespace softbus::command {
+
+class CommandService {
+public:
+    explicit CommandService(softbus::registry::PointRegistry* registry,
+                            std::chrono::milliseconds timeout = std::chrono::milliseconds(5000));
+
+    void attachBus(softbus::bus::Bus* bus);
+    CommandContext submit(const softbus::api::ExecuteCommand& command, bool waitForAck = true);
+    std::optional<CommandContext> findByRequestId(const std::string& requestId) const;
+
+private:
+    void handleAck(const softbus::bus::BusMessage& message);
+    std::string generateRequestId() const;
+
+    softbus::registry::PointRegistry* registry_{nullptr};
+    softbus::bus::Bus* bus_{nullptr};
+    softbus::bus::Bus::SubscriptionId ackSub_{0};
+    std::chrono::milliseconds timeout_;
+    mutable std::mutex mutex_;
+    std::unordered_map<std::string, CommandContext> contexts_;
+    std::condition_variable ackCv_;
+};
+
+} // namespace softbus::command

src/softbus_command/src/CommandContext.cpp

@@ -0,0 +1,21 @@
+#include <softbus_command/CommandContext.h>
+
+namespace softbus::command {
+
+std::string commandStateToString(CommandState state)
+{
+    switch (state) {
+    case CommandState::Validated: return "Validated";
+    case CommandState::Dispatched: return "Dispatched";
+    case CommandState::Delivered: return "Delivered";
+    case CommandState::Executing: return "Executing";
+    case CommandState::Succeeded: return "Succeeded";
+    case CommandState::Failed: return "Failed";
+    case CommandState::Timeout: return "Timeout";
+    case CommandState::Rejected: return "Rejected";
+    case CommandState::Cancelled: return "Cancelled";
+    default: return "Created";
+    }
+}
+
+} // namespace softbus::command

src/softbus_command/src/CommandService.cpp

@@ -0,0 +1,163 @@
+#include <softbus_command/CommandService.h>
+
+#include <softbus_core/identity/DeviceIdentity.h>
+#include <softbus_core/logging/Logger.h>
+
+namespace softbus::command {
+
+CommandService::CommandService(softbus::registry::PointRegistry* registry,
+                               std::chrono::milliseconds timeout)
+    : registry_(registry)
+    , timeout_(timeout)
+{
+}
+
+void CommandService::attachBus(softbus::bus::Bus* bus)
+{
+    bus_ = bus;
+    if (!bus_) {
+        return;
+    }
+    ackSub_ = bus_->subscribe("control/ack/#", [this](const softbus::bus::BusMessage& msg) {
+        handleAck(msg);
+    });
+}
+
+CommandContext CommandService::submit(const softbus::api::ExecuteCommand& command, bool waitForAck)
+{
+    CommandContext ctx;
+    ctx.command = command;
+    ctx.requestId = command.requestId.empty()
+        ? generateRequestId()
+        : command.requestId;
+    ctx.command.requestId = ctx.requestId;
+
+    std::string error;
+    if (!ctx.command.validate(&error)) {
+        ctx.state = CommandState::Rejected;
+        ctx.error = error;
+        return ctx;
+    }
+
+    if (!registry_) {
+        ctx.state = CommandState::Rejected;
+        ctx.error = "registry not available";
+        return ctx;
+    }
+
+    const auto entry = registry_->findByObjectRef(ctx.command.objectRef);
+    if (!entry) {
+        ctx.state = CommandState::Rejected;
+        ctx.error = "unknown objectRef";
+        return ctx;
+    }
+
+    if (ctx.command.command == "write") {
+        const double value = ctx.command.params.value("value", 0.0);
+        if (!registry_->validateWriteValue(*entry, value, &error)) {
+            ctx.state = CommandState::Rejected;
+            ctx.error = error;
+            return ctx;
+        }
+    } else if (!entry->writable) {
+        ctx.state = CommandState::Rejected;
+        ctx.error = "point is not writable";
+        return ctx;
+    }
+
+    ctx.state = CommandState::Validated;
+    {
+        std::lock_guard<std::mutex> lock(mutex_);
+        contexts_[ctx.requestId] = ctx;
+    }
+
+    if (!bus_) {
+        ctx.state = CommandState::Failed;
+        ctx.error = "bus not attached";
+        return ctx;
+    }
+
+    const std::string topic = entry->cmdTopic.empty()
+        ? registry_->topicRules().formatCommand(entry->deviceId, entry->pointId)
+        : entry->cmdTopic;
+
+    softbus::bus::BusMessage msg;
+    msg.type = softbus::bus::BusPayloadType::ExecuteCommand;
+    msg.command = ctx.command;
+    bus_->publish(topic, msg);
+
+    {
+        std::lock_guard<std::mutex> lock(mutex_);
+        contexts_[ctx.requestId].state = CommandState::Dispatched;
+    }
+    ctx.state = CommandState::Dispatched;
+
+    SB_LOG_INFO_TRACE("CommandService", ctx.command.traceId,
+        "command dispatched requestId=", ctx.requestId, " topic=", topic);
+
+    if (!waitForAck) {
+        return ctx;
+    }
+
+    std::unique_lock<std::mutex> lock(mutex_);
+    const bool done = ackCv_.wait_for(lock, timeout_, [&]() {
+        const auto it = contexts_.find(ctx.requestId);
+        if (it == contexts_.end()) {
+            return true;
+        }
+        return it->second.state == CommandState::Succeeded
+            || it->second.state == CommandState::Failed
+            || it->second.state == CommandState::Timeout;
+    });
+
+    const auto it = contexts_.find(ctx.requestId);
+    if (it == contexts_.end()) {
+        ctx.state = CommandState::Failed;
+        ctx.error = "context lost";
+        return ctx;
+    }
+    if (!done && it->second.state == CommandState::Dispatched) {
+        it->second.state = CommandState::Timeout;
+        it->second.error = "ack timeout";
+    }
+    return it->second;
+}
+
+std::optional<CommandContext> CommandService::findByRequestId(const std::string& requestId) const
+{
+    std::lock_guard<std::mutex> lock(mutex_);
+    const auto it = contexts_.find(requestId);
+    if (it == contexts_.end()) {
+        return std::nullopt;
+    }
+    return it->second;
+}
+
+void CommandService::handleAck(const softbus::bus::BusMessage& message)
+{
+    const auto& ack = message.ack;
+    if (ack.requestId.empty()) {
+        return;
+    }
+
+    std::lock_guard<std::mutex> lock(mutex_);
+    const auto it = contexts_.find(ack.requestId);
+    if (it == contexts_.end()) {
+        return;
+    }
+    it->second.state = ack.success ? CommandState::Succeeded : CommandState::Failed;
+    it->second.error = ack.error;
+    it->second.executedValue = ack.executedValue;
+    ackCv_.notify_all();
+
+    SB_LOG_INFO_TRACE("CommandService", ack.traceId,
+        "ack matched requestId=", ack.requestId,
+        " state=", commandStateToString(it->second.state));
+}
+
+std::string CommandService::generateRequestId() const
+{
+    return "req-" + softbus::core::DeviceIdentity::generateTraceId();
+}
+
+} // namespace softbus::command

src/softbus_core/CMakeLists.txt

@@ -9,6 +9,7 @@ add_library(softbus_core STATIC
     src/platform/PlatformSignal.cpp
     src/identity/DeviceIdentity.cpp
     src/models/ObjectModel.cpp
+    src/models/SoftbusTypes.cpp
 )
 
 softbus_export_include(softbus_core "${CMAKE_CURRENT_SOURCE_DIR}/include")

src/softbus_core/include/softbus_core/models/ObjectModel.h

@@ -23,12 +23,21 @@ struct ObjectRef {
     std::string toPath() const;
     static ObjectRef fromPath(const std::string& path);
     bool isValid() const;
+    bool hasPoint() const;
 };
 
 struct PointDef {
     std::string name;
     int registerAddress{0};
     std::string domain{"Process"};
+    std::string signalType{"AI"};
+    std::string physicalChannel;
+    std::string unit;
+    double minValue{0.0};
+    double maxValue{100.0};
+    std::string access{"read"};
+    bool writable{false};
+    double scale{1.0};
 };
 
 struct DeviceDef {

src/softbus_core/include/softbus_core/models/SoftbusTypes.h

@@ -0,0 +1,73 @@
+#pragma once
+
+#include <cstdint>
+#include <nlohmann/json.hpp>
+#include <string>
+
+namespace softbus::core {
+
+enum class SoftbusQuality {
+    Good,
+    Bad,
+    Uncertain,
+    OutOfRange,
+    Stale
+};
+
+enum class SignalType {
+    AI,
+    AO,
+    DI,
+    DO,
+    Unknown
+};
+
+enum class PointAccess {
+    Read,
+    Write,
+    ReadWrite
+};
+
+struct SoftbusDataPoint {
+    std::string deviceId;
+    std::string pointId;
+    double value{0.0};
+    std::string unit;
+    SoftbusQuality quality{SoftbusQuality::Good};
+    std::uint64_t sourceTimestampNs{0};
+    std::uint64_t sequence{0};
+    std::string objectRef;
+
+    nlohmann::json toJson() const;
+    static SoftbusDataPoint fromJson(const nlohmann::json& j);
+};
+
+struct SoftbusAck {
+    std::string requestId;
+    std::string traceId;
+    std::string deviceId;
+    std::string pointId;
+    bool success{false};
+    std::string error;
+    double executedValue{0.0};
+
+    nlohmann::json toJson() const;
+    static SoftbusAck fromJson(const nlohmann::json& j);
+};
+
+struct SoftbusEvent {
+    std::string deviceId;
+    std::string eventType;
+    std::string message;
+    std::uint64_t timestampNs{0};
+
+    nlohmann::json toJson() const;
+    static SoftbusEvent fromJson(const nlohmann::json& j);
+};
+
+std::string qualityToString(SoftbusQuality q);
+SignalType signalTypeFromString(const std::string& s);
+std::string signalTypeToString(SignalType t);
+PointAccess accessFromString(const std::string& s);
+
+} // namespace softbus::core

src/softbus_core/src/models/ObjectModel.cpp

@@ -43,6 +43,11 @@ bool ObjectRef::isValid() const
     return !site.empty() && !system.empty() && !asset.empty() && !device.empty();
 }
 
+bool ObjectRef::hasPoint() const
+{
+    return isValid() && !point.empty();
+}
+
 bool ObjectModelTree::loadFromFile(const std::string& path)
 {
     std::ifstream ifs(path);
@@ -68,6 +73,22 @@ bool ObjectModelTree::loadFromFile(const std::string& path)
                 point.name = pt.value("name", "");
                 point.registerAddress = pt.value("register", 0);
                 point.domain = pt.value("domain", "Process");
+                point.signalType = pt.value("signalType", "AI");
+                point.physicalChannel = pt.value("physicalChannel", "");
+                point.unit = pt.value("unit", "");
+                point.scale = pt.value("scale", 1.0);
+                point.access = pt.value("access", "read");
+                point.writable = pt.value("writable", false);
+                if (pt.contains("range") && pt["range"].is_array() && pt["range"].size() >= 2) {
+                    point.minValue = pt["range"][0].get<double>();
+                    point.maxValue = pt["range"][1].get<double>();
+                } else {
+                    point.minValue = pt.value("min", 0.0);
+                    point.maxValue = pt.value("max", 100.0);
+                }
+                if (point.access == "read_write" || point.access == "write") {
+                    point.writable = true;
+                }
                 device.points.push_back(std::move(point));
             }
         }

src/softbus_core/src/models/SoftbusTypes.cpp

@@ -0,0 +1,128 @@
+#include <softbus_core/models/SoftbusTypes.h>
+
+namespace softbus::core {
+
+nlohmann::json SoftbusDataPoint::toJson() const
+{
+    return nlohmann::json{
+        {"deviceId", deviceId},
+        {"pointId", pointId},
+        {"value", value},
+        {"unit", unit},
+        {"quality", qualityToString(quality)},
+        {"sourceTimestamp", sourceTimestampNs},
+        {"sequence", sequence},
+        {"objectRef", objectRef}
+    };
+}
+
+SoftbusDataPoint SoftbusDataPoint::fromJson(const nlohmann::json& j)
+{
+    SoftbusDataPoint dp;
+    dp.deviceId = j.value("deviceId", "");
+    dp.pointId = j.value("pointId", "");
+    dp.value = j.value("value", 0.0);
+    dp.unit = j.value("unit", "");
+    dp.sourceTimestampNs = j.value("sourceTimestamp", static_cast<std::uint64_t>(0));
+    dp.sequence = j.value("sequence", static_cast<std::uint64_t>(0));
+    dp.objectRef = j.value("objectRef", "");
+    const auto q = j.value("quality", "GOOD");
+    if (q == "BAD") {
+        dp.quality = SoftbusQuality::Bad;
+    } else if (q == "OUT_OF_RANGE") {
+        dp.quality = SoftbusQuality::OutOfRange;
+    } else if (q == "STALE") {
+        dp.quality = SoftbusQuality::Stale;
+    } else if (q == "UNCERTAIN") {
+        dp.quality = SoftbusQuality::Uncertain;
+    }
+    return dp;
+}
+
+nlohmann::json SoftbusAck::toJson() const
+{
+    return nlohmann::json{
+        {"type", "ack"},
+        {"requestId", requestId},
+        {"traceId", traceId},
+        {"deviceId", deviceId},
+        {"pointId", pointId},
+        {"success", success},
+        {"error", error},
+        {"executedValue", executedValue}
+    };
+}
+
+SoftbusAck SoftbusAck::fromJson(const nlohmann::json& j)
+{
+    SoftbusAck ack;
+    ack.requestId = j.value("requestId", "");
+    ack.traceId = j.value("traceId", "");
+    ack.deviceId = j.value("deviceId", "");
+    ack.pointId = j.value("pointId", "");
+    ack.success = j.value("success", false);
+    ack.error = j.value("error", "");
+    ack.executedValue = j.value("executedValue", 0.0);
+    return ack;
+}
+
+nlohmann::json SoftbusEvent::toJson() const
+{
+    return nlohmann::json{
+        {"deviceId", deviceId},
+        {"eventType", eventType},
+        {"message", message},
+        {"timestamp", timestampNs}
+    };
+}
+
+SoftbusEvent SoftbusEvent::fromJson(const nlohmann::json& j)
+{
+    SoftbusEvent ev;
+    ev.deviceId = j.value("deviceId", "");
+    ev.eventType = j.value("eventType", "");
+    ev.message = j.value("message", "");
+    ev.timestampNs = j.value("timestamp", static_cast<std::uint64_t>(0));
+    return ev;
+}
+
+std::string qualityToString(SoftbusQuality q)
+{
+    switch (q) {
+    case SoftbusQuality::Bad: return "BAD";
+    case SoftbusQuality::Uncertain: return "UNCERTAIN";
+    case SoftbusQuality::OutOfRange: return "OUT_OF_RANGE";
+    case SoftbusQuality::Stale: return "STALE";
+    default: return "GOOD";
+    }
+}
+
+SignalType signalTypeFromString(const std::string& s)
+{
+    if (s == "AI") return SignalType::AI;
+    if (s == "AO") return SignalType::AO;
+    if (s == "DI") return SignalType::DI;
+    if (s == "DO") return SignalType::DO;
+    return SignalType::Unknown;
+}
+
+std::string signalTypeToString(SignalType t)
+{
+    switch (t) {
+    case SignalType::AI: return "AI";
+    case SignalType::AO: return "AO";
+    case SignalType::DI: return "DI";
+    case SignalType::DO: return "DO";
+    default: return "Unknown";
+    }
+}
+
+PointAccess accessFromString(const std::string& s)
+{
+    if (s == "read") return PointAccess::Read;
+    if (s == "write") return PointAccess::Write;
+    if (s == "read_write") return PointAccess::ReadWrite;
+    return PointAccess::Read;
+}
+
+} // namespace softbus::core

src/softbus_daemon/CMakeLists.txt

@@ -3,10 +3,6 @@ softbus_package(softbus_daemon)
 add_executable(softbus_daemon
     src/main.cpp
     src/CoreService.cpp
-    src/PipelineEngine.cpp
-    src/MetadataRegistry.cpp
-    src/ObjectModelRegistry.cpp
-    src/CommandDispatcher.cpp
     src/IpcOptionalDBus.cpp
 )
 
@@ -14,6 +10,12 @@ softbus_export_include(softbus_daemon "${CMAKE_CURRENT_SOURCE_DIR}/include")
 
 target_link_libraries(softbus_daemon PRIVATE
     softbus_core
+    softbus_registry
+    softbus_bus
+    softbus_command
+    softbus_pipeline
+    softbus_monitor
+    softbus_storage
     softbus_transport
     softbus_api
     softbus_opcua

src/softbus_daemon/src/CoreService.cpp

@@ -1,14 +1,11 @@
 #include <softbus_daemon/core/CoreService.h>
 
-#include <softbus_daemon/command/CommandDispatcher.h>
+#include <softbus_command/CommandContext.h>
-#include <softbus_daemon/pipeline/PipelineEngine.h>
-
 #include <softbus_opcua/UaAddressSpaceBuilder.h>
 #include <softbus_opcua/UaEventBridge.h>
 #include <softbus_opcua/UaMethodBinder.h>
 #include <softbus_opcua/UaModelBinder.h>
 #include <softbus_opcua/UaServerEngine.h>
-#include <softbus_daemon/registry/ObjectModelRegistry.h>
 
 #include <softbus_core/identity/DeviceIdentity.h>
 #include <softbus_core/logging/Logger.h>
@@ -19,6 +16,7 @@
 #endif
 
 #include <filesystem>
+#include <nlohmann/json.hpp>
 #include <thread>
 #include <vector>
 
@@ -54,9 +52,27 @@ std::string findPluginInDir(const fs::path& dir, const std::string& pluginName)
     return {};
 }
 
+std::string commandResultJson(const softbus::command::CommandContext& ctx)
+{
+    nlohmann::json j;
+    j["success"] = ctx.state == softbus::command::CommandState::Succeeded;
+    j["requestId"] = ctx.requestId;
+    j["state"] = softbus::command::commandStateToString(ctx.state);
+    if (!ctx.error.empty()) {
+        j["error"] = ctx.error;
+    }
+    if (ctx.executedValue != 0.0) {
+        j["executedValue"] = ctx.executedValue;
+    }
+    return j.dump();
+}
+
 } // namespace
 
-CoreService::CoreService() = default;
+CoreService::CoreService()
+    : commandService_(&registry_)
+{
+}
 
 CoreService::~CoreService()
 {
@@ -84,11 +100,7 @@ std::string CoreService::resolvePluginPath(const std::string& pluginName,
         exeDir / ".." / "lib",
         cwd / "build" / "lib",
         fs::path("build") / "lib",
-        fs::path("build") / "bin",
-        cwd / "build" / "bin",
-        fs::path("lib"),
         cwd / "lib",
-        fs::path("bin"),
         cwd
     };
 
@@ -97,69 +109,93 @@ std::string CoreService::resolvePluginPath(const std::string& pluginName,
             return found;
         }
     }
-
+    return fs::absolute(profileDir / "lib" / pluginFileNames(pluginName).front()).string();
-    const auto fallbackNames = pluginFileNames(pluginName);
-    return fs::absolute(profileDir / "lib" / fallbackNames.front()).string();
 }
+//  主函数启动入口
+//  profilePath: 配置文件路径
+//  pointTablePath: 点表文件路径
+//  return: 是否启动成功
 
-bool CoreService::start(const std::string& profilePath, const std::string& modelPath)
+bool CoreService::start(const std::string& profilePath, const std::string& pointTablePath)
 {
+    // 加载配置文件 
     if (!profile_.loadFromFile(profilePath)) {
         SB_LOG_ERROR("CoreService", "Failed to load profile: ", profilePath);
         return false;
     }
 
-    objectRegistry_ = std::make_unique<ObjectModelRegistry>();
+    const fs::path configDir = fs::path(pointTablePath).parent_path();
-    if (!objectRegistry_->loadFromFile(modelPath)) {
+    // 加载主题规则文件
-        SB_LOG_ERROR("CoreService", "Failed to load model: ", modelPath);
+    registry_.loadTopicRules((configDir / "topic_rule.json").string());
+    if (!registry_.loadPointTable(pointTablePath)) {
+        SB_LOG_ERROR("CoreService", "Failed to load point table: ", pointTablePath);
         return false;
     }
-    model_ = objectRegistry_->tree();
+    model_ = registry_.tree();
 
     const std::string pluginPath = resolvePluginPath(profile_.plugins.modbus, profilePath);
     if (!modbusPlugin_.load(pluginPath)) {
         SB_LOG_ERROR("CoreService", "Failed to load modbus plugin: ", pluginPath);
         return false;
     }
+    // 加载点表文件
+    auto& bus = softbus::bus::Bus::instance();
+    // 将消息总线绑定到命令服务、存储服务和心跳监控服务
+    commandService_.attachBus(&bus);
+    storageService_.attachBus(&bus);
+    heartbeatMonitor_.attachBus(&bus);
 
+    // 创建OPC UA安全配置
     softbus::opcua::SecurityConfig security;
     security.enableTls = profile_.security.tlsEnabled;
     security.certPath = profile_.security.certPath;
     security.keyPath = profile_.security.keyPath;
 
+    // 创建OPC UA服务引擎
     uaEngine_ = std::make_unique<softbus::opcua::UaServerEngine>();
+    // 启动OPC UA服务
     if (!uaEngine_->start(profile_.opcua.port, profile_.opcua.applicationName, security)) {
         return false;
     }
 
+    // 创建OPC UA模型绑定器
     uaBinder_ = std::make_unique<softbus::opcua::UaModelBinder>(uaEngine_->nativeServer());
+    // 创建OPC UA方法绑定器
     uaMethodBinder_ = std::make_unique<softbus::opcua::UaMethodBinder>(uaEngine_->nativeServer());
+    // 创建OPC UA事件桥接器
     uaEventBridge_ = std::make_unique<softbus::opcua::UaEventBridge>(uaEngine_->nativeServer());
+    // 将消息总线绑定到OPC UA模型绑定器、OPC UA方法绑定器和OPC UA事件桥接器
+    uaBinder_->attachBus(&bus);
+    uaEventBridge_->attachBus(&bus);
 
+    // 创建OPC UA地址空间构建器
     softbus::opcua::UaAddressSpaceBuilder builder(uaEngine_->nativeServer());
+    // 构建OPC UA地址空间
     if (!builder.buildFromModel(model_, uaBinder_.get())) {
         SB_LOG_ERROR("CoreService", "Failed to build OPC UA address space");
         return false;
     }
 
-    pipeline_ = std::make_unique<PipelineEngine>(&modbusPlugin_, uaBinder_.get());
+    // 创建数据处理管道
+    pipeline_ = std::make_unique<softbus::pipeline::PipelineEngine>(
+        &modbusPlugin_, uaBinder_.get(), &registry_);
+    pipeline_->attachBus(&bus);
+    // 创建TCP上行监听服务器
     uplink_ = std::make_unique<softbus::transport::TcpUplinkServer>(
         io_, profile_.uplink.listenHost, profile_.uplink.listenPort);
-    commandDispatcher_ = std::make_unique<CommandDispatcher>(uplink_.get());
+    // 将消息总线绑定到TCP上行监听服务器
-
+    transportBridge_.attach(&bus, uplink_.get());
-    uplink_->setRawPacketHandler([this](const softbus::core::RawPacket& packet) {
+    // 绑定OPC UA方法执行命令回调
-        SB_LOG_INFO_TRACE("CoreService", packet.traceId, "RawPacket received");
+    // 将OPC UA方法执行命令回调绑定到消息总线
-        pipeline_->process(packet);
-    });
-
     uaMethodBinder_->bindExecuteCommand([this](const softbus::api::ExecuteCommand& cmd) {
-        return commandDispatcher_->dispatch(cmd);
+        const auto ctx = commandService_.submit(cmd, true);
+        return commandResultJson(ctx);
     });
     
     uplink_->start();
     ioThread_ = std::thread([this]() { io_.run(); });
     running_ = true;
-    SB_LOG_INFO("CoreService", "Daemon started");
+    SB_LOG_INFO("CoreService", "Daemon started (bus+registry+command pipeline)");
     return true;
 }
 
@@ -188,18 +224,24 @@ void CoreService::run(bool selfTest)
     softbus::core::PlatformSignal::install(nullptr);
 
     bool selfTestDone = false;
+    const auto selfTestStart = std::chrono::steady_clock::now();
 
     while (!softbus::core::PlatformSignal::shouldStop()) {
         if (uaEngine_) {
             uaEngine_->iterate();
         }
-        if (selfTest && !selfTestDone && commandDispatcher_) {
+        heartbeatMonitor_.tick();
+        if (selfTest && !selfTestDone
+            && std::chrono::steady_clock::now() - selfTestStart > std::chrono::seconds(3)) {
             softbus::api::ExecuteCommand cmd;
             cmd.traceId = softbus::core::DeviceIdentity::generateTraceId();
-            cmd.objectRef = "DemoSite/DemoSystem/DemoAsset/SimModbusDevice/Temperature";
+            cmd.objectRef = "DemoSite/DemoSystem/DemoAsset/EdgeDevice_01/ValveOpening";
             cmd.command = "write";
-            cmd.params = {{"value", 42.5}};
+            cmd.params = {{"value", 60.0}};
-            commandDispatcher_->dispatch(cmd);
+            const auto ctx = commandService_.submit(cmd, true);
+            SB_LOG_INFO("CoreService", "self-test command state=",
+                softbus::command::commandStateToString(ctx.state),
+                " executedValue=", ctx.executedValue);
             selfTestDone = true;
         }
         std::this_thread::sleep_for(std::chrono::milliseconds(10));

src/softbus_daemon/src/main.cpp

@@ -6,22 +6,22 @@
 int main(int argc, char* argv[])
 {
     std::string profilePath = "config/daemon_profile.json";
-    std::string modelPath = "config/softbus_model.json";
+    std::string pointTablePath = "config/point_table.json";
     bool selfTest = false;
 
     for (int i = 1; i < argc; ++i) {
         const std::string arg = argv[i];
         if (arg == "--config" && i + 1 < argc) {
             profilePath = argv[++i];
-        } else if (arg == "--model" && i + 1 < argc) {
+        } else if ((arg == "--point-table" || arg == "--model") && i + 1 < argc) {
-            modelPath = argv[++i];
+            pointTablePath = argv[++i];
         } else if (arg == "--self-test") {
             selfTest = true;
         }
     }
 
     softbus::daemon::CoreService service;
-    if (!service.start(profilePath, modelPath)) {
+    if (!service.start(profilePath, pointTablePath)) {
         std::cerr << "Failed to start softbus_daemon" << std::endl;
         return 1;
     }

src/softbus_edge/CMakeLists.txt

@@ -6,9 +6,18 @@ add_executable(softbus_edge
     src/UplinkClient.cpp
     src/DeviceFactory.cpp
     src/DeviceDiscovery.cpp
+    src/EdgeConfig.cpp
+    src/ShadowChannelStore.cpp
+    src/CommandExecutor.cpp
 )
 
 softbus_export_include(softbus_edge "${CMAKE_CURRENT_SOURCE_DIR}/include")
 
-target_link_libraries(softbus_edge PRIVATE softbus_core softbus_transport softbus_api asio)
+target_link_libraries(softbus_edge PRIVATE
+    softbus_core
+    softbus_registry
+    softbus_transport
+    softbus_api
+    asio
+)
 softbus_apply_platform_libs(softbus_edge)

src/softbus_edge/include/softbus_edge/config/EdgeConfig.h

@@ -0,0 +1,18 @@
+#pragma once
+
+#include <cstdint>
+#include <string>
+
+namespace softbus::edge {
+
+struct EdgeConfig {
+    std::string deviceId{"EdgeDevice_01"};
+    std::string daemonHost{"127.0.0.1"};
+    uint16_t daemonPort{9000};
+    int heartbeatPeriodMs{1000};
+    std::string pointTablePath{"config/point_table.json"};
+
+    bool loadFromFile(const std::string& path);
+};
+
+} // namespace softbus::edge

src/softbus_edge/include/softbus_edge/egress/CommandExecutor.h

@@ -0,0 +1,26 @@
+#pragma once
+
+#include <softbus_api/ExecuteCommand.h>
+#include <softbus_edge/egress/ShadowChannelStore.h>
+#include <softbus_registry/PointRegistry.h>
+#include <softbus_edge/egress/UplinkClient.h>
+
+namespace softbus::edge {
+
+class CommandExecutor {
+public:
+    CommandExecutor(softbus::registry::PointRegistry* registry,
+                    ShadowChannelStore* store,
+                    UplinkClient* uplink,
+                    std::string deviceId);
+
+    softbus::core::SoftbusAck execute(const softbus::api::ExecuteCommand& command);
+
+private:
+    softbus::registry::PointRegistry* registry_{nullptr};
+    ShadowChannelStore* store_{nullptr};
+    UplinkClient* uplink_{nullptr};
+    std::string deviceId_;
+};
+
+} // namespace softbus::edge

src/softbus_edge/include/softbus_edge/egress/ShadowChannelStore.h

@@ -0,0 +1,21 @@
+#pragma once
+
+#include <softbus_registry/PointTableEntry.h>
+
+#include <optional>
+#include <string>
+#include <unordered_map>
+
+namespace softbus::edge {
+
+class ShadowChannelStore {
+public:
+    bool write(const softbus::registry::PointTableEntry& entry, double engineeringValue);
+    std::optional<double> read(const std::string& physicalChannel) const;
+    int rawRegisterValue(const softbus::registry::PointTableEntry& entry, double engineeringValue) const;
+
+private:
+    std::unordered_map<std::string, double> channels_;
+};
+
+} // namespace softbus::edge

src/softbus_edge/include/softbus_edge/egress/UplinkClient.h

@@ -2,6 +2,7 @@
 
 #include <softbus_api/ExecuteCommand.h>
 #include <softbus_core/models/RawPacket.h>
+#include <softbus_core/models/SoftbusTypes.h>
 #include <softbus_transport/TcpUplinkClient.h>
 
 #include <functional>
@@ -18,6 +19,8 @@ public:
     bool connect();
     void disconnect();
     bool sendRawPacket(const softbus::core::RawPacket& packet);
+    bool sendAck(const softbus::core::SoftbusAck& ack);
+    bool sendHeartbeat(const std::string& deviceId);
     void setCommandHandler(std::function<void(const softbus::api::ExecuteCommand&)> handler);
     void poll();
 

src/softbus_edge/src/CommandExecutor.cpp

@@ -0,0 +1,69 @@
+#include <softbus_edge/egress/CommandExecutor.h>
+
+#include <softbus_core/logging/Logger.h>
+
+namespace softbus::edge {
+
+CommandExecutor::CommandExecutor(softbus::registry::PointRegistry* registry,
+                                 ShadowChannelStore* store,
+                                 UplinkClient* uplink,
+                                 std::string deviceId)
+    : registry_(registry)
+    , store_(store)
+    , uplink_(uplink)
+    , deviceId_(std::move(deviceId))
+{
+}
+
+softbus::core::SoftbusAck CommandExecutor::execute(const softbus::api::ExecuteCommand& command)
+{
+    softbus::core::SoftbusAck ack;
+    ack.requestId = command.requestId;
+    ack.traceId = command.traceId;
+
+    if (!registry_ || !store_) {
+        ack.success = false;
+        ack.error = "executor not initialized";
+        return ack;
+    }
+
+    const auto entry = registry_->findByObjectRef(command.objectRef);
+    if (!entry) {
+        ack.success = false;
+        ack.error = "unknown objectRef on edge";
+        return ack;
+    }
+
+    ack.deviceId = entry->deviceId;
+    ack.pointId = entry->pointId;
+
+    if (command.command != "write") {
+        ack.success = false;
+        ack.error = "unsupported command";
+        return ack;
+    }
+
+    const double value = command.params.value("value", 0.0);
+    std::string error;
+    if (!registry_->validateWriteValue(*entry, value, &error)) {
+        ack.success = false;
+        ack.error = error;
+        return ack;
+    }
+
+    store_->write(*entry, value);
+    ack.executedValue = value;
+    ack.success = true;
+
+    const int raw = store_->rawRegisterValue(*entry, value);
+    SB_LOG_INFO_TRACE("CommandExecutor", command.traceId,
+        "write ", entry->physicalChannel, " value=", value,
+        entry->unit, " raw=", raw, " (mock AO)");
+
+    if (uplink_) {
+        uplink_->sendAck(ack);
+    }
+    return ack;
+}
+
+} // namespace softbus::edge

src/softbus_edge/src/EdgeConfig.cpp

@@ -0,0 +1,28 @@
+#include <softbus_edge/config/EdgeConfig.h>
+
+#include <fstream>
+#include <nlohmann/json.hpp>
+
+namespace softbus::edge {
+
+bool EdgeConfig::loadFromFile(const std::string& path)
+{
+    std::ifstream ifs(path);
+    if (!ifs) {
+        return false;
+    }
+    nlohmann::json j;
+    ifs >> j;
+    if (!j.contains("edge")) {
+        return false;
+    }
+    const auto& edge = j["edge"];
+    deviceId = edge.value("deviceId", deviceId);
+    daemonHost = edge.value("daemonHost", daemonHost);
+    daemonPort = static_cast<uint16_t>(edge.value("daemonPort", static_cast<int>(daemonPort)));
+    heartbeatPeriodMs = edge.value("heartbeatPeriodMs", heartbeatPeriodMs);
+    pointTablePath = edge.value("pointTable", pointTablePath);
+    return true;
+}
+
+} // namespace softbus::edge

src/softbus_edge/src/ShadowChannelStore.cpp

@@ -0,0 +1,32 @@
+#include <softbus_edge/egress/ShadowChannelStore.h>
+
+#include <cmath>
+
+namespace softbus::edge {
+
+bool ShadowChannelStore::write(const softbus::registry::PointTableEntry& entry,
+                               double engineeringValue)
+{
+    channels_[entry.physicalChannel] = engineeringValue;
+    return true;
+}
+
+std::optional<double> ShadowChannelStore::read(const std::string& physicalChannel) const
+{
+    const auto it = channels_.find(physicalChannel);
+    if (it == channels_.end()) {
+        return std::nullopt;
+    }
+    return it->second;
+}
+
+int ShadowChannelStore::rawRegisterValue(const softbus::registry::PointTableEntry& entry,
+                                         double engineeringValue) const
+{
+    if (entry.scale <= 0.0) {
+        return static_cast<int>(std::lround(engineeringValue));
+    }
+    return static_cast<int>(std::lround(engineeringValue / entry.scale));
+}
+
+} // namespace softbus::edge

src/softbus_edge/src/UplinkClient.cpp

@@ -27,6 +27,16 @@ bool UplinkClient::sendRawPacket(const softbus::core::RawPacket& packet)
     return client_->sendRawPacket(packet);
 }
 
+bool UplinkClient::sendAck(const softbus::core::SoftbusAck& ack)
+{
+    return client_->sendAck(ack);
+}
+
+bool UplinkClient::sendHeartbeat(const std::string& deviceId)
+{
+    return client_->sendHeartbeat(deviceId);
+}
+
 void UplinkClient::setCommandHandler(std::function<void(const softbus::api::ExecuteCommand&)> handler)
 {
     client_->setCommandHandler(std::move(handler));

src/softbus_edge/src/main.cpp

@@ -1,11 +1,17 @@
+#include <softbus_edge/config/EdgeConfig.h>
+#include <softbus_edge/egress/CommandExecutor.h>
+#include <softbus_edge/egress/ShadowChannelStore.h>
 #include <softbus_edge/egress/UplinkClient.h>
-#include <softbus_edge/factory/DeviceFactory.h>
 #include <softbus_edge/ingress/MockModbusIngress.h>
 
+#include <softbus_core/identity/DeviceIdentity.h>
 #include <softbus_core/logging/Logger.h>
+#include <softbus_core/models/SoftbusTypes.h>
 #include <softbus_core/platform/PlatformSignal.h>
-
+#include <softbus_core/time/HwClock.h>
+#include <softbus_registry/PointRegistry.h>
 #include <chrono>
+#include <filesystem>
 #include <iostream>
 #include <string>
 #include <thread>
@@ -13,9 +19,8 @@
 namespace {
 
 struct EdgeOptions {
-    std::string daemonHost{"127.0.0.1"};
+    std::string configPath{"config/edge.json"};
-    uint16_t daemonPort{9000};
+    std::string objectRef{"DemoSite/DemoSystem/DemoAsset/EdgeDevice_01/MotorTemperature"};
-    std::string objectRef{"DemoSite/DemoSystem/DemoAsset/SimModbusDevice/Temperature"};
 };
 
 EdgeOptions parseArgs(int argc, char* argv[])
@@ -23,15 +28,8 @@ EdgeOptions parseArgs(int argc, char* argv[])
     EdgeOptions options;
     for (int i = 1; i < argc; ++i) {
         const std::string arg = argv[i];
-        if (arg == "--daemon" && i + 1 < argc) {
+        if (arg == "--config" && i + 1 < argc) {
-            const std::string endpoint = argv[++i];
+            options.configPath = argv[++i];
-            const auto pos = endpoint.find(':');
-            if (pos != std::string::npos) {
-                options.daemonHost = endpoint.substr(0, pos);
-                options.daemonPort = static_cast<uint16_t>(std::stoi(endpoint.substr(pos + 1)));
-            } else {
-                options.daemonHost = endpoint;
-            }
         } else if (arg == "--object-ref" && i + 1 < argc) {
             options.objectRef = argv[++i];
         }
@@ -45,36 +43,83 @@ int main(int argc, char* argv[])
 {
     const EdgeOptions options = parseArgs(argc, argv);
 
-    softbus::edge::DeviceFactory factory;
+    softbus::edge::EdgeConfig config;
-    const std::string runtimeDeviceId = factory.createRuntimeDeviceId();
+    if (!config.loadFromFile(options.configPath)) {
-    SB_LOG_INFO("Edge", "runtimeDeviceId=", runtimeDeviceId, " (local only, not persisted)");
+        std::cerr << "Failed to load edge config: " << options.configPath << std::endl;
+        return 1;
+    }
 
-    softbus::edge::UplinkClient uplink(options.daemonHost, options.daemonPort);
+    softbus::registry::PointRegistry registry;
-    uplink.setCommandHandler([](const softbus::api::ExecuteCommand& cmd) {
+    const auto configDir = std::filesystem::path(config.pointTablePath).parent_path();
+    registry.loadTopicRules((configDir / "topic_rule.json").string());
+    if (!registry.loadPointTable(config.pointTablePath)) {
+        std::cerr << "Failed to load point table: " << config.pointTablePath << std::endl;
+        return 1;
+    }
+
+    softbus::edge::ShadowChannelStore shadowStore;
+    softbus::edge::UplinkClient uplink(config.daemonHost, config.daemonPort);
+    softbus::edge::CommandExecutor executor(&registry, &shadowStore, &uplink, config.deviceId);
+    uplink.setCommandHandler([&executor](const softbus::api::ExecuteCommand& cmd) {
+        const auto ack = executor.execute(cmd);
+        if (ack.success) {
             SB_LOG_INFO_TRACE("Edge", cmd.traceId,
-            "executeCommand received: ", cmd.toJson().dump());
+                "command executed point=", ack.pointId, " value=", ack.executedValue);
+        } else {
+            SB_LOG_WARN_TRACE("Edge", cmd.traceId, "command failed: ", ack.error);
+        }
     });
 
     if (!uplink.connect()) {
         std::cerr << "Failed to connect to daemon at "
-                  << options.daemonHost << ":" << options.daemonPort << std::endl;
+                  << config.daemonHost << ":" << config.daemonPort << std::endl;
         return 1;
     }
 
-    softbus::edge::MockModbusIngress ingress(options.objectRef, runtimeDeviceId);
+    softbus::edge::MockModbusIngress ingress(options.objectRef, config.deviceId);
     ingress.open();
 
     softbus::core::PlatformSignal::install([]() {
         SB_LOG_INFO("Edge", "Shutdown requested");
     });
 
+    auto lastHeartbeat = std::chrono::steady_clock::now();
+
     while (!softbus::core::PlatformSignal::shouldStop()) {
         uplink.poll();
+
+        const auto now = std::chrono::steady_clock::now();
+        if (now - lastHeartbeat >= std::chrono::milliseconds(config.heartbeatPeriodMs)) {
+            uplink.sendHeartbeat(config.deviceId);
+            lastHeartbeat = now;
+        }
+
         for (const auto& packet : ingress.poll()) {
             SB_LOG_INFO_TRACE("Edge", packet.traceId,
                 "Uplink RawPacket objectRef=", packet.sourceId);
             uplink.sendRawPacket(packet);
         }
+
+        for (const auto& entry : registry.entries()) {
+            const auto value = shadowStore.read(entry.physicalChannel);
+            if (!value) {
+                continue;
+            }
+            softbus::core::RawPacket feedback;
+            feedback.traceId = softbus::core::DeviceIdentity::generateTraceId();
+            feedback.timestampNs = softbus::core::HwClock::nowNs();
+            feedback.sourceId = entry.objectRef;
+            feedback.protocolTag = "shadow";
+            const int raw = shadowStore.rawRegisterValue(entry, *value);
+            feedback.payload = {
+                static_cast<uint8_t>(entry.registerAddress >> 8),
+                static_cast<uint8_t>(entry.registerAddress & 0xFF),
+                static_cast<uint8_t>(raw >> 8),
+                static_cast<uint8_t>(raw & 0xFF)
+            };
+            uplink.sendRawPacket(feedback);
+        }
+
         std::this_thread::sleep_for(std::chrono::seconds(1));
     }
 

src/softbus_monitor/CMakeLists.txt

@@ -0,0 +1,8 @@
+softbus_package(softbus_monitor)
+
+add_library(softbus_monitor STATIC
+    src/HeartbeatMonitor.cpp
+)
+
+softbus_export_include(softbus_monitor "${CMAKE_CURRENT_SOURCE_DIR}/include")
+target_link_libraries(softbus_monitor PUBLIC softbus_core softbus_bus)

src/softbus_monitor/include/softbus_monitor/HeartbeatMonitor.h

@@ -0,0 +1,30 @@
+#pragma once
+
+#include <softbus_bus/Bus.h>
+
+#include <chrono>
+#include <mutex>
+#include <string>
+#include <unordered_map>
+
+namespace softbus::monitor {
+
+class HeartbeatMonitor {
+public:
+    explicit HeartbeatMonitor(std::chrono::milliseconds timeout = std::chrono::seconds(5));
+
+    void attachBus(softbus::bus::Bus* bus);
+    void tick();
+
+private:
+    void onHeartbeat(const softbus::bus::BusMessage& message);
+    void publishOfflineEvent(const std::string& deviceId);
+
+    softbus::bus::Bus* bus_{nullptr};
+    softbus::bus::Bus::SubscriptionId hbSub_{0};
+    std::chrono::milliseconds timeout_;
+    std::mutex mutex_;
+    std::unordered_map<std::string, std::chrono::steady_clock::time_point> lastSeen_;
+};
+
+} // namespace softbus::monitor

src/softbus_monitor/src/HeartbeatMonitor.cpp

@@ -0,0 +1,70 @@
+#include <softbus_monitor/HeartbeatMonitor.h>
+
+#include <softbus_core/logging/Logger.h>
+#include <softbus_core/time/HwClock.h>
+
+#include <vector>
+
+namespace softbus::monitor {
+
+HeartbeatMonitor::HeartbeatMonitor(std::chrono::milliseconds timeout)
+    : timeout_(timeout)
+{
+}
+
+void HeartbeatMonitor::attachBus(softbus::bus::Bus* bus)
+{
+    bus_ = bus;
+    if (!bus_) {
+        return;
+    }
+    hbSub_ = bus_->subscribe("system/heartbeat/#", [this](const softbus::bus::BusMessage& msg) {
+        onHeartbeat(msg);
+    });
+}
+
+void HeartbeatMonitor::onHeartbeat(const softbus::bus::BusMessage& message)
+{
+    if (message.deviceId.empty()) {
+        return;
+    }
+    std::lock_guard<std::mutex> lock(mutex_);
+    lastSeen_[message.deviceId] = std::chrono::steady_clock::now();
+}
+
+void HeartbeatMonitor::tick()
+{
+    const auto now = std::chrono::steady_clock::now();
+    std::vector<std::string> offline;
+    {
+        std::lock_guard<std::mutex> lock(mutex_);
+        for (const auto& [deviceId, last] : lastSeen_) {
+            if (now - last > timeout_) {
+                offline.push_back(deviceId);
+            }
+        }
+    }
+    for (const auto& deviceId : offline) {
+        publishOfflineEvent(deviceId);
+    }
+}
+
+void HeartbeatMonitor::publishOfflineEvent(const std::string& deviceId)
+{
+    if (!bus_) {
+        return;
+    }
+    softbus::core::SoftbusEvent ev;
+    ev.deviceId = deviceId;
+    ev.eventType = "edge_offline";
+    ev.message = "heartbeat timeout";
+    ev.timestampNs = softbus::core::HwClock::nowNs();
+
+    softbus::bus::BusMessage msg;
+    msg.type = softbus::bus::BusPayloadType::Event;
+    msg.event = ev;
+    bus_->publish("event/" + deviceId + "/edge_offline", msg);
+    SB_LOG_WARN("HeartbeatMonitor", "edge offline: ", deviceId);
+}
+
+} // namespace softbus::monitor

src/softbus_opcua/CMakeLists.txt

@@ -10,6 +10,6 @@ add_library(softbus_opcua STATIC
 
 softbus_export_include(softbus_opcua "${CMAKE_CURRENT_SOURCE_DIR}/include")
 
-target_link_libraries(softbus_opcua PUBLIC softbus_core softbus_api)
+target_link_libraries(softbus_opcua PUBLIC softbus_core softbus_api softbus_bus)
 softbus_apply_open62541(softbus_opcua)
 softbus_apply_platform_libs(softbus_opcua)

src/softbus_opcua/include/softbus_opcua/UaEventBridge.h

@@ -1,5 +1,7 @@
 #pragma once
 
+#include <softbus_bus/Bus.h>
+
 #include <string>
 
 struct UA_Server;
@@ -14,9 +16,14 @@ public:
     void emitDeviceOnline(const std::string& stableDeviceKey);
     void emitDeviceOffline(const std::string& stableDeviceKey);
     void emitParseError(const std::string& objectRef, const std::string& detail);
+    void attachBus(softbus::bus::Bus* bus);
 
 private:
+    void onBusMessage(const softbus::bus::BusMessage& message);
+
     UA_Server* server_{nullptr};
+    softbus::bus::Bus* bus_{nullptr};
+    softbus::bus::Bus::SubscriptionId eventSub_{0};
 };
 
 } // namespace softbus::opcua

src/softbus_opcua/include/softbus_opcua/UaMethodBinder.h

@@ -9,14 +9,14 @@ struct UA_Server;
 
 namespace softbus::opcua {
 
-using ExecuteCommandHandler = std::function<bool(const softbus::api::ExecuteCommand&)>;
+using ExecuteCommandHandler = std::function<std::string(const softbus::api::ExecuteCommand&)>;
 
 class UaMethodBinder {
 public:
     explicit UaMethodBinder(UA_Server* server);
 
     bool bindExecuteCommand(ExecuteCommandHandler handler);
-    bool dispatchCommand(const softbus::api::ExecuteCommand& command);
+    std::string dispatchCommand(const softbus::api::ExecuteCommand& command);
 
 private:
     UA_Server* server_{nullptr};

src/softbus_opcua/include/softbus_opcua/UaModelBinder.h

@@ -1,10 +1,13 @@
 #pragma once
 
+#include <softbus_bus/Bus.h>
 #include <softbus_core/models/DOMMessage.h>
 
 #include <map>
 #include <string>
 
+struct UA_Server;
+
 #if SOFTBUS_HAS_OPCUA
 #include <open62541/types.h>
 #else
@@ -18,10 +21,16 @@ public:
     explicit UaModelBinder(UA_Server* server);
 
     void registerPointNode(const std::string& objectRef, const UA_NodeId& nodeId);
+    void attachBus(softbus::bus::Bus* bus);
     bool bind(const softbus::core::DOMMessage& message);
+    bool bindValue(const std::string& objectRef, double value, const std::string& traceId = "");
 
 private:
+    void onBusMessage(const softbus::bus::BusMessage& message);
+
     UA_Server* server_{nullptr};
+    softbus::bus::Bus* bus_{nullptr};
+    softbus::bus::Bus::SubscriptionId dataSub_{0};
     std::map<std::string, UA_NodeId> pointNodes_;
 };
 

src/softbus_opcua/src/UaEventBridge.cpp

@@ -9,6 +9,25 @@ UaEventBridge::UaEventBridge(UA_Server* server)
 {
 }
 
+void UaEventBridge::attachBus(softbus::bus::Bus* bus)
+{
+    bus_ = bus;
+    if (!bus_) {
+        return;
+    }
+    eventSub_ = bus_->subscribe("event/#", [this](const softbus::bus::BusMessage& msg) {
+        onBusMessage(msg);
+    });
+}
+
+void UaEventBridge::onBusMessage(const softbus::bus::BusMessage& message)
+{
+    if (message.type != softbus::bus::BusPayloadType::Event) {
+        return;
+    }
+    emitDeviceOffline(message.event.deviceId + ": " + message.event.message);
+}
+
 void UaEventBridge::emitHeartbeat(const std::string& objectRef)
 {
     SB_LOG_DEBUG("UaEventBridge", "Heartbeat stub for ", objectRef);

src/softbus_opcua/src/UaMethodBinder.cpp

@@ -60,8 +60,7 @@ UA_StatusCode executeCommandMethod(UA_Server* server,
         return UA_STATUSCODE_GOOD;
     }
 
-    const bool ok = g_methodBinder->dispatchCommand(cmd);
+    const std::string result = g_methodBinder->dispatchCommand(cmd);
-    const std::string result = ok ? R"({"success":true})" : R"({"success":false})";
     UA_String out = UA_STRING_ALLOC(result.c_str());
     UA_Variant_setScalarCopy(output, &out, &UA_TYPES[UA_TYPES_STRING]);
     UA_String_clear(&out);
@@ -75,10 +74,10 @@ UaMethodBinder::UaMethodBinder(UA_Server* server)
 {
 }
 
-bool UaMethodBinder::dispatchCommand(const softbus::api::ExecuteCommand& command)
+std::string UaMethodBinder::dispatchCommand(const softbus::api::ExecuteCommand& command)
 {
     if (!handler_) {
-        return false;
+        return R"({"success":false,"error":"handler not bound"})";
     }
     SB_LOG_INFO_TRACE("UaMethodBinder", command.traceId,
         "ExecuteCommand invoked via OPC UA method");

src/softbus_opcua/src/UaModelBinder.cpp

@@ -13,6 +13,25 @@ UaModelBinder::UaModelBinder(UA_Server* server)
 {
 }
 
+void UaModelBinder::attachBus(softbus::bus::Bus* bus)
+{
+    bus_ = bus;
+    if (!bus_) {
+        return;
+    }
+    dataSub_ = bus_->subscribe("data/#", [this](const softbus::bus::BusMessage& msg) {
+        onBusMessage(msg);
+    });
+}
+
+void UaModelBinder::onBusMessage(const softbus::bus::BusMessage& message)
+{
+    if (message.type != softbus::bus::BusPayloadType::DataPoint) {
+        return;
+    }
+    bindValue(message.dataPoint.objectRef, message.dataPoint.value);
+}
+
 void UaModelBinder::registerPointNode(const std::string& objectRef, const UA_NodeId& nodeId)
 {
 #if SOFTBUS_HAS_OPCUA
@@ -27,43 +46,47 @@ void UaModelBinder::registerPointNode(const std::string& objectRef, const UA_Nod
 
 bool UaModelBinder::bind(const softbus::core::DOMMessage& message)
 {
-#if !SOFTBUS_HAS_OPCUA
-    (void)message;
-    return false;
-#else
-    if (!server_) {
-        return false;
-    }
-
     const std::string key = message.id.empty() ? message.sourceId : message.id;
-    const auto it = pointNodes_.find(key);
-    if (it == pointNodes_.end()) {
-        SB_LOG_WARN_TRACE("UaModelBinder", message.traceId,
-            "No OPC UA node registered for ", key);
-        return false;
-    }
-
     double numericValue = 0.0;
     if (message.value.contains("value")) {
         numericValue = message.value["value"].get<double>();
     } else if (message.value.contains("registerValue")) {
         numericValue = message.value["registerValue"].get<double>();
     }
+    return bindValue(key, numericValue, message.traceId);
+}
+
+bool UaModelBinder::bindValue(const std::string& objectRef, double value, const std::string& traceId)
+{
+#if !SOFTBUS_HAS_OPCUA
+    (void)objectRef;
+    (void)value;
+    (void)traceId;
+    return false;
+#else
+    if (!server_) {
+        return false;
+    }
+
+    const auto it = pointNodes_.find(objectRef);
+    if (it == pointNodes_.end()) {
+        SB_LOG_WARN_TRACE("UaModelBinder", traceId, "No OPC UA node registered for ", objectRef);
+        return false;
+    }
 
     UA_Variant variant;
     UA_Variant_init(&variant);
-    UA_Variant_setScalarCopy(&variant, &numericValue, &UA_TYPES[UA_TYPES_DOUBLE]);
+    UA_Variant_setScalarCopy(&variant, &value, &UA_TYPES[UA_TYPES_DOUBLE]);
     const UA_StatusCode status = UA_Server_writeValue(server_, it->second, variant);
     UA_Variant_clear(&variant);
 
     if (status == UA_STATUSCODE_GOOD) {
-        SB_LOG_INFO_TRACE("UaModelBinder", message.traceId,
+        SB_LOG_INFO_TRACE("UaModelBinder", traceId,
-            "Bound DOMMessage to OPC UA node ", key, " value=", numericValue);
+            "Bound value to OPC UA node ", objectRef, " value=", value);
         return true;
     }
 
-    SB_LOG_ERROR_TRACE("UaModelBinder", message.traceId,
+    SB_LOG_ERROR_TRACE("UaModelBinder", traceId, "Failed to write OPC UA node ", objectRef);
-        "Failed to write OPC UA node ", key);
     return false;
 #endif
 }

src/softbus_pipeline/CMakeLists.txt

@@ -0,0 +1,10 @@
+softbus_package(softbus_pipeline)
+
+add_library(softbus_pipeline STATIC
+    src/PipelineEngine.cpp
+)
+
+softbus_export_include(softbus_pipeline "${CMAKE_CURRENT_SOURCE_DIR}/include")
+target_link_libraries(softbus_pipeline PUBLIC
+    softbus_core softbus_bus softbus_registry softbus_opcua
+)

src/softbus_pipeline/include/softbus_pipeline/PipelineEngine.h

@@ -0,0 +1,31 @@
+#pragma once
+
+#include <softbus_bus/Bus.h>
+#include <softbus_core/plugin/PluginLoader.h>
+#include <softbus_opcua/UaModelBinder.h>
+#include <softbus_registry/PointRegistry.h>
+
+namespace softbus::pipeline {
+
+class PipelineEngine {
+public:
+    PipelineEngine(softbus::core::PluginLoader* plugin,
+                   softbus::opcua::UaModelBinder* binder,
+                   softbus::registry::PointRegistry* registry);
+
+    void attachBus(softbus::bus::Bus* bus);
+    void process(const softbus::core::RawPacket& packet);
+
+private:
+    void onBusMessage(const softbus::bus::BusMessage& message);
+    void publishDataPoint(const softbus::core::DOMMessage& message,
+                          const softbus::registry::PointTableEntry& entry);
+
+    softbus::core::PluginLoader* plugin_{nullptr};
+    softbus::opcua::UaModelBinder* binder_{nullptr};
+    softbus::registry::PointRegistry* registry_{nullptr};
+    softbus::bus::Bus* bus_{nullptr};
+    softbus::bus::Bus::SubscriptionId rawSub_{0};
+};
+
+} // namespace softbus::pipeline

src/softbus_pipeline/src/PipelineEngine.cpp

@@ -0,0 +1,108 @@
+#include <softbus_pipeline/PipelineEngine.h>
+
+#include <softbus_core/logging/Logger.h>
+#include <softbus_core/time/HwClock.h>
+
+#include <vector>
+
+namespace softbus::pipeline {
+
+PipelineEngine::PipelineEngine(softbus::core::PluginLoader* plugin,
+                               softbus::opcua::UaModelBinder* binder,
+                               softbus::registry::PointRegistry* registry)
+    : plugin_(plugin)
+    , binder_(binder)
+    , registry_(registry)
+{
+}
+
+void PipelineEngine::attachBus(softbus::bus::Bus* bus)
+{
+    bus_ = bus;
+    if (!bus_) {
+        return;
+    }
+    rawSub_ = bus_->subscribe("edge/raw/#", [this](const softbus::bus::BusMessage& msg) {
+        onBusMessage(msg);
+    });
+}
+
+void PipelineEngine::onBusMessage(const softbus::bus::BusMessage& message)
+{
+    if (message.type == softbus::bus::BusPayloadType::RawPacket) {
+        process(message.rawPacket);
+    }
+}
+
+void PipelineEngine::process(const softbus::core::RawPacket& packet)
+{
+    if (!plugin_ || !plugin_->parser() || !binder_) {
+        return;
+    }
+
+    const uint8_t* data = packet.payload.data();
+    std::size_t size = packet.payload.size();
+    std::vector<uint8_t> frame = packet.payload;
+
+    if (plugin_->framer()) {
+        const auto framed = plugin_->framer()->extract(data, size);
+        if (framed.success) {
+            frame = framed.frame;
+            data = frame.data();
+            size = frame.size();
+        }
+    }
+
+    auto parsed = plugin_->parser()->parseFrame(data, size, packet.sourceId);
+    if (!parsed.success) {
+        SB_LOG_WARN_TRACE("PipelineEngine", packet.traceId, "Failed to parse RawPacket");
+        return;
+    }
+
+    parsed.message.traceId = packet.traceId;
+    parsed.message.timestampNs = packet.timestampNs;
+
+    if (registry_) {
+        const auto entry = registry_->findByObjectRef(parsed.message.id);
+        if (entry) {
+            publishDataPoint(parsed.message, *entry);
+        }
+    }
+
+    binder_->bind(parsed.message);
+    SB_LOG_INFO_TRACE("PipelineEngine", packet.traceId,
+        "RawPacket converted to DOMMessage id=", parsed.message.id);
+}
+
+void PipelineEngine::publishDataPoint(const softbus::core::DOMMessage& message,
+                                      const softbus::registry::PointTableEntry& entry)
+{
+    if (!bus_) {
+        return;
+    }
+
+    double value = 0.0;
+    if (message.value.contains("value")) {
+        value = message.value["value"].get<double>();
+    } else if (message.value.contains("registerValue")) {
+        value = message.value["registerValue"].get<double>();
+    }
+
+    softbus::core::SoftbusDataPoint dp;
+    dp.deviceId = entry.deviceId;
+    dp.pointId = entry.pointId;
+    dp.value = value;
+    dp.unit = entry.unit;
+    dp.objectRef = entry.objectRef;
+    dp.sourceTimestampNs = message.timestampNs;
+
+    softbus::bus::BusMessage busMsg;
+    busMsg.type = softbus::bus::BusPayloadType::DataPoint;
+    busMsg.dataPoint = dp;
+    const std::string topic = entry.dataTopic.empty()
+        ? registry_->topicRules().formatData(entry.deviceId, entry.pointId)
+        : entry.dataTopic;
+    bus_->publish(topic, busMsg);
+}
+
+} // namespace softbus::pipeline

src/softbus_registry/CMakeLists.txt

@@ -0,0 +1,9 @@
+softbus_package(softbus_registry)
+
+add_library(softbus_registry STATIC
+    src/PointRegistry.cpp
+    src/PointTableEntry.cpp
+)
+
+softbus_export_include(softbus_registry "${CMAKE_CURRENT_SOURCE_DIR}/include")
+target_link_libraries(softbus_registry PUBLIC softbus_core nlohmann_json::nlohmann_json)

src/softbus_registry/include/softbus_registry/PointRegistry.h

@@ -0,0 +1,42 @@
+#pragma once
+
+#include <softbus_core/models/ObjectModel.h>
+#include <softbus_registry/PointTableEntry.h>
+
+#include <optional>
+#include <string>
+#include <unordered_map>
+#include <vector>
+
+namespace softbus::registry {
+
+class PointRegistry {
+public:
+    bool loadPointTable(const std::string& path);
+    bool loadTopicRules(const std::string& path);
+
+    const softbus::core::ObjectModelTree& tree() const { return tree_; }
+    const TopicRules& topicRules() const { return topicRules_; }
+    const std::vector<PointTableEntry>& entries() const { return entries_; }
+
+    std::optional<PointTableEntry> findByObjectRef(const std::string& objectRef) const;
+    std::optional<PointTableEntry> findByTopic(const std::string& topic) const;
+    std::optional<PointTableEntry> findByOpcuaNodeId(const std::string& nodeId) const;
+    std::optional<PointTableEntry> findByDevicePoint(const std::string& deviceId,
+                                                   const std::string& pointId) const;
+    std::optional<softbus::core::ObjectRef> findPointByRegister(int registerAddress) const;
+    bool validateWriteValue(const PointTableEntry& entry, double value, std::string* error) const;
+
+private:
+    void indexEntry(const PointTableEntry& entry);
+
+    softbus::core::ObjectModelTree tree_;
+    TopicRules topicRules_;
+    std::vector<PointTableEntry> entries_;
+    std::unordered_map<std::string, std::size_t> byObjectRef_;
+    std::unordered_map<std::string, std::size_t> byTopic_;
+    std::unordered_map<std::string, std::size_t> byOpcuaNodeId_;
+    std::unordered_map<std::string, std::size_t> byDevicePoint_;
+};
+
+} // namespace softbus::registry

src/softbus_registry/include/softbus_registry/PointTableEntry.h

@@ -0,0 +1,41 @@
+#pragma once
+
+#include <string>
+
+namespace softbus::registry {
+
+struct PointTableEntry {
+    std::string deviceId;
+    std::string pointId;
+    std::string name;
+    std::string displayName;
+    std::string objectRef;
+    std::string signalType;
+    std::string physicalChannel;
+    std::string unit;
+    double minValue{0.0};
+    double maxValue{100.0};
+    std::string access{"read"};
+    bool writable{false};
+    int registerAddress{0};
+    double scale{1.0};
+    std::string opcuaNodeId;
+    std::string cmdTopic;
+    std::string ackTopic;
+    std::string dataTopic;
+};
+
+struct TopicRules {
+    std::string data{"data/{deviceId}/{pointId}"};
+    std::string command{"control/cmd/{deviceId}/{pointId}/set"};
+    std::string ack{"control/ack/{deviceId}/{pointId}"};
+    std::string event{"event/{deviceId}/{eventType}"};
+    std::string heartbeat{"system/heartbeat/{deviceId}"};
+
+    std::string formatData(const std::string& deviceId, const std::string& pointId) const;
+    std::string formatCommand(const std::string& deviceId, const std::string& pointId) const;
+    std::string formatAck(const std::string& deviceId, const std::string& pointId) const;
+    std::string formatHeartbeat(const std::string& deviceId) const;
+};
+
+} // namespace softbus::registry

src/softbus_registry/src/PointRegistry.cpp

@@ -0,0 +1,199 @@
+#include <softbus_registry/PointRegistry.h>
+
+#include <softbus_core/identity/DeviceIdentity.h>
+
+#include <fstream>
+#include <nlohmann/json.hpp>
+
+namespace softbus::registry {
+
+bool PointRegistry::loadPointTable(const std::string& path)
+{
+    std::ifstream ifs(path);
+    if (!ifs) {
+        return false;
+    }
+    nlohmann::json j;
+    ifs >> j;
+
+    tree_.site = j.value("site", "");
+    tree_.system = j.value("system", "");
+    tree_.asset = j.value("asset", "");
+    tree_.devices.clear();
+    entries_.clear();
+    byObjectRef_.clear();
+    byTopic_.clear();
+    byOpcuaNodeId_.clear();
+    byDevicePoint_.clear();
+
+    if (!j.contains("devices")) {
+        return tree_.isValid();
+    }
+
+    for (const auto& dev : j["devices"]) {
+        softbus::core::DeviceDef device;
+        device.stableDeviceKey = dev.value("stableDeviceKey", dev.value("deviceId", ""));
+        device.protocol = dev.value("protocol", "modbus");
+        const std::string deviceId = dev.value("deviceId", device.stableDeviceKey);
+
+        if (!dev.contains("points")) {
+            tree_.devices.push_back(std::move(device));
+            continue;
+        }
+
+        for (const auto& pt : dev["points"]) {
+            softbus::core::PointDef pointDef;
+            pointDef.name = pt.value("name", pt.value("pointId", ""));
+            pointDef.registerAddress = pt.value("register", 0);
+            pointDef.domain = pt.value("domain", "Process");
+            pointDef.signalType = pt.value("signalType", "AI");
+            pointDef.physicalChannel = pt.value("physicalChannel", "");
+            pointDef.unit = pt.value("unit", "");
+            pointDef.scale = pt.value("scale", 1.0);
+            pointDef.access = pt.value("access", "read");
+            pointDef.writable = pt.value("writable", false);
+            if (pt.contains("range") && pt["range"].is_array() && pt["range"].size() >= 2) {
+                pointDef.minValue = pt["range"][0].get<double>();
+                pointDef.maxValue = pt["range"][1].get<double>();
+            }
+            if (pointDef.access == "read_write" || pointDef.access == "write") {
+                pointDef.writable = true;
+            }
+            device.points.push_back(pointDef);
+
+            PointTableEntry entry;
+            entry.deviceId = deviceId;
+            entry.pointId = pt.value("pointId", pointDef.name);
+            entry.name = pointDef.name;
+            entry.displayName = pt.value("displayName", entry.name);
+            entry.signalType = pointDef.signalType;
+            entry.physicalChannel = pointDef.physicalChannel;
+            entry.unit = pointDef.unit;
+            entry.minValue = pointDef.minValue;
+            entry.maxValue = pointDef.maxValue;
+            entry.access = pointDef.access;
+            entry.writable = pointDef.writable;
+            entry.registerAddress = pointDef.registerAddress;
+            entry.scale = pointDef.scale;
+
+            const auto objectRef = softbus::core::DeviceIdentity::buildObjectRef(
+                tree_, device.stableDeviceKey, entry.name);
+            entry.objectRef = objectRef.toPath();
+
+            entry.opcuaNodeId = pt.value(
+                "opcuaNodeId", "ns=1;s=" + entry.objectRef);
+            entry.dataTopic = pt.value(
+                "dataTopic", topicRules_.formatData(entry.deviceId, entry.pointId));
+            entry.cmdTopic = pt.value(
+                "cmdTopic", topicRules_.formatCommand(entry.deviceId, entry.pointId));
+            entry.ackTopic = pt.value(
+                "ackTopic", topicRules_.formatAck(entry.deviceId, entry.pointId));
+
+            entries_.push_back(entry);
+            indexEntry(entries_.back());
+        }
+        tree_.devices.push_back(std::move(device));
+    }
+    return tree_.isValid() && !entries_.empty();
+}
+
+bool PointRegistry::loadTopicRules(const std::string& path)
+{
+    std::ifstream ifs(path);
+    if (!ifs) {
+        return false;
+    }
+    nlohmann::json j;
+    ifs >> j;
+    topicRules_.data = j.value("data", topicRules_.data);
+    topicRules_.command = j.value("command", topicRules_.command);
+    topicRules_.ack = j.value("ack", topicRules_.ack);
+    topicRules_.event = j.value("event", topicRules_.event);
+    topicRules_.heartbeat = j.value("heartbeat", topicRules_.heartbeat);
+    return true;
+}
+
+void PointRegistry::indexEntry(const PointTableEntry& entry)
+{
+    const std::size_t idx = entries_.size() - 1;
+    byObjectRef_[entry.objectRef] = idx;
+    byOpcuaNodeId_[entry.opcuaNodeId] = idx;
+    byDevicePoint_[entry.deviceId + "/" + entry.pointId] = idx;
+    if (!entry.dataTopic.empty()) {
+        byTopic_[entry.dataTopic] = idx;
+    }
+    if (!entry.cmdTopic.empty()) {
+        byTopic_[entry.cmdTopic] = idx;
+    }
+    if (!entry.ackTopic.empty()) {
+        byTopic_[entry.ackTopic] = idx;
+    }
+}
+
+std::optional<PointTableEntry> PointRegistry::findByObjectRef(const std::string& objectRef) const
+{
+    const auto it = byObjectRef_.find(objectRef);
+    if (it == byObjectRef_.end()) {
+        return std::nullopt;
+    }
+    return entries_[it->second];
+}
+
+std::optional<PointTableEntry> PointRegistry::findByTopic(const std::string& topic) const
+{
+    const auto it = byTopic_.find(topic);
+    if (it == byTopic_.end()) {
+        return std::nullopt;
+    }
+    return entries_[it->second];
+}
+
+std::optional<PointTableEntry> PointRegistry::findByOpcuaNodeId(const std::string& nodeId) const
+{
+    const auto it = byOpcuaNodeId_.find(nodeId);
+    if (it == byOpcuaNodeId_.end()) {
+        return std::nullopt;
+    }
+    return entries_[it->second];
+}
+
+std::optional<PointTableEntry> PointRegistry::findByDevicePoint(const std::string& deviceId,
+                                                                const std::string& pointId) const
+{
+    const auto it = byDevicePoint_.find(deviceId + "/" + pointId);
+    if (it == byDevicePoint_.end()) {
+        return std::nullopt;
+    }
+    return entries_[it->second];
+}
+
+std::optional<softbus::core::ObjectRef> PointRegistry::findPointByRegister(int registerAddress) const
+{
+    for (const auto& entry : entries_) {
+        if (entry.registerAddress == registerAddress) {
+            return softbus::core::ObjectRef::fromPath(entry.objectRef);
+        }
+    }
+    return std::nullopt;
+}
+
+bool PointRegistry::validateWriteValue(const PointTableEntry& entry, double value,
+                                       std::string* error) const
+{
+    if (!entry.writable) {
+        if (error) {
+            *error = "point is not writable";
+        }
+        return false;
+    }
+    if (value < entry.minValue || value > entry.maxValue) {
+        if (error) {
+            *error = "value out of range [" + std::to_string(entry.minValue) + ","
+                + std::to_string(entry.maxValue) + "]";
+        }
+        return false;
+    }
+    return true;
+}
+
+} // namespace softbus::registry

src/softbus_registry/src/PointTableEntry.cpp

@@ -0,0 +1,45 @@
+#include <softbus_registry/PointTableEntry.h>
+
+namespace softbus::registry {
+
+namespace {
+
+std::string replaceToken(std::string pattern, const std::string& token, const std::string& value)
+{
+    const auto pos = pattern.find(token);
+    if (pos == std::string::npos) {
+        return pattern;
+    }
+    pattern.replace(pos, token.size(), value);
+    return pattern;
+}
+
+} // namespace
+
+std::string TopicRules::formatData(const std::string& deviceId, const std::string& pointId) const
+{
+    std::string topic = data;
+    topic = replaceToken(topic, "{deviceId}", deviceId);
+    return replaceToken(topic, "{pointId}", pointId);
+}
+
+std::string TopicRules::formatCommand(const std::string& deviceId, const std::string& pointId) const
+{
+    std::string topic = command;
+    topic = replaceToken(topic, "{deviceId}", deviceId);
+    return replaceToken(topic, "{pointId}", pointId);
+}
+
+std::string TopicRules::formatAck(const std::string& deviceId, const std::string& pointId) const
+{
+    std::string topic = ack;
+    topic = replaceToken(topic, "{deviceId}", deviceId);
+    return replaceToken(topic, "{pointId}", pointId);
+}
+
+std::string TopicRules::formatHeartbeat(const std::string& deviceId) const
+{
+    return replaceToken(heartbeat, "{deviceId}", deviceId);
+}
+
+} // namespace softbus::registry

src/softbus_storage/CMakeLists.txt

@@ -0,0 +1,8 @@
+softbus_package(softbus_storage)
+
+add_library(softbus_storage STATIC
+    src/StorageService.cpp
+)
+
+softbus_export_include(softbus_storage "${CMAKE_CURRENT_SOURCE_DIR}/include")
+target_link_libraries(softbus_storage PUBLIC softbus_core softbus_bus)

src/softbus_storage/include/softbus_storage/StorageService.h

@@ -0,0 +1,33 @@
+#pragma once
+
+#include <softbus_bus/Bus.h>
+
+#include <fstream>
+#include <mutex>
+#include <string>
+
+namespace softbus::storage {
+
+class StorageService {
+public:
+    explicit StorageService(std::string dataDir = "data/logs");
+
+    void attachBus(softbus::bus::Bus* bus);
+    bool open();
+
+private:
+    void onBusMessage(const softbus::bus::BusMessage& message);
+    void appendLine(const std::string& fileName, const std::string& line);
+
+    std::string dataDir_;
+    softbus::bus::Bus* bus_{nullptr};
+    softbus::bus::Bus::SubscriptionId dataSub_{0};
+    softbus::bus::Bus::SubscriptionId ackSub_{0};
+    softbus::bus::Bus::SubscriptionId eventSub_{0};
+    std::mutex mutex_;
+    std::ofstream timeseries_;
+    std::ofstream commandLog_;
+    std::ofstream eventLog_;
+};
+
+} // namespace softbus::storage

src/softbus_storage/src/StorageService.cpp

@@ -0,0 +1,70 @@
+#include <softbus_storage/StorageService.h>
+
+#include <filesystem>
+
+namespace softbus::storage {
+
+namespace fs = std::filesystem;
+
+StorageService::StorageService(std::string dataDir)
+    : dataDir_(std::move(dataDir))
+{
+}
+
+bool StorageService::open()
+{
+    fs::create_directories(dataDir_);
+    timeseries_.open(dataDir_ + "/timeseries.jsonl", std::ios::app);
+    commandLog_.open(dataDir_ + "/command_log.jsonl", std::ios::app);
+    eventLog_.open(dataDir_ + "/event_log.jsonl", std::ios::app);
+    return timeseries_.is_open() && commandLog_.is_open() && eventLog_.is_open();
+}
+
+void StorageService::attachBus(softbus::bus::Bus* bus)
+{
+    bus_ = bus;
+    if (!bus_) {
+        return;
+    }
+    open();
+    dataSub_ = bus_->subscribe("data/#", [this](const softbus::bus::BusMessage& msg) {
+        onBusMessage(msg);
+    });
+    ackSub_ = bus_->subscribe("control/ack/#", [this](const softbus::bus::BusMessage& msg) {
+        onBusMessage(msg);
+    });
+    eventSub_ = bus_->subscribe("event/#", [this](const softbus::bus::BusMessage& msg) {
+        onBusMessage(msg);
+    });
+}
+
+void StorageService::onBusMessage(const softbus::bus::BusMessage& message)
+{
+    switch (message.type) {
+    case softbus::bus::BusPayloadType::DataPoint:
+        appendLine("timeseries", message.dataPoint.toJson().dump());
+        break;
+    case softbus::bus::BusPayloadType::Ack:
+        appendLine("command", message.ack.toJson().dump());
+        break;
+    case softbus::bus::BusPayloadType::Event:
+        appendLine("event", message.event.toJson().dump());
+        break;
+    default:
+        break;
+    }
+}
+
+void StorageService::appendLine(const std::string& fileName, const std::string& line)
+{
+    std::lock_guard<std::mutex> lock(mutex_);
+    if (fileName == "timeseries" && timeseries_.is_open()) {
+        timeseries_ << line << '\n';
+    } else if (fileName == "command" && commandLog_.is_open()) {
+        commandLog_ << line << '\n';
+    } else if (fileName == "event" && eventLog_.is_open()) {
+        eventLog_ << line << '\n';
+    }
+}
+
+} // namespace softbus::storage

src/softbus_transport/CMakeLists.txt

@@ -4,9 +4,10 @@ add_library(softbus_transport STATIC
     src/UplinkWire.cpp
     src/TcpUplinkServer.cpp
     src/TcpUplinkClient.cpp
+    src/TransportBusBridge.cpp
 )
 
 softbus_export_include(softbus_transport "${CMAKE_CURRENT_SOURCE_DIR}/include")
 
-target_link_libraries(softbus_transport PUBLIC softbus_core softbus_api asio)
+target_link_libraries(softbus_transport PUBLIC softbus_core softbus_api softbus_bus asio)
 softbus_apply_platform_libs(softbus_transport)

src/softbus_transport/include/softbus_transport/TcpUplinkClient.h

@@ -22,7 +22,10 @@ public:
     void setCommandHandler(CommandHandler handler);
     bool connect();
     void disconnect();
+    bool sendCommand(const softbus::api::ExecuteCommand& command);
     bool sendRawPacket(const softbus::core::RawPacket& packet);
+    bool sendAck(const softbus::core::SoftbusAck& ack);
+    bool sendHeartbeat(const std::string& deviceId);
     void poll();
 
     bool isConnected() const { return connected_; }

src/softbus_transport/include/softbus_transport/TcpUplinkServer.h

@@ -14,6 +14,7 @@
 namespace softbus::transport {
 
 using RawPacketHandler = std::function<void(const softbus::core::RawPacket&)>;
+using WireMessageHandler = std::function<void(const WireMessage&)>;
 using CommandSender = std::function<bool(const softbus::api::ExecuteCommand&)>;
 
 class TcpUplinkServer {
@@ -22,6 +23,7 @@ public:
     ~TcpUplinkServer();
 
     void setRawPacketHandler(RawPacketHandler handler);
+    void setWireMessageHandler(WireMessageHandler handler);
     void start();
     void stop();
     bool sendCommand(const softbus::api::ExecuteCommand& command);
@@ -34,6 +36,7 @@ private:
     asio::io_context& io_;
     asio::ip::tcp::acceptor acceptor_;
     RawPacketHandler rawHandler_;
+    WireMessageHandler wireHandler_;
     std::shared_ptr<asio::ip::tcp::socket> clientSocket_;
     std::string readBuffer_;
     std::mutex sendMutex_;

src/softbus_transport/include/softbus_transport/TransportBusBridge.h

@@ -0,0 +1,21 @@
+#pragma once
+
+#include <softbus_bus/Bus.h>
+#include <softbus_transport/TcpUplinkServer.h>
+
+namespace softbus::transport {
+
+class TransportBusBridge {
+public:
+    void attach(softbus::bus::Bus* bus, TcpUplinkServer* server);
+
+private:
+    void onBusMessage(const softbus::bus::BusMessage& message);
+    void onWireMessage(const WireMessage& message);
+
+    softbus::bus::Bus* bus_{nullptr};
+    TcpUplinkServer* server_{nullptr};
+    softbus::bus::Bus::SubscriptionId cmdSub_{0};
+};
+
+} // namespace softbus::transport

src/softbus_transport/include/softbus_transport/UplinkWire.h

@@ -1,6 +1,7 @@
 #pragma once
 
 #include <softbus_core/models/RawPacket.h>
+#include <softbus_core/models/SoftbusTypes.h>
 
 #include <softbus_api/ExecuteCommand.h>
 
@@ -14,6 +15,8 @@ namespace softbus::transport {
 enum class WireMessageType {
     RawPacket,
     ExecuteCommand,
+    Ack,
+    Heartbeat,
     Unknown
 };
 
@@ -21,10 +24,14 @@ struct WireMessage {
     WireMessageType type{WireMessageType::Unknown};
     softbus::core::RawPacket rawPacket;
     softbus::api::ExecuteCommand command;
+    softbus::core::SoftbusAck ack;
+    std::string heartbeatDeviceId;
 };
 
 std::string encodeWireMessage(const softbus::core::RawPacket& packet);
 std::string encodeWireMessage(const softbus::api::ExecuteCommand& command);
+std::string encodeWireMessage(const softbus::core::SoftbusAck& ack);
+std::string encodeWireMessage(const std::string& deviceId);
 std::optional<WireMessage> decodeWireMessage(const std::string& line);
 
 std::string base64Encode(const uint8_t* data, std::size_t size);

src/softbus_transport/src/TcpUplinkClient.cpp

@@ -62,6 +62,42 @@ void TcpUplinkClient::disconnect()
         asio::make_work_guard(io_));
 }
 
+bool TcpUplinkClient::sendCommand(const softbus::api::ExecuteCommand& command)
+{
+    if (!socket_ || !socket_->is_open()) {
+        return false;
+    }
+    const auto line = encodeWireMessage(command);
+    std::lock_guard<std::mutex> lock(sendMutex_);
+    std::error_code ec;
+    asio::write(*socket_, asio::buffer(line), ec);
+    return !ec;
+}
+
+bool TcpUplinkClient::sendAck(const softbus::core::SoftbusAck& ack)
+{
+    if (!socket_ || !socket_->is_open()) {
+        return false;
+    }
+    const auto line = encodeWireMessage(ack);
+    std::lock_guard<std::mutex> lock(sendMutex_);
+    std::error_code ec;
+    asio::write(*socket_, asio::buffer(line), ec);
+    return !ec;
+}
+
+bool TcpUplinkClient::sendHeartbeat(const std::string& deviceId)
+{
+    if (!socket_ || !socket_->is_open()) {
+        return false;
+    }
+    const auto line = encodeWireMessage(deviceId);
+    std::lock_guard<std::mutex> lock(sendMutex_);
+    std::error_code ec;
+    asio::write(*socket_, asio::buffer(line), ec);
+    return !ec;
+}
+
 bool TcpUplinkClient::sendRawPacket(const softbus::core::RawPacket& packet)
 {
     if (!socket_ || !socket_->is_open()) {

src/softbus_transport/src/TcpUplinkServer.cpp

@@ -22,6 +22,11 @@ void TcpUplinkServer::setRawPacketHandler(RawPacketHandler handler)
     rawHandler_ = std::move(handler);
 }
 
+void TcpUplinkServer::setWireMessageHandler(WireMessageHandler handler)
+{
+    wireHandler_ = std::move(handler);
+}
+
 void TcpUplinkServer::start()
 {
     running_ = true;
@@ -107,6 +112,9 @@ void TcpUplinkServer::handleLine(const std::string& line)
     if (!msg) {
         return;
     }
+    if (wireHandler_) {
+        wireHandler_(*msg);
+    }
     if (msg->type == WireMessageType::RawPacket && rawHandler_) {
         rawHandler_(msg->rawPacket);
     }

src/softbus_transport/src/TransportBusBridge.cpp

@@ -0,0 +1,70 @@
+#include <softbus_transport/TransportBusBridge.h>
+
+#include <softbus_core/logging/Logger.h>
+
+namespace softbus::transport {
+
+void TransportBusBridge::attach(softbus::bus::Bus* bus, TcpUplinkServer* server)
+{
+    bus_ = bus;
+    server_ = server;
+    if (!bus_ || !server_) {
+        return;
+    }
+
+    server_->setWireMessageHandler([this](const WireMessage& msg) {
+        onWireMessage(msg);
+    });
+
+    cmdSub_ = bus_->subscribe("control/cmd/#", [this](const softbus::bus::BusMessage& msg) {
+        onBusMessage(msg);
+    });
+    bus_->subscribe("control/cmd/injected", [this](const softbus::bus::BusMessage& msg) {
+        onBusMessage(msg);
+    });
+}
+
+void TransportBusBridge::onBusMessage(const softbus::bus::BusMessage& message)
+{
+    if (!server_ || message.type != softbus::bus::BusPayloadType::ExecuteCommand) {
+        return;
+    }
+    if (!server_->sendCommand(message.command)) {
+        SB_LOG_WARN("TransportBusBridge", "failed to send command to edge topic=", message.topic);
+    }
+}
+
+void TransportBusBridge::onWireMessage(const WireMessage& message)
+{
+    if (!bus_) {
+        return;
+    }
+
+    softbus::bus::BusMessage busMsg;
+    switch (message.type) {
+    case WireMessageType::RawPacket:
+        busMsg.type = softbus::bus::BusPayloadType::RawPacket;
+        busMsg.rawPacket = message.rawPacket;
+        bus_->publish("edge/raw/" + message.rawPacket.sourceId, busMsg);
+        break;
+    case WireMessageType::Ack:
+        busMsg.type = softbus::bus::BusPayloadType::Ack;
+        busMsg.ack = message.ack;
+        bus_->publish("control/ack/" + message.ack.deviceId + "/" + message.ack.pointId, busMsg);
+        break;
+    case WireMessageType::Heartbeat:
+        busMsg.type = softbus::bus::BusPayloadType::Heartbeat;
+        busMsg.deviceId = message.heartbeatDeviceId;
+        bus_->publish("system/heartbeat/" + message.heartbeatDeviceId, busMsg);
+        break;
+    case WireMessageType::ExecuteCommand:
+        busMsg.type = softbus::bus::BusPayloadType::ExecuteCommand;
+        busMsg.command = message.command;
+        bus_->publish("control/cmd/injected", busMsg);
+        break;
+    default:
+        break;
+    }
+}
+
+} // namespace softbus::transport

src/softbus_transport/src/UplinkWire.cpp

@@ -1,5 +1,6 @@
 #include <softbus_transport/UplinkWire.h>
 
+#include <chrono>
 #include <nlohmann/json.hpp>
 
 namespace softbus::transport {
@@ -78,6 +79,23 @@ std::string encodeWireMessage(const softbus::api::ExecuteCommand& command)
     return command.toJson().dump() + "\n";
 }
 
+std::string encodeWireMessage(const softbus::core::SoftbusAck& ack)
+{
+    return ack.toJson().dump() + "\n";
+}
+
+std::string encodeWireMessage(const std::string& deviceId)
+{
+    nlohmann::json j{
+        {"type", "heartbeat"},
+        {"deviceId", deviceId},
+        {"timestampNs", static_cast<std::uint64_t>(
+            std::chrono::duration_cast<std::chrono::nanoseconds>(
+                std::chrono::system_clock::now().time_since_epoch()).count())}
+    };
+    return j.dump() + "\n";
+}
+
 std::optional<WireMessage> decodeWireMessage(const std::string& line)
 {
     if (line.empty()) {
@@ -104,6 +122,16 @@ std::optional<WireMessage> decodeWireMessage(const std::string& line)
         msg.command = softbus::api::ExecuteCommand::fromJson(j);
         return msg;
     }
+    if (type == "ack") {
+        msg.type = WireMessageType::Ack;
+        msg.ack = softbus::core::SoftbusAck::fromJson(j);
+        return msg;
+    }
+    if (type == "heartbeat") {
+        msg.type = WireMessageType::Heartbeat;
+        msg.heartbeatDeviceId = j.value("deviceId", "");
+        return msg;
+    }
     return std::nullopt;
 }
 

tests/integration/valve_control_loop.sh

@@ -0,0 +1,26 @@
+#!/usr/bin/env bash
+set -euo pipefail
+ROOT="$(cd "$(dirname "$0")/../.." && pwd)"
+cd "$ROOT"
+
+fuser -k 9000/tcp 4840/tcp 2>/dev/null || true
+sleep 1
+
+./build/bin/softbus_daemon \
+  --config config/daemon_profile.json \
+  --point-table config/point_table.json \
+  --self-test > /tmp/valve_loop_daemon.log 2>&1 &
+DAEMON_PID=$!
+sleep 1
+./build/bin/softbus_edge --config config/edge.json > /tmp/valve_loop_edge.log 2>&1 &
+EDGE_PID=$!
+sleep 8
+
+kill "$DAEMON_PID" "$EDGE_PID" 2>/dev/null || true
+fuser -k 9000/tcp 4840/tcp 2>/dev/null || true
+
+grep -q "state=Succeeded" /tmp/valve_loop_daemon.log
+grep -q "executedValue=60" /tmp/valve_loop_daemon.log
+grep -q "ValveOpening value=60" /tmp/valve_loop_edge.log
+
+echo "valve_control_loop: PASS"

tools/CMakeLists.txt

@@ -0,0 +1,11 @@
+add_executable(cmd_sender cmd_sender.cpp)
+target_link_libraries(cmd_sender PRIVATE
+    softbus_core softbus_api softbus_transport softbus_registry asio
+)
+softbus_apply_platform_libs(cmd_sender)
+
+add_executable(topic_watcher topic_watcher.cpp)
+target_link_libraries(topic_watcher PRIVATE
+    softbus_bus softbus_core
+)
+softbus_apply_platform_libs(topic_watcher)

tools/cmd_sender.cpp

@@ -0,0 +1,45 @@
+#include <softbus_api/ExecuteCommand.h>
+#include <softbus_core/identity/DeviceIdentity.h>
+#include <softbus_transport/TcpUplinkClient.h>
+
+#include <iostream>
+#include <string>
+
+int main(int argc, char* argv[])
+{
+    std::string host = "127.0.0.1";
+    uint16_t port = 9000;
+    std::string objectRef = "DemoSite/DemoSystem/DemoAsset/EdgeDevice_01/ValveOpening";
+    double value = 60.0;
+
+    for (int i = 1; i < argc; ++i) {
+        const std::string arg = argv[i];
+        if (arg == "--daemon" && i + 1 < argc) {
+            const std::string endpoint = argv[++i];
+            const auto pos = endpoint.find(':');
+            if (pos != std::string::npos) {
+                host = endpoint.substr(0, pos);
+                port = static_cast<uint16_t>(std::stoi(endpoint.substr(pos + 1)));
+            }
+        } else if (arg == "--object-ref" && i + 1 < argc) {
+            objectRef = argv[++i];
+        } else if (arg == "--value" && i + 1 < argc) {
+            value = std::stod(argv[++i]);
+        }
+    }
+
+    softbus::api::ExecuteCommand cmd;
+    cmd.requestId = "req-cli-" + softbus::core::DeviceIdentity::generateTraceId();
+    cmd.traceId = softbus::core::DeviceIdentity::generateTraceId();
+    cmd.objectRef = objectRef;
+    cmd.command = "write";
+    cmd.params = {{"value", value}};
+
+    std::cout << cmd.toJson().dump() << std::endl;
+    std::cout << "Use OPC UA ExecuteCommand or: softbus_daemon --self-test" << std::endl;
+    std::cout << "Note: uplink port " << host << ":" << port
+              << " is reserved for edge agent connection." << std::endl;
+    (void)host;
+    (void)port;
+    return 0;
+}

tools/topic_watcher.cpp

@@ -0,0 +1,15 @@
+#include <softbus_bus/Bus.h>
+
+#include <iostream>
+#include <string>
+
+int main()
+{
+    auto& bus = softbus::bus::Bus::instance();
+    bus.subscribe("#", [](const softbus::bus::BusMessage& msg) {
+        std::cout << "[topic] " << msg.topic << std::endl;
+    });
+    std::cout << "topic_watcher: attach to daemon process is standalone demo only." << std::endl;
+    std::cout << "Use daemon logs or integrate watcher via shared bus in-process." << std::endl;
+    return 0;
+}