diff --git a/CMakeLists.txt b/CMakeLists.txt
index 33c1407..05b3ed1 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -119,14 +119,26 @@ qt_add_executable(softbus_daemon
     src/core/plugin_system/IProtocolPlugin.h
     src/core/plugin_system/PluginManager.h
     src/core/plugin_system/PluginManager.cpp
-    src/core/plugin_system/builtins/ModbusRtuProtocolPlugin.h
-    src/core/plugin_system/builtins/ModbusRtuProtocolPlugin.cpp
+    src/message_bus/pipeline/protocol/modbusrtu/ModbusRtuProtocolPlugin.h
+    src/message_bus/pipeline/protocol/modbusrtu/ModbusRtuProtocolPlugin.cpp
+    src/core/models/Types.h
+    src/core/models/MetadataDef.h
+    src/core/models/DataMapping.h
+    src/core/models/DOMMessage.h
+    src/core/metadata/MetadataRegistry.h
+    src/core/metadata/MetadataRegistry.cpp
+    src/core/metadata/ProfileRegistry.h
+    src/core/metadata/ProfileRegistry.cpp
 
     # message bus pipeline
     src/message_bus/pipeline/PayloadRef.h
     src/message_bus/pipeline/PipelineContext.h
     src/message_bus/pipeline/PipelineEngine.h
     src/message_bus/pipeline/PipelineEngine.cpp
+    src/message_bus/pipeline/protocol/ProtocolEnvelope.h
+    src/message_bus/pipeline/protocol/modbusrtu/ModbusRtuPdu.h
+    src/message_bus/pipeline/protocol/modbusrtu/ModbusRtuCodec.h
+    src/message_bus/pipeline/protocol/modbusrtu/ModbusRtuCodec.cpp
     src/message_bus/pipeline/stages/1_framers/IFramer.h
     src/message_bus/pipeline/stages/1_framers/ModbusRtuFraming.h
     src/message_bus/pipeline/stages/1_framers/ModbusRtuFraming.cpp
@@ -136,6 +148,8 @@ qt_add_executable(softbus_daemon
     src/message_bus/pipeline/stages/1_framers/PassthroughFramer.cpp
     src/message_bus/pipeline/stages/2_parsers/IParserStage.h
     src/message_bus/pipeline/stages/3_filters/IFilterStage.h
+    src/message_bus/pipeline/stages/3_filters/ProtocolParseFilter.h
+    src/message_bus/pipeline/stages/3_filters/ProtocolParseFilter.cpp
     src/message_bus/pipeline/stages/4_validators/IValidator.h
 
     # message bus ingress/egress
@@ -202,6 +216,35 @@ if(SOFTBUS_BUILD_TESTS)
         Qt6::Core
         Qt6::Test
     )
+
+    qt_add_executable(softbus_protocol_parse_filter_test
+        tests/protocol_parse_filter_test.cpp
+        src/core/metadata/MetadataRegistry.cpp
+        src/core/metadata/ProfileRegistry.cpp
+        src/message_bus/pipeline/stages/3_filters/ProtocolParseFilter.cpp
+    )
+
+    qt_add_executable(softbus_modbus_rtu_pdu_codec_test
+        tests/modbus_rtu_pdu_codec_test.cpp
+        src/message_bus/pipeline/protocol/modbusrtu/ModbusRtuCodec.cpp
+        src/message_bus/pipeline/stages/1_framers/ModbusRtuFraming.cpp
+    )
+    target_include_directories(softbus_modbus_rtu_pdu_codec_test PRIVATE
+        ${CMAKE_SOURCE_DIR}
+        ${CMAKE_SOURCE_DIR}/src
+    )
+    target_link_libraries(softbus_modbus_rtu_pdu_codec_test PRIVATE
+        Qt6::Core
+        Qt6::Test
+    )
+    target_include_directories(softbus_protocol_parse_filter_test PRIVATE
+        ${CMAKE_SOURCE_DIR}
+        ${CMAKE_SOURCE_DIR}/src
+    )
+    target_link_libraries(softbus_protocol_parse_filter_test PRIVATE
+        Qt6::Core
+        Qt6::Test
+    )
 endif()
 
 install(TARGETS softbus_daemon
diff --git a/bin/softbus_daemon b/bin/softbus_daemon
index a25886d..1f53231 100755
Binary files a/bin/softbus_daemon and b/bin/softbus_daemon differ
diff --git a/src/device_bus/manager/DeviceBusManager.cpp b/src/device_bus/manager/DeviceBusManager.cpp
index e0a6bc9..8cfc0a9 100644
--- a/src/device_bus/manager/DeviceBusManager.cpp
+++ b/src/device_bus/manager/DeviceBusManager.cpp
@@ -10,7 +10,7 @@
 #include "message_bus/ingress/DriverIngressAdapter.h"
 #include "message_bus/pipeline/PipelineEngine.h"
 #include "core/plugin_system/PluginManager.h"
-#include "core/plugin_system/builtins/ModbusRtuProtocolPlugin.h"
+#include "message_bus/pipeline/protocol/modbusrtu/ModbusRtuProtocolPlugin.h"
 #include "device_bus/manager/SerialDeviceManager.h"
 #include "devices/DeviceTypes.h"
 #include "utils/logs/logging.h"
diff --git a/src/devices/physical/SerialDevice.cpp b/src/devices/physical/SerialDevice.cpp
index 4833e91..7993672 100644
--- a/src/devices/physical/SerialDevice.cpp
+++ b/src/devices/physical/SerialDevice.cpp
@@ -114,6 +114,7 @@ void SerialDevice::onRead(const std::vector<std::uint8_t>& data)
     LOG_INFO() << "SerialDevice: onRead data size" << data.size() << "endpoint" << m_config.endpoint;
     // 将数据推送到 ingress 队列 包含端口名、协议、数据、时间戳、traceId
     softbus::message_bus::pipeline::PipelineContext ctx;
+    ctx.direction = softbus::message_bus::pipeline::BusDirection::Upstream;
     ctx.endpoint = m_config.endpoint;
     ctx.protocolHint = m_info.protocol;
     ctx.frameKind    = softbus::message_bus::pipeline::FrameKind::StreamChunk;
diff --git a/src/message_bus/pipeline/PipelineContext.h b/src/message_bus/pipeline/PipelineContext.h
index f882815..214baa6 100644
--- a/src/message_bus/pipeline/PipelineContext.h
+++ b/src/message_bus/pipeline/PipelineContext.h
@@ -8,8 +8,11 @@
 #include <QJsonObject>
 #include <QMetaType>
 #include <QString>
+#include <QVector>
 
+#include "core/models/DOMMessage.h"
 #include "message_bus/pipeline/PayloadRef.h"
+#include "message_bus/pipeline/protocol/ProtocolEnvelope.h"
 
 namespace softbus::message_bus::pipeline
 {
@@ -43,7 +46,12 @@ struct PipelineContext
     BusDirection direction{BusDirection::Unknown};
     std::uint64_t frameSeq{0};
 
+    /// 强类型协议解析结果（推荐下游使用）；多协议扩展时只增不改 variant。
+    softbus::message_bus::pipeline::protocol::ProtocolEnvelope protocol;
+
+    /// 兼容旧逻辑与日志；新代码请优先使用 `protocol`，勿依赖此处 JSON 字段名。
     QJsonObject parsed;
+    QVector<softbus::core::models::DOMMessage> domMessages;
     QString framerError; // 成帧/Pipeline 旁路诊断（可选）
 };
 
diff --git a/src/message_bus/pipeline/PipelineEngine.cpp b/src/message_bus/pipeline/PipelineEngine.cpp
index 86a24f6..61904e0 100644
--- a/src/message_bus/pipeline/PipelineEngine.cpp
+++ b/src/message_bus/pipeline/PipelineEngine.cpp
@@ -7,6 +7,7 @@
 #include <functional>
 
 #include "message_bus/pipeline/stages/1_framers/ModbusRtuFraming.h"
+#include "message_bus/pipeline/stages/3_filters/ProtocolParseFilter.h"
 #include "utils/logs/logging.h"
 
 namespace softbus::message_bus::pipeline
@@ -360,16 +361,7 @@ void PipelineEngine::parseWorkerLoop()
             }
             continue;
         }
