refactor: reorganize tests/tools, share python protocol helpers

.github/workflows/ci.yml

@@ -0,0 +1,41 @@
+name: CI
+
+on:
+  push:
+  pull_request:
+
+env:
+  CARGO_TERM_COLOR: always
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Install Linux system dependencies
+        run: |
+          sudo apt-get update
+          sudo apt-get install -y libudev-dev libxkbcommon-dev libxcb1-dev libxcb-icccm4-dev libxcb-shape0-dev libxcb-xfixes0-dev libgl1-mesa-dev libgtk-3-dev
+
+      - uses: dtolnay/rust-toolchain@stable
+        with:
+          components: rustfmt, clippy
+
+      - uses: Swatinem/rust-cache@v2
+
+      - name: Check formatting
+        run: cargo fmt --check
+
+      - name: Clippy
+        run: cargo clippy --workspace --all-targets -- -D warnings
+
+      - name: Smoke-check Python tools
+        run: |
+          python3 -m py_compile tools/*.py
+          for f in tools/test_plot.py tools/test_drone.py tools/test_ansi.py tools/stress_test.py; do
+            python3 "$f" --help >/dev/null
+          done
+
+      - name: Run tests
+        run: cargo test --workspace

Cargo.lock

@@ -432,15 +432,30 @@ version = "1.5.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f2032f911046de80f0a198e0901378627c33f59ea0ac00e363d481118bd70a53"
 
+[[package]]
+name = "bit-set"
+version = "0.8.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "08807e080ed7f9d5433fa9b275196cfc35414f66a0c79d864dc51a0d825231a3"
+dependencies = [
+ "bit-vec 0.8.0",
+]
+
 [[package]]
 name = "bit-set"
 version = "0.9.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "34ddef2995421ab6a5c779542c81ee77c115206f4ad9d5a8e05f4ff49716a3dd"
 dependencies = [
- "bit-vec",
+ "bit-vec 0.9.1",
 ]
 
+[[package]]
+name = "bit-vec"
+version = "0.8.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5e764a1d40d510daf35e07be9eb06e75770908c27d411ee6c92109c9840eaaf7"
+
 [[package]]
 name = "bit-vec"
 version = "0.9.1"
@@ -1117,6 +1132,12 @@ dependencies = [
  "miniz_oxide",
 ]
 
+[[package]]
+name = "fnv"
+version = "1.0.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3f9eec918d3f24069decb9af1554cad7c880e2da24a9afd88aca000531ab82c1"
+
 [[package]]
 name = "foldhash"
 version = "0.1.5"
@@ -2044,7 +2065,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "0dd91265cc2454558f659b3b4b9640f0ddb8cc6521277f166b8a8c181c898079"
 dependencies = [
  "arrayvec",
- "bit-set",
+ "bit-set 0.9.1",
  "bitflags 2.12.1",
  "cfg-if",
  "cfg_aliases",
@@ -2680,6 +2701,7 @@ dependencies = [
  "pipeview-core",
  "serde",
  "serde_json",
+ "tempfile",
  "thiserror 2.0.18",
  "tokio",
  "tracing",
@@ -2694,6 +2716,7 @@ dependencies = [
  "bytes",
  "hex",
  "nom 8.0.0",
+ "proptest",
  "serde",
  "serde_json",
  "serialport",
@@ -2799,6 +2822,15 @@ dependencies = [
  "zerovec",
 ]
 
+[[package]]
+name = "ppv-lite86"
+version = "0.2.21"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "85eae3c4ed2f50dcfe72643da4befc30deadb458a9b590d720cde2f2b1e97da9"
+dependencies = [
+ "zerocopy",
+]
+
 [[package]]
 name = "presser"
 version = "0.3.1"
@@ -2861,12 +2893,37 @@ version = "1.0.18"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "3d595e54a326bc53c1c197b32d295e14b169e3cfeaa8dc82b529f947fba6bcf5"
 
+[[package]]
+name = "proptest"
+version = "1.11.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4b45fcc2344c680f5025fe57779faef368840d0bd1f42f216291f0dc4ace4744"
+dependencies = [
+ "bit-set 0.8.0",
+ "bit-vec 0.8.0",
+ "bitflags 2.12.1",
+ "num-traits",
+ "rand",
+ "rand_chacha",
+ "rand_xorshift",
+ "regex-syntax",
+ "rusty-fork",
+ "tempfile",
+ "unarray",
+]
+
 [[package]]
 name = "pxfm"
 version = "0.1.29"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "e0c5ccf5294c6ccd63a74f1565028353830a9c2f5eb0c682c355c471726a6e3f"
 
+[[package]]
+name = "quick-error"
+version = "1.2.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a1d01941d82fa2ab50be1e79e6714289dd7cde78eba4c074bc5a4374f650dfe0"
+
 [[package]]
 name = "quick-error"
 version = "2.0.1"
@@ -2904,6 +2961,44 @@ version = "6.0.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f8dcc9c7d52a811697d2151c701e0d08956f92b0e24136cf4cf27b57a6a0d9bf"
 
+[[package]]
+name = "rand"
+version = "0.9.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b9ef1d0d795eb7d84685bca4f72f3649f064e6641543d3a8c415898726a57b41"
+dependencies = [
+ "rand_chacha",
+ "rand_core",
+]
+
+[[package]]
+name = "rand_chacha"
+version = "0.9.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d3022b5f1df60f26e1ffddd6c66e8aa15de382ae63b3a0c1bfc0e4d3e3f325cb"
+dependencies = [
+ "ppv-lite86",
+ "rand_core",
+]
+
+[[package]]
+name = "rand_core"
+version = "0.9.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "76afc826de14238e6e8c374ddcc1fa19e374fd8dd986b0d2af0d02377261d83c"
+dependencies = [
+ "getrandom 0.3.4",
+]
+
+[[package]]
+name = "rand_xorshift"
+version = "0.4.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "513962919efc330f829edb2535844d1b912b0fbe2ca165d613e4e8788bb05a5a"
+dependencies = [
+ "rand_core",
+]
+
 [[package]]
 name = "range-alloc"
 version = "0.1.5"
@@ -3053,6 +3148,18 @@ version = "1.0.22"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b39cdef0fa800fc44525c84ccb54a029961a8215f9619753635a9c0d2538d46d"
 
+[[package]]
+name = "rusty-fork"
+version = "0.3.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cc6bf79ff24e648f6da1f8d1f011e9cac26491b619e6b9280f2b47f1774e6ee2"
+dependencies = [
+ "fnv",
+ "quick-error 1.2.3",
+ "tempfile",
+ "wait-timeout",
+]
+
 [[package]]
 name = "same-file"
 version = "1.0.6"
@@ -3477,7 +3584,7 @@ dependencies = [
  "fax",
  "flate2",
  "half",
- "quick-error",
+ "quick-error 2.0.1",
  "weezl",
  "zune-jpeg",
 ]
@@ -3683,6 +3790,12 @@ dependencies = [
  "windows-sys 0.61.2",
 ]
 
+[[package]]
+name = "unarray"
+version = "0.1.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "eaea85b334db583fe3274d12b4cd1880032beab409c0d774be044d4480ab9a94"
+
 [[package]]
 name = "unescaper"
 version = "0.1.8"
@@ -3793,6 +3906,15 @@ dependencies = [
  "memchr",
 ]
 
+[[package]]
+name = "wait-timeout"
+version = "0.2.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "09ac3b126d3914f9849036f826e054cbabdc8519970b8998ddaf3b5bd3c65f11"
+dependencies = [
+ "libc",
+]
+
 [[package]]
 name = "walkdir"
 version = "2.5.0"
@@ -4130,8 +4252,8 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "02da3ad1b568337f25513b317870960ef87073ea0945502e44b864b67a8c77b7"
 dependencies = [
  "arrayvec",
- "bit-set",
+ "bit-set 0.9.1",
- "bit-vec",
+ "bit-vec 0.9.1",
  "bitflags 2.12.1",
  "bytemuck",
  "cfg_aliases",
@@ -4202,7 +4324,7 @@ dependencies = [
  "android_system_properties",
  "arrayvec",
  "ash",
- "bit-set",
+ "bit-set 0.9.1",
  "bitflags 2.12.1",
  "block2 0.6.2",
  "bytemuck",

README.md

@@ -229,7 +229,7 @@ return {
 }
 ```
 
-参考实现：`tests/lua_line_framer.lua`（按 `\n` 分割的行分帧器）
+参考实现：`crates/pipeview-client/tests/fixtures/lua_line_framer.lua`（按 `\n` 分割的行分帧器）
 
 ### 解码器 API
 
@@ -363,11 +363,11 @@ RUST_LOG=pipeview_gui=debug cargo run -p pipeview-gui  # 详细日志
 crates/
   pipeview-core/        # 传输、分帧、协议
   pipeview-client/      # 会话管理、Lua 运行时
+    tests/fixtures/     # Lua 测试夹具
   pipeview-gui/         # egui 桌面应用
 examples/              # Lua 脚本示例
   drone_plot.lua       # 飞控遥测波形解码器
   drone_text.lua       # 飞控遥测文本解码器
-tests/                 # Lua 测试 fixture
 tools/                 # 开发辅助工具
   test_plot.py         # 波形测试数据生成器
   test_drone.py        # 飞控测试数据生成器

README_en.md

@@ -217,7 +217,7 @@ return {
 }
 ```
 
-Reference: `tests/lua_line_framer.lua` (line-based framer splitting on `\n`).
+Reference: `crates/pipeview-client/tests/fixtures/lua_line_framer.lua` (line-based framer splitting on `\n`).
 
 ### Decoder API
 
@@ -347,11 +347,11 @@ RUST_LOG=pipeview_gui=debug cargo run -p pipeview-gui
 crates/
   pipeview-core/        # Transport, framing, protocol
   pipeview-client/      # Session management, Lua runtime
+    tests/fixtures/     # Lua test fixtures
   pipeview-gui/         # egui desktop application
 examples/              # Lua script examples
   drone_plot.lua       # Drone telemetry plot decoder
   drone_text.lua       # Drone telemetry text decoder
-tests/                 # Lua test fixtures
 tools/                 # Development utilities
   test_plot.py         # Waveform test data generator
   test_drone.py        # Drone test data generator

crates/pipeview-client/Cargo.toml

@@ -15,3 +15,4 @@ thiserror = { workspace = true }
 [dev-dependencies]
 tokio = { workspace = true, features = ["full"] }
 tracing-subscriber = { workspace = true }
+tempfile = "3"

crates/pipeview-client/src/history.rs

@@ -57,3 +57,116 @@ impl RingBuffer<DecodedEntry> {
             .collect()
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::event::DecodedEntry;
+    use pipeview_core::protocol::DecodedData;
+
+    #[test]
+    fn push_evicts_oldest_when_limit_reached() {
+        let mut rb = RingBuffer::new(3);
+        rb.push(1);
+        rb.push(2);
+        rb.push(3);
+        rb.push(4);
+        rb.push(5);
+
+        assert_eq!(rb.len(), 3);
+        assert_eq!(rb.get(0), Some(&3));
+        assert_eq!(rb.get(1), Some(&4));
+        assert_eq!(rb.get(2), Some(&5));
+        assert_eq!(rb.get(3), None);
+    }
+
+    #[test]
+    fn iter_yields_oldest_to_newest() {
+        let mut rb = RingBuffer::new(3);
+        rb.push("a");
+        rb.push("b");
+        rb.push("c");
+
+        let items: Vec<_> = rb.iter().collect();
+        assert_eq!(items, vec![&"a", &"b", &"c"]);
+    }
+
+    #[test]
+    fn clear_empties_buffer() {
+        let mut rb = RingBuffer::new(3);
+        rb.push(1);
+        rb.push(2);
+        rb.clear();
+
+        assert!(rb.is_empty());
+        assert_eq!(rb.len(), 0);
+        assert_eq!(rb.get(0), None);
+    }
+
+    #[test]
+    fn set_limit_trims_excess() {
+        let mut rb = RingBuffer::new(5);
+        for i in 0..5 {
+            rb.push(i);
+        }
+        rb.set_limit(2);
+
+        assert_eq!(rb.len(), 2);
+        assert_eq!(rb.get(0), Some(&3));
+        assert_eq!(rb.get(1), Some(&4));
+    }
+
+    #[test]
+    fn set_limit_can_grow_without_data_loss() {
+        let mut rb = RingBuffer::new(2);
+        rb.push(1);
+        rb.push(2);
+        rb.set_limit(5);
+
+        assert_eq!(rb.len(), 2);
+        assert_eq!(rb.get(0), Some(&1));
+        assert_eq!(rb.get(1), Some(&2));
+    }
+
+    #[test]
+    fn drain_recent_respects_count_and_order() {
+        let mut rb = RingBuffer::new(5);
+        for i in 0..5 {
+            rb.push(i);
+        }
+
+        let empty: Vec<i32> = rb.drain_recent(0);
+        assert!(empty.is_empty());
+
+        let last_two = rb.drain_recent(2);
+        assert_eq!(last_two, vec![3, 4]);
+
+        let all = rb.drain_recent(10);
+        assert_eq!(all, vec![0, 1, 2, 3, 4]);
+    }
+
+    #[test]
+    fn entries_for_pipeline_filters_by_name() {
+        let mut rb = RingBuffer::new(10);
+        rb.push(DecodedEntry {
+            pipeline_name: "a".into(),
+            data: DecodedData::Text("one".into()),
+        });
+        rb.push(DecodedEntry {
+            pipeline_name: "b".into(),
+            data: DecodedData::Text("two".into()),
+        });
+        rb.push(DecodedEntry {
+            pipeline_name: "a".into(),
+            data: DecodedData::Text("three".into()),
+        });
+
+        let a = rb.entries_for_pipeline("a");
+        assert_eq!(a.len(), 2);
+        assert_eq!(a[0].pipeline_name, "a");
+        assert_eq!(a[1].pipeline_name, "a");
+
+        let none = rb.entries_for_pipeline("missing");
+        assert!(none.is_empty());
+    }
+}

crates/pipeview-client/tests/lua_fixtures.rs

@@ -0,0 +1,66 @@
+use std::fs;
+use std::path::PathBuf;
+
+use mlua::{Function, Lua, Table};
+
+fn fixture_path(name: &str) -> PathBuf {
+    PathBuf::from(env!("CARGO_MANIFEST_DIR"))
+        .join("tests")
+        .join("fixtures")
+        .join(name)
+}
+
+fn load_fixture(name: &str) -> (Lua, Table) {
+    let lua = Lua::new();
+    let src = fs::read_to_string(fixture_path(name)).unwrap();
+    let table: Table = lua.load(&src).set_name(name).eval().unwrap();
+    (lua, table)
+}
+
+#[test]
+fn lua_line_framer_fixture_splits_crlf_and_lf() {
+    let (lua, table) = load_fixture("lua_line_framer.lua");
+    let feed: Function = table.get("feed").unwrap();
+
+    let input = lua.create_string(b"hello\r\nworld\n").unwrap();
+    let frames: Vec<String> = feed.call(input).unwrap();
+    assert_eq!(frames, vec!["hello", "world"]);
+}
+
+#[test]
+fn lua_line_framer_fixture_handles_split_chunks_and_flush() {
+    let (lua, table) = load_fixture("lua_line_framer.lua");
+    let feed: Function = table.get("feed").unwrap();
+    let flush: Function = table.get("flush").unwrap();
+    let pending_len: Function = table.get("pending_len").unwrap();
+    let reset: Function = table.get("reset").unwrap();
+
+    let frames: Vec<String> = feed.call(lua.create_string(b"par").unwrap()).unwrap();
+    assert!(frames.is_empty());
+    assert_eq!(pending_len.call::<usize>(()).unwrap(), 3);
+
+    let frames: Vec<String> = feed.call(lua.create_string(b"tial\n").unwrap()).unwrap();
+    assert_eq!(frames, vec!["partial"]);
+
+    let flushed: Option<String> = flush.call(()).unwrap();
+    assert_eq!(flushed.as_deref(), None);
+
+    reset.call::<()>(()).unwrap();
+    assert_eq!(pending_len.call::<usize>(()).unwrap(), 0);
+}
+
+#[test]
+fn lua_text_decoder_fixture_decodes_text_and_rejects_empty() {
+    let (lua, table) = load_fixture("lua_text_decoder.lua");
+    let decode: Function = table.get("decode").unwrap();
+
+    let decoded: Option<Table> = decode.call(lua.create_string(b"abc").unwrap()).unwrap();
+    let decoded = decoded.expect("non-empty frame should decode");
+    let kind: String = decoded.get("kind").unwrap();
+    let data: String = decoded.get("data").unwrap();
+    assert_eq!(kind, "text");
+    assert_eq!(data, "abc");
+
+    let empty: Option<Table> = decode.call(lua.create_string(b"").unwrap()).unwrap();
+    assert!(empty.is_none(), "empty frame should be rejected");
+}

crates/pipeview-client/tests/lua_tests.rs

@@ -1,6 +1,7 @@
+use std::io::Write;
+use std::path::{Path, PathBuf};
 use std::sync::Once;
 use std::time::Duration;
-use std::{fs, path::PathBuf};
 
 use mlua::Lua;
 use tokio::io::{AsyncReadExt, AsyncWriteExt};
@@ -17,14 +18,30 @@ fn init_tracing() {
     });
 }
 
-fn write_temp_lua(name: &str, script: &str) -> PathBuf {
+struct TempScript {
-    let nonce = std::time::SystemTime::now()
+    path: PathBuf,
-        .duration_since(std::time::UNIX_EPOCH)
+    _guard: tempfile::TempPath,
-        .unwrap()
+}
-        .as_nanos();
+
-    let path = std::env::temp_dir().join(format!("pipeview_{name}_{nonce}.lua"));
+impl TempScript {
-    fs::write(&path, script).unwrap();
+    fn new(name: &str, script: &str) -> Self {
-    path
+        let mut file = tempfile::Builder::new()
+            .prefix(&format!("pipeview_{name}_"))
+            .suffix(".lua")
+            .tempfile()
+            .unwrap();
+        file.write_all(script.as_bytes()).unwrap();
+        let path = file.path().to_path_buf();
+        let guard = file.into_temp_path();
+        Self {
+            path,
+            _guard: guard,
+        }
+    }
+
+    fn path(&self) -> &Path {
+        &self.path
+    }
 }
 
 fn text_pipeline_lua() -> &'static str {
@@ -208,7 +225,7 @@ async fn lua_session_custom_lua_pipeline() {
     let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
     let addr = listener.local_addr().unwrap().to_string();
 
-    let framer_path = write_temp_lua(
+    let framer_script = TempScript::new(
         "custom_framer",
         r#"
         local buffer = ""
@@ -239,7 +256,7 @@ async fn lua_session_custom_lua_pipeline() {
         }
         "#,
     );
-    let decoder_path = write_temp_lua(
+    let decoder_script = TempScript::new(
         "custom_decoder",
         r#"
         return {
@@ -287,14 +304,12 @@ async fn lua_session_custom_lua_pipeline() {
         sess:close()
         "#,
         addr,
-        framer_path.display(),
+        framer_script.path().display(),
-        decoder_path.display()
+        decoder_script.path().display()
     );
 
     lua.load(&script).exec_async().await.unwrap();
     server.await.unwrap();
-    fs::remove_file(framer_path).unwrap();
-    fs::remove_file(decoder_path).unwrap();
 }
 
 // ── session:on_data callback ─────────────────────────────────────

crates/pipeview-core/Cargo.toml

@@ -15,3 +15,6 @@ serde_json = { workspace = true }
 tracing = { workspace = true }
 thiserror = { workspace = true }
 hex = "0.4"
+
+[dev-dependencies]
+proptest = "1"

crates/pipeview-core/tests/properties.rs

@@ -0,0 +1,96 @@
+use proptest::prelude::*;
+
+use pipeview_core::frame::Framer;
+use pipeview_core::frame::cobs::{CobsFramer, cobs_decode, cobs_encode};
+use pipeview_core::frame::length::{LengthConfig, LengthPrefixedFramer};
+use pipeview_core::protocol::Endian;
+use pipeview_core::protocol::ProtocolDecoder;
+use pipeview_core::protocol::hex::{HexConfig, HexDecoder};
+use pipeview_core::protocol::plot::{PlotConfig, PlotDecoder, SampleType};
+use pipeview_core::protocol::text::{TextDecoder, TextEncoding};
+
+proptest! {
+    #![proptest_config(ProptestConfig::with_cases(256))]
+
+    // ── COBS ─────────────────────────────────────────────────────────
+
+    #[test]
+    fn cobs_roundtrip_any_payload(payload in prop::collection::vec(any::<u8>(), 0..512)) {
+        let encoded = cobs_encode(&payload);
+        let decoded = cobs_decode(&encoded).unwrap();
+        prop_assert_eq!(&decoded, &payload);
+
+        let mut framer = CobsFramer::new(1024 * 1024);
+        let mut wire = encoded;
+        wire.push(0x00);
+        let frames = framer.feed(&wire);
+        prop_assert_eq!(frames.len(), 1);
+        prop_assert_eq!(&frames[0], &payload);
+    }
+
+    // ── Length-prefixed ──────────────────────────────────────────────
+
+    #[test]
+    fn length_roundtrip_any_payload(
+        len_bytes in prop_oneof![Just(1usize), Just(2usize), Just(4usize)],
+        little_endian in any::<bool>(),
+        includes_self in any::<bool>(),
+        payload in prop::collection::vec(any::<u8>(), 0..200)
+    ) {
+        let endian = if little_endian { Endian::Little } else { Endian::Big };
+        let raw_len = if includes_self {
+            payload.len() + len_bytes
+        } else {
+            payload.len()
+        };
+
+        let mut wire = Vec::with_capacity(len_bytes + payload.len());
+        match (len_bytes, endian) {
+            (1, _) => wire.push(raw_len as u8),
+            (2, Endian::Big) => wire.extend_from_slice(&(raw_len as u16).to_be_bytes()),
+            (2, Endian::Little) => wire.extend_from_slice(&(raw_len as u16).to_le_bytes()),
+            (4, Endian::Big) => wire.extend_from_slice(&(raw_len as u32).to_be_bytes()),
+            (4, Endian::Little) => wire.extend_from_slice(&(raw_len as u32).to_le_bytes()),
+            _ => unreachable!(),
+        }
+        wire.extend_from_slice(&payload);
+
+        let config = LengthConfig {
+            len_bytes,
+            endian,
+            length_includes_self: includes_self,
+            max_payload: 1024 * 1024,
+        };
+        let mut framer = LengthPrefixedFramer::new(config);
+        let frames = framer.feed(&wire);
+        prop_assert_eq!(frames.len(), 1);
+        prop_assert_eq!(&frames[0], &payload);
+    }
+
+    // ── Protocol decoders never panic on arbitrary bytes ─────────────
+
+    #[test]
+    fn plot_decode_arbitrary_bytes_never_panics(bytes in prop::collection::vec(any::<u8>(), 0..256)) {
+        let decoder = PlotDecoder::new(PlotConfig {
+            sample_type: SampleType::F32,
+            endian: Endian::Little,
+            channels: 2,
+            format: pipeview_core::protocol::plot::PlotFormat::Interleaved,
+        });
+        let _ = decoder.decode(&bytes);
+    }
+
+    #[test]
+    fn hex_decode_arbitrary_bytes_never_panics(bytes in prop::collection::vec(any::<u8>(), 0..256)) {
+        let decoder = HexDecoder::new(HexConfig::default());
+        let decoded = decoder.decode(&bytes);
+        prop_assert!(decoded.is_some());
+    }
+
+    #[test]
+    fn latin1_decode_arbitrary_bytes_never_panics(bytes in prop::collection::vec(any::<u8>(), 0..256)) {
+        let decoder = TextDecoder::new(TextEncoding::Latin1);
+        let decoded = decoder.decode(&bytes);
+        prop_assert!(decoded.is_some());
+    }
+}

crates/pipeview-gui/src/app.rs

@@ -4,7 +4,9 @@ use std::time::{Duration, Instant};
 
 use crate::buffers::{HexBuffer, PlotBuffer, TextBuffer};
 use crate::logging::{self, LogWriter};
-use crate::panels::{config, console, font_settings, hex_view, plot_view, send_panel, session_controls, sidebar};
+use crate::panels::{
+    config, console, font_settings, hex_view, plot_view, send_panel, session_controls, sidebar,
+};
 use crate::perf::{DrainStats, GuiProfiler, GuiSnapshot};
 use crate::shortcuts::{self, default_bindings};
 use crate::ui_fonts::{self, FontCandidate, UiFontSettings};
@@ -557,7 +559,10 @@ impl XserialApp {
                 let (badge, status_text) = tab.status.badge();
 
                 ui.horizontal(|ui| {
-                    ui.add(Label::new(egui::RichText::new(format!("Session {}", tab.id)).heading()).selectable(false));
+                    ui.add(
+                        Label::new(egui::RichText::new(format!("Session {}", tab.id)).heading())
+                            .selectable(false),
+                    );
                     ui.add(Label::new(badge).selectable(false));
                     ui.separator();
                     ui.add(Label::new(status_text).selectable(false));
@@ -579,7 +584,8 @@ impl XserialApp {
                     }
                 });
 
-                let auto_reconnect_changed = session_controls::render_session_controls(ui, &manager, tab);
+                let auto_reconnect_changed =
+                    session_controls::render_session_controls(ui, &manager, tab);
                 persist_state |= auto_reconnect_changed;
 
                 let mut display_changed = false;
@@ -598,7 +604,10 @@ impl XserialApp {
                 session_controls::render_search_bar(ui, tab);
 
                 if let ConnectionStatus::Error(message) = &tab.status {
-                    ui.add(Label::new(egui::RichText::new(message).color(Color32::RED)).selectable(false));
+                    ui.add(
+                        Label::new(egui::RichText::new(message).color(Color32::RED))
+                            .selectable(false),
+                    );
                 }
 
                 let full = ui.available_rect_before_wrap();
@@ -620,11 +629,21 @@ impl XserialApp {
                         let started = Instant::now();
                         match tab.view {
                             View::Text => {
-                                let line_count = console::render(ui, &tab.console, *display, tab.search.active.then_some(&tab.search));
+                                let line_count = console::render(
+                                    ui,
+                                    &tab.console,
+                                    *display,
+                                    tab.search.active.then_some(&tab.search),
+                                );
                                 text_render = Some((started.elapsed(), line_count));
                             }
                             View::Hex => {
-                                let line_count = hex_view::render(ui, &tab.hex, *display, tab.search.active.then_some(&tab.search));
+                                let line_count = hex_view::render(
+                                    ui,
+                                    &tab.hex,
+                                    *display,
+                                    tab.search.active.then_some(&tab.search),
+                                );
                                 hex_render = Some((started.elapsed(), line_count));
                             }
                             View::Plot => {

crates/pipeview-gui/src/panels/console.rs

@@ -37,8 +37,7 @@ pub fn render(
             input.pointer.button_down(egui::PointerButton::Primary)
                 && input.pointer.hover_pos().is_some_and(|pos| {
                     let area = ui.max_rect();
-                    pos.y > area.bottom() - EDGE_SCROLL_ZONE
+                    pos.y > area.bottom() - EDGE_SCROLL_ZONE && pos.y < area.bottom() + 20.0
-                        && pos.y < area.bottom() + 20.0
                 })
         });
 
@@ -56,9 +55,8 @@ pub fn render(
     // offset for the next frame and request a repaint for smooth continuous scrolling.
     if near_bottom_edge && line_count > 0 {
         let mut state = output.state;
-        let max_offset = (line_count as f32 * row_height)
+        let max_offset =
-            - output.inner_rect.height()
+            (line_count as f32 * row_height) - output.inner_rect.height() + EDGE_SCROLL_ZONE;
-            + EDGE_SCROLL_ZONE;
         state.offset.y = (state.offset.y + EDGE_SCROLL_NUDGE).min(max_offset.max(0.0));
         state.store(ui.ctx(), output.id);
         ui.ctx().request_repaint();
@@ -74,7 +72,7 @@ fn monospace_format(style: &egui::Style) -> TextFormat {
             .get(&TextStyle::Monospace)
             .cloned()
             .unwrap_or_default(),
-            color: Color32::WHITE,
+        color: Color32::WHITE,
         ..Default::default()
     }
 }

crates/pipeview-gui/src/panels/hex_view.rs

@@ -38,8 +38,7 @@ pub fn render(
             input.pointer.button_down(egui::PointerButton::Primary)
                 && input.pointer.hover_pos().is_some_and(|pos| {
                     let area = ui.max_rect();
-                    pos.y > area.bottom() - EDGE_SCROLL_ZONE
+                    pos.y > area.bottom() - EDGE_SCROLL_ZONE && pos.y < area.bottom() + 20.0
-                        && pos.y < area.bottom() + 20.0
                 })
         });
 
@@ -57,9 +56,8 @@ pub fn render(
     // offset for the next frame and request a repaint for smooth continuous scrolling.
     if near_bottom_edge {
         let mut state = output.state;
-        let max_offset = (line_count as f32 * row_height)
+        let max_offset =
-            - output.inner_rect.height()
+            (line_count as f32 * row_height) - output.inner_rect.height() + EDGE_SCROLL_ZONE;
-            + EDGE_SCROLL_ZONE;
         state.offset.y = (state.offset.y + EDGE_SCROLL_NUDGE).min(max_offset.max(0.0));
         state.store(ui.ctx(), output.id);
         ui.ctx().request_repaint();

crates/pipeview-gui/src/panels/send_panel.rs

@@ -44,18 +44,14 @@ pub fn render_send_panel(ui: &mut egui::Ui, manager: &SessionManager, tab: &mut
 
     let log_toggled = ui
         .horizontal(|ui| {
-            let toggled = ui
+            let toggled = ui.checkbox(&mut tab.log_enabled, "Log to file").changed();
-                .checkbox(&mut tab.log_enabled, "Log to file")
-                .changed();
             ui.checkbox(&mut tab.show_sent, "Show sent data");
             toggled
         })
         .inner;
     if log_toggled {
         if tab.log_enabled {
-            tab.log_path = default_log_path(tab.id)
+            tab.log_path = default_log_path(tab.id).to_string_lossy().to_string();
-                .to_string_lossy()
-                .to_string();
             tab.log_writer = LogWriter::open(&tab.log_path).ok();
         } else {
             tab.log_writer = None;
@@ -63,9 +59,7 @@ pub fn render_send_panel(ui: &mut egui::Ui, manager: &SessionManager, tab: &mut
     }
     if tab.log_enabled {
         ui.horizontal(|ui| {
-            let changed = ui
+            let changed = ui.text_edit_singleline(&mut tab.log_path).lost_focus();
-                .text_edit_singleline(&mut tab.log_path)
-                .lost_focus();
             if changed && !tab.log_path.is_empty() {
                 tab.log_writer = LogWriter::open(&tab.log_path).ok();
             }
@@ -96,13 +90,13 @@ pub fn render_send_panel(ui: &mut egui::Ui, manager: &SessionManager, tab: &mut
         send_clicked = ui.button("Send").clicked();
         if let Some(status) = &tab.send_status {
             ui.add(
-                Label::new(
+                Label::new(egui::RichText::new(status).color(
-                    egui::RichText::new(status).color(if status.starts_with("Send failed") {
+                    if status.starts_with("Send failed") {
                         Color32::RED
                     } else {
                         Color32::GRAY
-                    }),
+                    },
-                )
+                ))
                 .selectable(false),
             );
         }

crates/pipeview-gui/src/panels/session_controls.rs

@@ -19,7 +19,9 @@ pub fn render_search_bar(ui: &mut egui::Ui, tab: &mut SessionTab) {
         }
         if response.changed() {
             let matches = match tab.view {
-                View::Text => tab.console.search(&tab.search.query, tab.search.case_sensitive),
+                View::Text => tab
+                    .console
+                    .search(&tab.search.query, tab.search.case_sensitive),
                 View::Hex => tab.hex.search(&tab.search.query, tab.search.case_sensitive),
                 View::Plot => Vec::new(),
             };
@@ -48,7 +50,9 @@ pub fn render_search_bar(ui: &mut egui::Ui, tab: &mut SessionTab) {
 
         if ui.checkbox(&mut tab.search.case_sensitive, "Aa").changed() {
             let matches = match tab.view {
-                View::Text => tab.console.search(&tab.search.query, tab.search.case_sensitive),
+                View::Text => tab
+                    .console
+                    .search(&tab.search.query, tab.search.case_sensitive),
                 View::Hex => tab.hex.search(&tab.search.query, tab.search.case_sensitive),
                 View::Plot => Vec::new(),
             };

tools/pv_protocol.py

@@ -0,0 +1,158 @@
+#!/usr/bin/env python3
+"""Shared wire-format helpers for pipeview test tools.
+
+Keep these functions in sync with:
+  - crates/pipeview-core/src/frame/cobs.rs        (cobs_encode)
+  - crates/pipeview-core/src/protocol/mixed.rs    (XP plot packet header)
+"""
+
+import math
+import struct
+
+PLOT_ESCAPE = 0x1E
+PLOT_MARKER = ord("P")
+
+DISCONNECT_WINERRORS = {
+    10053,  # Software caused connection abort
+    10054,  # Connection reset by peer
+    10058,  # Socket shutdown race on Windows
+}
+
+# XP plot packet header field values (see protocol/mixed.rs)
+PLOT_PACKET_MAGIC = b"XP"
+PLOT_PACKET_VERSION = 1
+PLOT_FORMAT_IDS = {
+    "interleaved": 0,
+    "block": 1,
+    "xy": 2,
+}
+SAMPLE_TYPE_F32 = 8
+ENDIAN_LITTLE = 0
+
+
+def cobs_encode(payload: bytes) -> bytes:
+    """Consistent Overhead Byte Stuffing encoder (matches Rust cobs_encode)."""
+    if not payload:
+        return b"\x01"
+
+    out = bytearray([0])
+    code_index = 0
+    code = 1
+    for byte in payload:
+        if byte == 0:
+            out[code_index] = code
+            code_index = len(out)
+            out.append(0)
+            code = 1
+        else:
+            out.append(byte)
+            code += 1
+            if code == 0xFF:
+                out[code_index] = code
+                code_index = len(out)
+                out.append(0)
+                code = 1
+    out[code_index] = code
+    return bytes(out)
+
+
+def build_plot_payload(
+    sample_index: int,
+    channels: int,
+    plot_format: str,
+    samples_per_channel: int,
+    amplitude: float,
+    frequency_hz: float,
+    sample_rate_hz: float,
+) -> bytes:
+    """Build a raw little-endian f32 plot payload without any framing."""
+    if channels <= 0:
+        raise ValueError("channels must be >= 1")
+    if plot_format not in PLOT_FORMAT_IDS:
+        raise ValueError(f"unsupported plot format: {plot_format}")
+    if plot_format == "xy" and channels != 2:
+        raise ValueError("xy format requires exactly 2 channels")
+
+    values = []
+    if plot_format == "xy":
+        for offset in range(samples_per_channel):
+            t = (sample_index + offset) / sample_rate_hz
+            values.extend(
+                [
+                    amplitude * math.cos(2 * math.pi * frequency_hz * t),
+                    amplitude * math.sin(2 * math.pi * frequency_hz * t),
+                ]
+            )
+    elif plot_format == "block":
+        for ch in range(channels):
+            phase = 2 * math.pi * ch / channels
+            for offset in range(samples_per_channel):
+                t = (sample_index + offset) / sample_rate_hz
+                values.append(
+                    amplitude * math.sin(2 * math.pi * frequency_hz * t + phase)
+                )
+    else:  # interleaved
+        for offset in range(samples_per_channel):
+            t = (sample_index + offset) / sample_rate_hz
+            for ch in range(channels):
+                phase = 2 * math.pi * ch / channels
+                values.append(
+                    amplitude * math.sin(2 * math.pi * frequency_hz * t + phase)
+                )
+
+    return struct.pack(f"<{len(values)}f", *values)
+
+
+def build_plot_packet(
+    payload: bytes,
+    channels: int,
+    samples_per_channel: int,
+    plot_format: str,
+) -> bytes:
+    """Build an XP plot packet (the payload inside a MixedTextPlot COBS frame)."""
+    if not 1 <= channels <= 255:
+        raise ValueError("channels must be in 1..255")
+    if not 0 <= samples_per_channel <= 0xFFFF:
+        raise ValueError("samples_per_channel must fit in u16")
+    if plot_format not in PLOT_FORMAT_IDS:
+        raise ValueError(f"unsupported plot format: {plot_format}")
+
+    header = bytearray()
+    header.extend(PLOT_PACKET_MAGIC)
+    header.append(PLOT_PACKET_VERSION)
+    header.append(PLOT_FORMAT_IDS[plot_format])
+    header.append(SAMPLE_TYPE_F32)
+    header.append(ENDIAN_LITTLE)
+    header.append(channels)
+    header.extend(struct.pack("<H", samples_per_channel))
+    header.extend(struct.pack("<I", len(payload)))
+    header.extend(payload)
+    return bytes(header)
+
+
+def build_mixed_plot_frame(
+    payload: bytes,
+    channels: int,
+    samples_per_channel: int,
+    plot_format: str,
+) -> bytes:
+    """Build a complete MixedTextPlot plot frame (escape + marker + COBS packet + 0x00)."""
+    packet = build_plot_packet(
+        payload=payload,
+        channels=channels,
+        samples_per_channel=samples_per_channel,
+        plot_format=plot_format,
+    )
+    return bytes([PLOT_ESCAPE, PLOT_MARKER]) + cobs_encode(packet) + b"\x00"
+
+
+def is_disconnect_error(err: OSError) -> bool:
+    """Return True when the socket error is an expected client-disconnect error."""
+    return isinstance(
+        err,
+        (
+            BrokenPipeError,
+            ConnectionAbortedError,
+            ConnectionResetError,
+        ),
+    ) or getattr(err, "winerror", None) in DISCONNECT_WINERRORS

tools/stress_test.py

@@ -1,8 +1,8 @@
 #!/usr/bin/env python3
 """
-pipeview-tauri 极限压力测试脚本
+pipeview GUI（egui）极限压力测试脚本
 
-模拟各种高数据量场景，测试 Tauri 前端的渲染性能和稳定性：
+模拟各种高数据量场景，测试 pipeview GUI 的渲染性能和稳定性：
   - 文本洪水：大文本行 + ANSI 颜色
   - 十六进制洪水：大批量 hex dump
   - 波形洪水：高频 plot 采样点
@@ -20,20 +20,11 @@ import argparse
 import math
 import random
 import socket
-import struct
-import sys
 import threading
 import time
 from collections import defaultdict
 
-PLOT_ESCAPE = 0x1E
+import pv_protocol
-PLOT_MARKER = ord("P")
-
-DISCONNECT_WINERRORS = {
-    10053,  # Software caused connection abort
-    10054,  # Connection reset by peer
-    10058,  # Socket shutdown race on Windows
-}
 
 # ── ANSI Color Palette ─────────────────────────────────────────────
 
@@ -108,57 +99,32 @@ def build_plot_frame(
     frequency_hz: float,
     sample_rate_hz: float,
 ) -> bytes:
-    """Build a raw COBS-encoded plot frame (matching test_plot.py format)."""
+    """Build a raw little-endian f32 plot payload (no framing)."""
-    values = []
+    return pv_protocol.build_plot_payload(
-    if plot_format == "xy":
+        sample_index=sample_index,
-        for _ in range(samples_per_channel):
+        channels=channels,
-            t = (sample_index * samples_per_channel + len(values)) / sample_rate_hz
+        plot_format=plot_format,
-            values.append(amplitude * math.sin(2 * math.pi * frequency_hz * t))
+        samples_per_channel=samples_per_channel,
-            values.append(amplitude * math.cos(2 * math.pi * frequency_hz * t))
+        amplitude=amplitude,
-    elif plot_format == "block":
+        frequency_hz=frequency_hz,
-        for ch in range(channels):
+        sample_rate_hz=sample_rate_hz,
-            phase = 2 * math.pi * ch / channels
+    )
-            for _ in range(samples_per_channel):
-                t = (sample_index * samples_per_channel + len(values) // channels) / sample_rate_hz
-                values.append(amplitude * math.sin(2 * math.pi * frequency_hz * t + phase))
-    else:  # interleaved
-        for _ in range(samples_per_channel):
-            t = (sample_index * samples_per_channel + len(values) // channels) / sample_rate_hz
-            for ch in range(channels):
-                phase = 2 * math.pi * ch / channels
-                values.append(amplitude * math.sin(2 * math.pi * frequency_hz * t + phase))
-
-    # Pack as f32 little-endian
-    payload = struct.pack(f"<{len(values)}f", *values)
-    return cobs_encode(payload)
-
-
-def cobs_encode(data: bytes) -> bytes:
-    """Consistent Overhead Byte Stuffing encoder."""
-    result = bytearray()
-    block_start = 0
-    while block_start < len(data):
-        end = min(block_start + 254, len(data))
-        block = data[block_start:end]
-        if end < len(data):
-            result.append(len(block) + 1)
-            result.extend(block)
-        else:
-            result.append(len(block) + 1 if len(block) < 254 else 255)
-            result.extend(block)
-            result.append(0)
-        block_start = end
-    return bytes(result)
 
 
 def build_mixed_frame(seq: int, text_rate_per_plot: int, channels: int) -> bytes:
-    """Build a MixedTextPlot frame: text lines + one COBS plot frame."""
+    """Build a MixedTextPlot frame: text lines + one XP plot frame."""
     frames = []
     for i in range(text_rate_per_plot):
         frames.append(generate_text_line(seq * text_rate_per_plot + i, ansi=True))
-    # Add one plot frame per text_rate_per_plot text lines
+    plot_payload = build_plot_frame(seq, channels, "interleaved", 32, 100.0, 10.0, 1000.0)
-    plot_data = build_plot_frame(seq, channels, "interleaved", 32, 100.0, 10.0, 1000.0)
+    frames.append(
-    frames.append(bytes([PLOT_ESCAPE, PLOT_MARKER]) + plot_data)
+        pv_protocol.build_mixed_plot_frame(
+            payload=plot_payload,
+            channels=channels,
+            samples_per_channel=32,
+            plot_format="interleaved",
+        )
+    )
     return b"".join(frames)
 
 
@@ -273,7 +239,7 @@ class StressServer:
                     last_count_time = time.time()
 
         except OSError as e:
-            if hasattr(e, "winerror") and e.winerror in DISCONNECT_WINERRORS:
+            if pv_protocol.is_disconnect_error(e):
                 pass
             elif not self.running:
                 pass
@@ -324,7 +290,7 @@ class StressServer:
 
 def main():
     parser = argparse.ArgumentParser(
-        description="pipeview-tauri 极限压力测试",
+        description="pipeview GUI（egui）极限压力测试",
         formatter_class=argparse.RawDescriptionHelpFormatter,
         epilog="""
 示例:

tools/test_plot.py

@@ -4,115 +4,10 @@
 import argparse
 import math
 import socket
-import struct
 import threading
 import time
 
-PLOT_ESCAPE = 0x1E
+import pv_protocol
-PLOT_MARKER = ord("P")
-DISCONNECT_WINERRORS = {
-    10053,  # Software caused connection abort
-    10054,  # Connection reset by peer
-    10058,  # Socket shutdown race on Windows
-}
-
-
-def build_frame(
-    sample_index: int,
-    channels: int,
-    plot_format: str,
-    samples_per_channel: int,
-    amplitude: float,
-    frequency_hz: float,
-    sample_rate_hz: float,
-) -> bytes:
-    values = []
-    if plot_format == "xy":
-        for offset in range(samples_per_channel):
-            t = (sample_index + offset) / sample_rate_hz
-            x = amplitude * math.cos(2 * math.pi * frequency_hz * t)
-            y = amplitude * math.sin(2 * math.pi * frequency_hz * t)
-            values.extend([x, y])
-        return struct.pack(f"<{len(values)}f", *values)
-
-    for offset in range(samples_per_channel):
-        t = (sample_index + offset) / sample_rate_hz
-        for ch in range(channels):
-            phase = 2 * math.pi * ch / max(channels, 1)
-            value = amplitude * math.sin(2 * math.pi * frequency_hz * t + phase)
-            values.append(value)
-    return struct.pack(f"<{len(values)}f", *values)
-
-
-def cobs_encode(payload: bytes) -> bytes:
-    if not payload:
-        return b"\x01"
-
-    out = bytearray([0])
-    code_index = 0
-    code = 1
-    for byte in payload:
-        if byte == 0:
-            out[code_index] = code
-            code_index = len(out)
-            out.append(0)
-            code = 1
-        else:
-            out.append(byte)
-            code += 1
-            if code == 0xFF:
-                out[code_index] = code
-                code_index = len(out)
-                out.append(0)
-                code = 1
-    out[code_index] = code
-    return bytes(out)
-
-
-def build_plot_packet(
-    payload: bytes,
-    channels: int,
-    samples_per_channel: int,
-    plot_format: str,
-) -> bytes:
-    format_id = {"interleaved": 0, "block": 1, "xy": 2}[plot_format]
-    header = bytearray()
-    header.extend(b"XP")
-    header.append(1)  # version
-    header.append(format_id)
-    header.append(8)  # f32
-    header.append(0)  # little-endian
-    header.append(channels)
-    header.extend(struct.pack("<H", samples_per_channel))
-    header.extend(struct.pack("<I", len(payload)))
-    header.extend(payload)
-    return bytes(header)
-
-
-def build_mixed_plot_frame(
-    payload: bytes,
-    channels: int,
-    samples_per_channel: int,
-    plot_format: str,
-) -> bytes:
-    packet = build_plot_packet(
-        payload=payload,
-        channels=channels,
-        samples_per_channel=samples_per_channel,
-        plot_format=plot_format,
-    )
-    return bytes([PLOT_ESCAPE, PLOT_MARKER]) + cobs_encode(packet) + b"\x00"
-
-
-def is_disconnect_error(err: OSError) -> bool:
-    return isinstance(
-        err,
-        (
-            BrokenPipeError,
-            ConnectionAbortedError,
-            ConnectionResetError,
-        ),
-    ) or getattr(err, "winerror", None) in DISCONNECT_WINERRORS
 
 
 def main() -> None:
@@ -217,8 +112,8 @@ def main() -> None:
     else:
         print("  framer    = Line")
         print("  decoder   = Text")
-        print("  lua test  = Lua framer tests/lua_line_framer.lua")
+        print("  lua test  = Lua framer crates/pipeview-client/tests/fixtures/lua_line_framer.lua")
-        print("              Lua decoder tests/lua_text_decoder.lua\n")
+        print("              Lua decoder crates/pipeview-client/tests/fixtures/lua_text_decoder.lua\n")
     if args.wire_format == "text":
         print(f"sending text only at {args.text_interval:.3f}s intervals")
         print(f"sample clock: {args.rate:.1f} samples/sec\n")
@@ -268,7 +163,7 @@ def main() -> None:
                             next_text_at += args.text_interval
                             continue
 
-                        plot_payload = build_frame(
+                        plot_payload = pv_protocol.build_plot_payload(
                             sample_index=sample_index,
                             channels=args.channels,
                             plot_format=args.format,
@@ -278,7 +173,7 @@ def main() -> None:
                             sample_rate_hz=args.rate,
                         )
                         if args.wire_format == "mixed":
-                            frame = build_mixed_plot_frame(
+                            frame = pv_protocol.build_mixed_plot_frame(
                                 payload=plot_payload,
                                 channels=args.channels,
                                 samples_per_channel=samples_per_channel,
@@ -308,7 +203,7 @@ def main() -> None:
                         if wait > 0:
                             time.sleep(wait)
                 except OSError as err:
-                    if not is_disconnect_error(err):
+                    if not pv_protocol.is_disconnect_error(err):
                         raise
                     if args.wire_format == "text":
                         print(f"[conn -] {addr} ({text_count} text lines)")