-        std::ostringstream oss;
-        oss << std::hex << std::setfill('0');
-        for (std::size_t i = 0; i < ctx.payload.block->size; ++i) {
-            oss << std::setw(2) << static_cast<int>(ctx.payload.block->data[i]);
-            if (i + 1 < ctx.payload.block->size) {
-                oss << ' ';
-            }
-        }
-        const QString hexPreview = QString::fromStdString(oss.str());
-        LOG_INFO() << "the data is"<< hexPreview;
+
 
         const bool ok = runStages234(ctx);
         if (m_rtConfig.orderedEgress && m_mergeQ) {
@@ -449,9 +441,39 @@ bool PipelineEngine::runStages234(PipelineContext& ctx)
     if (ctx.frameKind != FrameKind::CompleteFrame) {
         return false;
     }
+    std::ostringstream oss;
+    oss << std::hex << std::setfill('0');
+    for (std::size_t i = 0; i < ctx.payload.block->size; ++i) {
+        oss << std::setw(2) << static_cast<int>(ctx.payload.block->data[i]);
+        if (i + 1 < ctx.payload.block->size) {
+            oss << ' ';
+        }
+    }
+    const QString hexPreview = QString::fromStdString(oss.str());
+    LOG_INFO() << "the data is"<< hexPreview;
+    LOG_INFO() << "the protocol hint is"<< ctx.protocolHint;
+    const QString directionText = (ctx.direction == BusDirection::Upstream)
+                                      ? QStringLiteral("Upstream")
+                                      : (ctx.direction == BusDirection::Downstream)
+                                            ? QStringLiteral("Downstream")
+                                            : QStringLiteral("Unknown");
+    LOG_INFO() << "the direction is" << directionText;
     if (!stage2Parser(ctx)) {
         return false;
     }
+    QJsonObject parsed = ctx.parsed;
+    LOG_INFO() << "the parsed is"<< parsed;
+    
+    softbus::message_bus::pipeline::protocol::ProtocolEnvelope protocol = ctx.protocol;
+    if (const auto* modbusPdu = std::get_if<softbus::message_bus::pipeline::protocol::ModbusRtuPdu>(&protocol.pdu)) {
+        LOG_INFO() << "the protocol text is" << softbus::message_bus::pipeline::protocol::toText(*modbusPdu);
+    } else {
+        LOG_INFO() << "the protocol text is" << QStringLiteral("ProtocolEnvelopePdu::monostate");
+    }
+
+    m_count++;
+    LOG_INFO() << "the count of command is" << m_count;
+
     if (!stage3Filter(ctx)) {
         return false;
     }
@@ -504,8 +526,8 @@ bool PipelineEngine::stage2Parser(PipelineContext& ctx)
 
 bool PipelineEngine::stage3Filter(PipelineContext& ctx)
 {
-    (void)ctx;
-    return true;
+    static softbus::message_bus::pipeline::stages::filters::ProtocolParseFilter s_filter;
+    return s_filter.apply(ctx);
 }
 
 } // namespace softbus::message_bus::pipeline
diff --git a/src/message_bus/pipeline/PipelineEngine.h b/src/message_bus/pipeline/PipelineEngine.h
index 15b168b..b29fa59 100644
--- a/src/message_bus/pipeline/PipelineEngine.h
+++ b/src/message_bus/pipeline/PipelineEngine.h
@@ -122,6 +122,7 @@ private:
 
     std::mutex                 m_mergeStateMutex;
     QHash<QString, MergeState> m_mergeByEndpoint;
+    int m_count{0}; // 计数器
 };
 
 } // namespace softbus::message_bus::pipeline