Rewrite README in Chinese and English, add drone examples

Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent)

Co-authored-by: Sisyphus <clio-agent@sisyphuslabs.ai>

crates/pipeview-client/Cargo.toml

@@ -0,0 +1,17 @@
+[package]
+name = "pipeview-client"
+version = "0.1.0"
+edition = "2024"
+
+[dependencies]
+pipeview-core = { path = "../pipeview-core" }
+tokio = { workspace = true, features = ["sync", "time"] }
+tracing = { workspace = true }
+serde = { workspace = true }
+serde_json = { workspace = true }
+mlua = { workspace = true }
+thiserror = { workspace = true }
+
+[dev-dependencies]
+tokio = { workspace = true, features = ["full"] }
+tracing-subscriber = { workspace = true }

crates/pipeview-client/src/cmd.rs

@@ -0,0 +1,13 @@
+use crate::config::SessionConfig;
+
+pub enum SessionCmd {
+    Send(Vec<u8>),
+    Connect,
+    Disconnect,
+    Reconnect,
+    SetAutoReconnect(bool),
+    Close,
+    Reconfigure(SessionConfig),
+    SetDtr(bool),
+    SetRts(bool),
+}

crates/pipeview-client/src/config.rs

@@ -0,0 +1,478 @@
+use serde::{Deserialize, Serialize};
+use pipeview_core::frame::{
+    Endian, Framer,
+    cobs::CobsFramer,
+    fixed::FixedLengthFramer,
+    length::{LengthConfig, LengthPrefixedFramer},
+    line::{LineConfig, LineFramer},
+    mixed::{MixedTextPlotConfig as MixedFramerConfig, MixedTextPlotFramer},
+};
+use pipeview_core::protocol::{
+    ProtocolDecoder,
+    hex::{HexConfig, HexDecoder},
+    mixed::{MixedTextPlotConfig as MixedDecoderConfig, MixedTextPlotDecoder},
+    plot::{PlotConfig, PlotDecoder, PlotFormat, SampleType},
+    text::{TextDecoder, TextEncoding},
+};
+use pipeview_core::transport::TransportConfig;
+
+use crate::error::Result;
+use crate::lua::codec::{LuaDecoder, LuaFramer};
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub enum FramerConfig {
+    Line {
+        #[serde(default = "default_true")]
+        strip_cr: bool,
+        #[serde(default = "default_max_line")]
+        max_line_len: usize,
+    },
+    Fixed {
+        frame_len: usize,
+    },
+    Length {
+        #[serde(default = "default_len_bytes")]
+        len_bytes: usize,
+        #[serde(default)]
+        endian: Endian,
+        #[serde(default)]
+        length_includes_self: bool,
+        #[serde(default = "default_max_payload")]
+        max_payload: usize,
+    },
+    Cobs {
+        #[serde(default = "default_max_frame")]
+        max_frame: usize,
+    },
+    MixedTextPlot {
+        #[serde(default = "default_true")]
+        strip_cr: bool,
+        #[serde(default = "default_max_line")]
+        max_line_len: usize,
+        #[serde(default = "default_max_frame")]
+        max_plot_frame: usize,
+    },
+    Lua {
+        script_path: String,
+    },
+}
+
+fn default_true() -> bool {
+    true
+}
+fn default_max_line() -> usize {
+    1024 * 1024
+}
+fn default_len_bytes() -> usize {
+    2
+}
+fn default_max_payload() -> usize {
+    1024 * 1024
+}
+fn default_max_frame() -> usize {
+    1024 * 1024
+}
+
+impl FramerConfig {
+    pub fn build(self) -> Result<Box<dyn Framer>> {
+        Ok(match self {
+            FramerConfig::Line {
+                strip_cr,
+                max_line_len,
+            } => Box::new(LineFramer::new(LineConfig {
+                strip_cr,
+                max_line_len,
+            })) as Box<dyn Framer>,
+            FramerConfig::Fixed { frame_len } => Box::new(FixedLengthFramer::new(frame_len)),
+            FramerConfig::Length {
+                len_bytes,
+                endian,
+                length_includes_self,
+                max_payload,
+            } => Box::new(LengthPrefixedFramer::new(LengthConfig {
+                len_bytes,
+                endian,
+                length_includes_self,
+                max_payload,
+            })),
+            FramerConfig::Cobs { max_frame } => Box::new(CobsFramer::new(max_frame)),
+            FramerConfig::MixedTextPlot {
+                strip_cr,
+                max_line_len,
+                max_plot_frame,
+            } => Box::new(MixedTextPlotFramer::new(MixedFramerConfig {
+                strip_cr,
+                max_line_len,
+                max_plot_frame,
+            })),
+            FramerConfig::Lua { script_path } => {
+                Box::new(LuaFramer::from_script_path(script_path)?)
+            }
+        })
+    }
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub enum DecoderConfig {
+    Text {
+        #[serde(default)]
+        encoding: TextEncoding,
+    },
+    Hex {
+        #[serde(default)]
+        uppercase: bool,
+        #[serde(default = "default_sep")]
+        separator: String,
+        #[serde(default = "default_one")]
+        bytes_per_group: usize,
+        #[serde(default)]
+        endian: Endian,
+    },
+    Plot {
+        sample_type: SampleType,
+        #[serde(default)]
+        endian: Endian,
+        #[serde(default = "default_one")]
+        channels: usize,
+        #[serde(default)]
+        format: PlotFormat,
+    },
+    MixedTextPlot {
+        #[serde(default)]
+        encoding: TextEncoding,
+    },
+    Lua {
+        script_path: String,
+    },
+}
+
+fn default_sep() -> String {
+    String::from(" ")
+}
+fn default_one() -> usize {
+    1
+}
+
+impl DecoderConfig {
+    pub fn build(self) -> Result<Box<dyn ProtocolDecoder>> {
+        Ok(match self {
+            DecoderConfig::Text { encoding } => {
+                Box::new(TextDecoder::new(encoding)) as Box<dyn ProtocolDecoder>
+            }
+            DecoderConfig::Hex {
+                uppercase,
+                separator,
+                bytes_per_group,
+                endian,
+            } => Box::new(HexDecoder::new(HexConfig {
+                uppercase,
+                separator,
+                bytes_per_group,
+                endian,
+            })),
+            DecoderConfig::Plot {
+                sample_type,
+                endian,
+                channels,
+                format,
+            } => Box::new(PlotDecoder::new(PlotConfig {
+                sample_type,
+                endian,
+                channels,
+                format,
+            })),
+            DecoderConfig::MixedTextPlot { encoding } => {
+                Box::new(MixedTextPlotDecoder::new(MixedDecoderConfig { encoding }))
+            }
+            DecoderConfig::Lua { script_path } => {
+                Box::new(LuaDecoder::from_script_path(script_path)?)
+            }
+        })
+    }
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct PipelineConfig {
+    pub name: String,
+    pub framer: FramerConfig,
+    pub decoder: DecoderConfig,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct SessionConfig {
+    pub transport: TransportConfig,
+    #[serde(default)]
+    pub pipelines: Vec<PipelineConfig>,
+    #[serde(default = "default_history")]
+    pub history_limit: usize,
+    #[serde(default)]
+    pub auto_reconnect: bool,
+}
+
+fn default_history() -> usize {
+    10_000
+}
+
+impl Default for SessionConfig {
+    fn default() -> Self {
+        Self {
+            transport: TransportConfig::Tcp {
+                addr: "127.0.0.1:8080".into(),
+            },
+            pipelines: Vec::new(),
+            history_limit: default_history(),
+            auto_reconnect: false,
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use pipeview_core::protocol::Endian;
+    use pipeview_core::protocol::plot::{PlotFormat, SampleType};
+    use pipeview_core::protocol::text::TextEncoding;
+    use pipeview_core::transport::TransportConfig;
+
+    #[test]
+    fn framer_line_serde_roundtrip() {
+        let json = r#"{"Line":{"strip_cr":true,"max_line_len":4096}}"#;
+        let cfg: FramerConfig = serde_json::from_str(json).unwrap();
+        match &cfg {
+            FramerConfig::Line {
+                strip_cr,
+                max_line_len,
+            } => {
+                assert!(*strip_cr);
+                assert_eq!(*max_line_len, 4096);
+            }
+            _ => panic!("expected Line"),
+        }
+        let back = serde_json::to_string(&cfg).unwrap();
+        let _: FramerConfig = serde_json::from_str(&back).unwrap();
+    }
+
+    #[test]
+    fn framer_fixed_serde() {
+        let cfg: FramerConfig = serde_json::from_str(r#"{"Fixed":{"frame_len":128}}"#).unwrap();
+        assert!(matches!(cfg, FramerConfig::Fixed { frame_len: 128 }));
+    }
+
+    #[test]
+    fn framer_length_serde() {
+        let json = r#"{"Length":{"len_bytes":4,"endian":"Little","length_includes_self":true}}"#;
+        let cfg: FramerConfig = serde_json::from_str(json).unwrap();
+        match cfg {
+            FramerConfig::Length {
+                len_bytes,
+                endian,
+                length_includes_self,
+                ..
+            } => {
+                assert_eq!(len_bytes, 4);
+                assert_eq!(endian, Endian::Little);
+                assert!(length_includes_self);
+            }
+            _ => panic!("expected Length"),
+        }
+    }
+
+    #[test]
+    fn framer_cobs_serde() {
+        let cfg: FramerConfig = serde_json::from_str(r#"{"Cobs":{"max_frame":8192}}"#).unwrap();
+        assert!(matches!(cfg, FramerConfig::Cobs { max_frame: 8192 }));
+    }
+
+    #[test]
+    fn framer_mixed_serde() {
+        let json =
+            r#"{"MixedTextPlot":{"strip_cr":true,"max_line_len":4096,"max_plot_frame":8192}}"#;
+        let cfg: FramerConfig = serde_json::from_str(json).unwrap();
+        match cfg {
+            FramerConfig::MixedTextPlot {
+                strip_cr,
+                max_line_len,
+                max_plot_frame,
+            } => {
+                assert!(strip_cr);
+                assert_eq!(max_line_len, 4096);
+                assert_eq!(max_plot_frame, 8192);
+            }
+            _ => panic!("expected MixedTextPlot"),
+        }
+    }
+
+    #[test]
+    fn framer_lua_serde() {
+        let cfg: FramerConfig =
+            serde_json::from_str(r#"{"Lua":{"script_path":"/tmp/framer.lua"}}"#).unwrap();
+        match cfg {
+            FramerConfig::Lua { script_path } => assert_eq!(script_path, "/tmp/framer.lua"),
+            _ => panic!("expected Lua"),
+        }
+    }
+
+    #[test]
+    fn decoder_text_serde() {
+        let cfg: DecoderConfig = serde_json::from_str(r#"{"Text":{"encoding":"Latin1"}}"#).unwrap();
+        match cfg {
+            DecoderConfig::Text { encoding } => assert_eq!(encoding, TextEncoding::Latin1),
+            _ => panic!("expected Text"),
+        }
+    }
+
+    #[test]
+    fn decoder_hex_serde() {
+        let json =
+            r#"{"Hex":{"uppercase":true,"separator":":","bytes_per_group":2,"endian":"Little"}}"#;
+        let cfg: DecoderConfig = serde_json::from_str(json).unwrap();
+        match cfg {
+            DecoderConfig::Hex {
+                uppercase,
+                separator,
+                bytes_per_group,
+                endian,
+            } => {
+                assert!(uppercase);
+                assert_eq!(separator, ":");
+                assert_eq!(bytes_per_group, 2);
+                assert_eq!(endian, Endian::Little);
+            }
+            _ => panic!("expected Hex"),
+        }
+    }
+
+    #[test]
+    fn decoder_plot_serde() {
+        let json = r#"{"Plot":{"sample_type":"F32","endian":"Little","channels":2,"format":"Interleaved"}}"#;
+        let cfg: DecoderConfig = serde_json::from_str(json).unwrap();
+        match cfg {
+            DecoderConfig::Plot {
+                sample_type,
+                endian,
+                channels,
+                format,
+            } => {
+                assert_eq!(sample_type, SampleType::F32);
+                assert_eq!(channels, 2);
+                assert_eq!(format, PlotFormat::Interleaved);
+                assert_eq!(endian, Endian::Little);
+            }
+            _ => panic!("expected Plot"),
+        }
+    }
+
+    #[test]
+    fn decoder_mixed_serde() {
+        let cfg: DecoderConfig =
+            serde_json::from_str(r#"{"MixedTextPlot":{"encoding":"Utf8"}}"#).unwrap();
+        match cfg {
+            DecoderConfig::MixedTextPlot { encoding } => assert_eq!(encoding, TextEncoding::Utf8),
+            _ => panic!("expected MixedTextPlot"),
+        }
+    }
+
+    #[test]
+    fn decoder_lua_serde() {
+        let cfg: DecoderConfig =
+            serde_json::from_str(r#"{"Lua":{"script_path":"/tmp/decoder.lua"}}"#).unwrap();
+        match cfg {
+            DecoderConfig::Lua { script_path } => assert_eq!(script_path, "/tmp/decoder.lua"),
+            _ => panic!("expected Lua"),
+        }
+    }
+
+    #[test]
+    fn session_config_full_roundtrip() {
+        let json = r#"{
+            "transport":{"Tcp":{"addr":"192.168.1.1:9999"}},
+            "pipelines":[
+                {"name":"t","framer":{"Line":{}},"decoder":{"Text":{}}},
+                {"name":"h","framer":{"Fixed":{"frame_len":16}},"decoder":{"Hex":{}}}
+            ],
+            "history_limit":5000,
+            "auto_reconnect":true
+        }"#;
+        let cfg: SessionConfig = serde_json::from_str(json).unwrap();
+        assert!(cfg.auto_reconnect);
+        assert_eq!(cfg.history_limit, 5000);
+        assert_eq!(cfg.pipelines.len(), 2);
+        assert!(
+            matches!(&cfg.transport, TransportConfig::Tcp { addr } if addr == "192.168.1.1:9999")
+        );
+        let back = serde_json::to_string_pretty(&cfg).unwrap();
+        let _: SessionConfig = serde_json::from_str(&back).unwrap();
+    }
+
+    #[test]
+    fn framer_build_does_not_panic() {
+        for cfg in [
+            FramerConfig::Line {
+                strip_cr: true,
+                max_line_len: 256,
+            },
+            FramerConfig::Fixed { frame_len: 8 },
+            FramerConfig::Length {
+                len_bytes: 2,
+                endian: Endian::Big,
+                length_includes_self: false,
+                max_payload: 65536,
+            },
+            FramerConfig::Cobs { max_frame: 1024 },
+            FramerConfig::MixedTextPlot {
+                strip_cr: true,
+                max_line_len: 256,
+                max_plot_frame: 1024,
+            },
+        ] {
+            let f = cfg.build().unwrap();
+            assert_eq!(f.pending_len(), 0);
+        }
+    }
+
+    #[test]
+    fn decoder_build_returns_correct_name() {
+        assert_eq!(
+            DecoderConfig::Text {
+                encoding: TextEncoding::Utf8
+            }
+            .build()
+            .unwrap()
+            .name(),
+            "Text"
+        );
+        assert_eq!(
+            DecoderConfig::Hex {
+                uppercase: false,
+                separator: " ".into(),
+                bytes_per_group: 1,
+                endian: Endian::Big
+            }
+            .build()
+            .unwrap()
+            .name(),
+            "Hex"
+        );
+        assert_eq!(
+            DecoderConfig::Plot {
+                sample_type: SampleType::I16,
+                endian: Endian::Big,
+                channels: 2,
+                format: PlotFormat::XY
+            }
+            .build()
+            .unwrap()
+            .name(),
+            "Plot"
+        );
+        assert_eq!(
+            DecoderConfig::MixedTextPlot {
+                encoding: TextEncoding::Utf8
+            }
+            .build()
+            .unwrap()
+            .name(),
+            "MixedTextPlot"
+        );
+    }
+}

crates/pipeview-client/src/error.rs

@@ -0,0 +1,17 @@
+use thiserror::Error;
+
+#[derive(Error, Debug)]
+pub enum Error {
+    #[error("Connection failed: {0}")]
+    ConnectionFailed(String),
+    #[error("Session error: {0}")]
+    Session(String),
+    #[error("Lua error: {0}")]
+    Lua(#[from] mlua::Error),
+    #[error("I/O error: {0}")]
+    Io(#[from] std::io::Error),
+    #[error("{0}")]
+    Other(String),
+}
+
+pub type Result<T> = std::result::Result<T, Error>;

crates/pipeview-client/src/event.rs

@@ -0,0 +1,7 @@
+use pipeview_core::protocol::DecodedData;
+
+#[derive(Debug, Clone)]
+pub struct DecodedEntry {
+    pub pipeline_name: String,
+    pub data: DecodedData,
+}

crates/pipeview-client/src/history.rs

@@ -0,0 +1,59 @@
+use crate::event::DecodedEntry;
+use std::collections::VecDeque;
+
+pub struct RingBuffer<T> {
+    buf: VecDeque<T>,
+    limit: usize,
+}
+
+impl<T> RingBuffer<T> {
+    pub fn new(limit: usize) -> Self {
+        Self {
+            buf: VecDeque::with_capacity(limit.min(1024)),
+            limit,
+        }
+    }
+    pub fn push(&mut self, item: T) {
+        if self.buf.len() >= self.limit {
+            self.buf.pop_front();
+        }
+        self.buf.push_back(item);
+    }
+    pub fn len(&self) -> usize {
+        self.buf.len()
+    }
+    pub fn is_empty(&self) -> bool {
+        self.buf.is_empty()
+    }
+    pub fn iter(&self) -> impl Iterator<Item = &T> {
+        self.buf.iter()
+    }
+    pub fn get(&self, index: usize) -> Option<&T> {
+        self.buf.get(index)
+    }
+    pub fn clear(&mut self) {
+        self.buf.clear();
+    }
+    pub fn set_limit(&mut self, limit: usize) {
+        self.limit = limit;
+        while self.buf.len() > self.limit {
+            self.buf.pop_front();
+        }
+    }
+    pub fn drain_recent(&self, count: usize) -> Vec<T>
+    where
+        T: Clone,
+    {
+        let start = self.buf.len().saturating_sub(count);
+        self.buf.iter().skip(start).cloned().collect()
+    }
+}
+
+impl RingBuffer<DecodedEntry> {
+    pub fn entries_for_pipeline(&self, name: &str) -> Vec<&DecodedEntry> {
+        self.buf
+            .iter()
+            .filter(|e| e.pipeline_name == name)
+            .collect()
+    }
+}

crates/pipeview-client/src/lib.rs

@@ -0,0 +1,15 @@
+pub mod cmd;
+pub mod config;
+pub mod error;
+pub mod event;
+pub mod history;
+pub mod lua;
+pub mod manager;
+pub mod session;
+
+pub use config::{DecoderConfig, FramerConfig, PipelineConfig, SessionConfig};
+pub use error::{Error, Result};
+pub use event::DecodedEntry;
+pub use history::RingBuffer;
+pub use manager::SessionManager;
+pub use session::{Session, SessionEvent, SessionHandle, SessionId};

crates/pipeview-client/src/lua/codec.rs

@@ -0,0 +1,404 @@
+use std::fs;
+use std::sync::Mutex;
+
+use mlua::{Function, Lua, RegistryKey, Table, Value};
+use tracing::warn;
+use pipeview_core::frame::Framer;
+use pipeview_core::protocol::plot::{PlotFormat, PlotFrame, SampleType};
+use pipeview_core::protocol::{DecodedData, ProtocolDecoder};
+
+use crate::error::{Error, Result};
+
+pub struct LuaFramer {
+    lua: Lua,
+    feed: RegistryKey,
+    flush: RegistryKey,
+    reset: RegistryKey,
+    pending_len: RegistryKey,
+    script_path: String,
+}
+
+impl LuaFramer {
+    pub fn from_script_path(script_path: impl Into<String>) -> Result<Self> {
+        let script_path = script_path.into();
+        let lua = Lua::new();
+        let table = load_script_table(&lua, &script_path)?;
+        let feed = registry_function(&lua, &table, "feed", &script_path)?;
+        let flush = registry_function(&lua, &table, "flush", &script_path)?;
+        let reset = registry_function(&lua, &table, "reset", &script_path)?;
+        let pending_len = registry_function(&lua, &table, "pending_len", &script_path)?;
+
+        Ok(Self {
+            lua,
+            feed,
+            flush,
+            reset,
+            pending_len,
+            script_path,
+        })
+    }
+}
+
+impl Framer for LuaFramer {
+    fn feed(&mut self, data: &[u8]) -> Vec<Vec<u8>> {
+        let result = (|| -> mlua::Result<Vec<Vec<u8>>> {
+            let function: Function = self.lua.registry_value(&self.feed)?;
+            let bytes = self.lua.create_string(data)?;
+            frames_from_value(function.call::<Value>(bytes)?)
+        })();
+
+        match result {
+            Ok(frames) => frames,
+            Err(err) => {
+                warn!(script = %self.script_path, error = %err, "lua framer feed failed");
+                Vec::new()
+            }
+        }
+    }
+
+    fn flush(&mut self) -> Option<Vec<u8>> {
+        let result = (|| -> mlua::Result<Option<Vec<u8>>> {
+            let function: Function = self.lua.registry_value(&self.flush)?;
+            optional_bytes_from_value(function.call::<Value>(())?)
+        })();
+
+        match result {
+            Ok(frame) => frame,
+            Err(err) => {
+                warn!(script = %self.script_path, error = %err, "lua framer flush failed");
+                None
+            }
+        }
+    }
+
+    fn reset(&mut self) {
+        let result = (|| -> mlua::Result<()> {
+            let function: Function = self.lua.registry_value(&self.reset)?;
+            function.call::<()>(())
+        })();
+
+        if let Err(err) = result {
+            warn!(script = %self.script_path, error = %err, "lua framer reset failed");
+        }
+    }
+
+    fn pending_len(&self) -> usize {
+        let result = (|| -> mlua::Result<usize> {
+            let function: Function = self.lua.registry_value(&self.pending_len)?;
+            function.call::<usize>(())
+        })();
+
+        match result {
+            Ok(len) => len,
+            Err(err) => {
+                warn!(script = %self.script_path, error = %err, "lua framer pending_len failed");
+                0
+            }
+        }
+    }
+}
+
+pub struct LuaDecoder {
+    lua: Lua,
+    decode: RegistryKey,
+    script_path: String,
+    call_lock: Mutex<()>,
+}
+
+impl LuaDecoder {
+    pub fn from_script_path(script_path: impl Into<String>) -> Result<Self> {
+        let script_path = script_path.into();
+        let lua = Lua::new();
+        let table = load_script_table(&lua, &script_path)?;
+        let decode = registry_function(&lua, &table, "decode", &script_path)?;
+
+        Ok(Self {
+            lua,
+            decode,
+            script_path,
+            call_lock: Mutex::new(()),
+        })
+    }
+}
+
+impl ProtocolDecoder for LuaDecoder {
+    fn name(&self) -> &str {
+        "Lua"
+    }
+
+    fn decode(&self, frame: &[u8]) -> Option<DecodedData> {
+        let _guard = self
+            .call_lock
+            .lock()
+            .expect("lua decoder call mutex poisoned");
+        let result = (|| -> mlua::Result<Option<DecodedData>> {
+            let function: Function = self.lua.registry_value(&self.decode)?;
+            let bytes = self.lua.create_string(frame)?;
+            decoded_from_value(function.call::<Value>(bytes)?, frame)
+        })();
+
+        match result {
+            Ok(decoded) => decoded,
+            Err(err) => {
+                warn!(script = %self.script_path, error = %err, "lua decoder decode failed");
+                None
+            }
+        }
+    }
+}
+
+fn load_script_table(lua: &Lua, script_path: &str) -> Result<Table> {
+    let script = fs::read_to_string(script_path)?;
+    match lua.load(&script).eval::<Value>()? {
+        Value::Table(table) => Ok(table),
+        other => Err(Error::Other(format!(
+            "lua script '{script_path}' must return a table, got {}",
+            other.type_name()
+        ))),
+    }
+}
+
+fn registry_function(
+    lua: &Lua,
+    table: &Table,
+    name: &str,
+    script_path: &str,
+) -> Result<RegistryKey> {
+    let value: Value = table.get(name)?;
+    match value {
+        Value::Function(function) => Ok(lua.create_registry_value(function)?),
+        other => Err(Error::Other(format!(
+            "lua script '{script_path}' field '{name}' must be a function, got {}",
+            other.type_name()
+        ))),
+    }
+}
+
+fn frames_from_value(value: Value) -> mlua::Result<Vec<Vec<u8>>> {
+    match value {
+        Value::Nil => Ok(Vec::new()),
+        Value::String(bytes) => Ok(vec![bytes.as_bytes().to_vec()]),
+        Value::Table(table) => {
+            let mut frames = Vec::new();
+            for value in table.sequence_values::<mlua::String>() {
+                frames.push(value?.as_bytes().to_vec());
+            }
+            Ok(frames)
+        }
+        other => Err(mlua::Error::RuntimeError(format!(
+            "framer feed must return nil, string, or table of strings, got {}",
+            other.type_name()
+        ))),
+    }
+}
+
+fn optional_bytes_from_value(value: Value) -> mlua::Result<Option<Vec<u8>>> {
+    match value {
+        Value::Nil => Ok(None),
+        Value::String(bytes) => Ok(Some(bytes.as_bytes().to_vec())),
+        other => Err(mlua::Error::RuntimeError(format!(
+            "framer flush must return nil or string, got {}",
+            other.type_name()
+        ))),
+    }
+}
+
+fn decoded_from_value(value: Value, raw: &[u8]) -> mlua::Result<Option<DecodedData>> {
+    match value {
+        Value::Nil => Ok(None),
+        Value::String(text) => Ok(Some(DecodedData::Text(text.to_str()?.to_string()))),
+        Value::Table(table) => decoded_from_table(table, raw).map(Some),
+        other => Err(mlua::Error::RuntimeError(format!(
+            "decoder decode must return nil, string, or table, got {}",
+            other.type_name()
+        ))),
+    }
+}
+
+fn decoded_from_table(table: Table, raw: &[u8]) -> mlua::Result<DecodedData> {
+    let kind: String = table.get("kind")?;
+    match kind.as_str() {
+        "text" => Ok(DecodedData::Text(table.get("data")?)),
+        "hex" => Ok(DecodedData::Hex(table.get("data")?)),
+        "binary" => {
+            let bytes: mlua::String = table.get("data")?;
+            Ok(DecodedData::Binary(bytes.as_bytes().to_vec()))
+        }
+        "plot" => Ok(DecodedData::Plot(plot_from_table(table, raw)?)),
+        other => Err(mlua::Error::RuntimeError(format!(
+            "unknown lua decoder kind '{other}'"
+        ))),
+    }
+}
+
+fn plot_from_table(table: Table, raw: &[u8]) -> mlua::Result<PlotFrame> {
+    let channels_table: Table = table.get("channels")?;
+    let mut channels = Vec::new();
+    for channel in channels_table.sequence_values::<Table>() {
+        let channel = channel?;
+        let mut samples = Vec::new();
+        for sample in channel.sequence_values::<f64>() {
+            samples.push(sample?);
+        }
+        channels.push(samples);
+    }
+
+    Ok(PlotFrame {
+        channels,
+        raw: raw.to_vec(),
+        sample_type: match table.get::<Option<String>>("sample_type")?.as_deref() {
+            Some(value) => parse_sample_type(value)?,
+            None => SampleType::F64,
+        },
+        format: match table.get::<Option<String>>("format")?.as_deref() {
+            Some(value) => parse_plot_format(value)?,
+            None => PlotFormat::Interleaved,
+        },
+    })
+}
+
+fn parse_sample_type(value: &str) -> mlua::Result<SampleType> {
+    match value {
+        "i8" | "I8" => Ok(SampleType::I8),
+        "u8" | "U8" => Ok(SampleType::U8),
+        "i16" | "I16" => Ok(SampleType::I16),
+        "u16" | "U16" => Ok(SampleType::U16),
+        "i32" | "I32" => Ok(SampleType::I32),
+        "u32" | "U32" => Ok(SampleType::U32),
+        "i64" | "I64" => Ok(SampleType::I64),
+        "u64" | "U64" => Ok(SampleType::U64),
+        "f32" | "F32" => Ok(SampleType::F32),
+        "f64" | "F64" => Ok(SampleType::F64),
+        other => Err(mlua::Error::RuntimeError(format!(
+            "unknown plot sample_type '{other}'"
+        ))),
+    }
+}
+
+fn parse_plot_format(value: &str) -> mlua::Result<PlotFormat> {
+    match value {
+        "Interleaved" | "interleaved" => Ok(PlotFormat::Interleaved),
+        "Block" | "block" => Ok(PlotFormat::Block),
+        "XY" | "xy" => Ok(PlotFormat::XY),
+        other => Err(mlua::Error::RuntimeError(format!(
+            "unknown plot format '{other}'"
+        ))),
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::path::PathBuf;
+    use std::time::{SystemTime, UNIX_EPOCH};
+
+    fn write_script(name: &str, script: &str) -> PathBuf {
+        let nonce = SystemTime::now()
+            .duration_since(UNIX_EPOCH)
+            .unwrap()
+            .as_nanos();
+        let path = std::env::temp_dir().join(format!("pipeview_{name}_{nonce}.lua"));
+        fs::write(&path, script).unwrap();
+        path
+    }
+
+    #[test]
+    fn lua_framer_splits_lines_across_chunks() {
+        let path = write_script(
+            "framer",
+            r#"
+            local buffer = ""
+            return {
+                feed = function(bytes)
+                    buffer = buffer .. bytes
+                    local frames = {}
+                    while true do
+                        local i = buffer:find("\n", 1, true)
+                        if not i then break end
+                        frames[#frames + 1] = buffer:sub(1, i - 1)
+                        buffer = buffer:sub(i + 1)
+                    end
+                    return frames
+                end,
+                flush = function()
+                    if #buffer == 0 then return nil end
+                    local frame = buffer
+                    buffer = ""
+                    return frame
+                end,
+                reset = function()
+                    buffer = ""
+                end,
+                pending_len = function()
+                    return #buffer
+                end,
+            }
+            "#,
+        );
+
+        let mut framer = LuaFramer::from_script_path(path.to_string_lossy()).unwrap();
+        assert_eq!(framer.feed(b"one\nt"), vec![b"one".to_vec()]);
+        assert_eq!(framer.pending_len(), 1);
+        assert_eq!(framer.feed(b"wo\nthree"), vec![b"two".to_vec()]);
+        assert_eq!(framer.flush(), Some(b"three".to_vec()));
+        assert_eq!(framer.pending_len(), 0);
+
+        fs::remove_file(path).unwrap();
+    }
+
+    #[test]
+    fn lua_decoder_maps_text_binary_and_plot() {
+        let path = write_script(
+            "decoder",
+            r#"
+            return {
+                decode = function(frame)
+                    if frame == "skip" then
+                        return nil
+                    elseif frame == "bin" then
+                        return { kind = "binary", data = "\1\2" }
+                    elseif frame == "plot" then
+                        return {
+                            kind = "plot",
+                            channels = { { 1.0, 2.0 }, { 3.5, 4.5 } },
+                            sample_type = "F64",
+                            format = "Block",
+                        }
+                    end
+                    return { kind = "text", data = "decoded:" .. frame }
+                end,
+            }
+            "#,
+        );
+
+        let decoder = LuaDecoder::from_script_path(path.to_string_lossy()).unwrap();
+        assert!(decoder.decode(b"skip").is_none());
+        assert!(matches!(
+            decoder.decode(b"abc"),
+            Some(DecodedData::Text(text)) if text == "decoded:abc"
+        ));
+        assert!(matches!(
+            decoder.decode(b"bin"),
+            Some(DecodedData::Binary(bytes)) if bytes == vec![1, 2]
+        ));
+        match decoder.decode(b"plot").unwrap() {
+            DecodedData::Plot(frame) => {
+                assert_eq!(frame.channels, vec![vec![1.0, 2.0], vec![3.5, 4.5]]);
+                assert_eq!(frame.sample_type, SampleType::F64);
+                assert_eq!(frame.format, PlotFormat::Block);
+                assert_eq!(frame.raw, b"plot");
+            }
+            other => panic!("expected plot, got {other:?}"),
+        }
+
+        fs::remove_file(path).unwrap();
+    }
+
+    #[test]
+    fn lua_framer_requires_expected_functions() {
+        let path = write_script("bad_framer", "return { feed = function() return {} end }");
+        let result = LuaFramer::from_script_path(path.to_string_lossy());
+        assert!(result.is_err());
+        fs::remove_file(path).unwrap();
+    }
+}

crates/pipeview-client/src/lua/mod.rs

@@ -0,0 +1,170 @@
+use std::collections::HashMap;
+use std::sync::{Arc, Mutex, Weak};
+
+use mlua::{Lua, LuaSerdeExt, Result as LuaResult, Table, Value};
+
+use crate::config::SessionConfig;
+use crate::manager::SessionManager;
+
+pub(crate) mod codec;
+mod session_api;
+
+#[derive(Clone)]
+pub struct LuaRuntime {
+    manager: SessionManager,
+    callback_states: Arc<Mutex<HashMap<u64, Weak<session_api::CallbackState>>>>,
+}
+
+impl LuaRuntime {
+    pub fn new(manager: SessionManager) -> Self {
+        Self {
+            manager,
+            callback_states: Arc::new(Mutex::new(HashMap::new())),
+        }
+    }
+
+    pub fn manager(&self) -> &SessionManager {
+        &self.manager
+    }
+
+    pub fn register_callback_state(
+        &self,
+        session_id: u64,
+        state: &Arc<session_api::CallbackState>,
+    ) {
+        self.callback_states
+            .lock()
+            .expect("lua callback state map poisoned")
+            .insert(session_id, Arc::downgrade(state));
+    }
+
+    pub fn unregister_callback_state(&self, session_id: u64) {
+        self.callback_states
+            .lock()
+            .expect("lua callback state map poisoned")
+            .remove(&session_id);
+    }
+
+    pub async fn pump_callbacks(
+        &self,
+        lua: &Lua,
+        limit_per_session: Option<usize>,
+    ) -> LuaResult<usize> {
+        let states: Vec<(u64, Arc<session_api::CallbackState>)> = {
+            let mut states = self
+                .callback_states
+                .lock()
+                .expect("lua callback state map poisoned");
+            let mut live = Vec::with_capacity(states.len());
+            states.retain(|session_id, weak| {
+                if let Some(state) = weak.upgrade() {
+                    live.push((*session_id, state));
+                    true
+                } else {
+                    false
+                }
+            });
+            live
+        };
+
+        let mut pumped = 0;
+        for (_session_id, state) in states {
+            pumped += state.pump(lua, limit_per_session).await?;
+        }
+        Ok(pumped)
+    }
+}
+
+pub fn register(lua: &Lua) -> LuaResult<()> {
+    register_with_manager(lua, SessionManager::new())
+}
+
+pub fn register_with_manager(lua: &Lua, manager: SessionManager) -> LuaResult<()> {
+    let runtime = LuaRuntime::new(manager);
+    let pipeview = lua.create_table()?;
+
+    pipeview.set("open", {
+        let runtime = runtime.clone();
+        lua.create_async_function(move |lua, config: Table| {
+            let runtime = runtime.clone();
+            async move {
+                let cfg: SessionConfig = lua.from_value(Value::Table(config))?;
+                let handle = runtime.manager().create(cfg);
+                let userdata = session_api::LuaSessionHandle::new(handle, runtime);
+                lua.create_userdata(userdata)
+            }
+        })?
+    })?;
+
+    pipeview.set(
+        "list_ports",
+        lua.create_function(|_, ()| {
+            let ports = pipeview_core::transport::serial::SerialTransport::list_ports();
+            Ok(ports.into_iter().map(|p| p.port_name).collect::<Vec<_>>())
+        })?,
+    )?;
+
+    pipeview.set("sleep", {
+        let runtime = runtime.clone();
+        lua.create_async_function(move |lua, ms: u64| {
+            let runtime = runtime.clone();
+            async move {
+                tokio::time::sleep(std::time::Duration::from_millis(ms)).await;
+                let _ = runtime.pump_callbacks(&lua, None).await?;
+                Ok(())
+            }
+        })?
+    })?;
+
+    pipeview.set("poll", {
+        let runtime = runtime.clone();
+        lua.create_async_function(move |lua, limit_per_session: Option<usize>| {
+            let runtime = runtime.clone();
+            async move { runtime.pump_callbacks(&lua, limit_per_session).await }
+        })?
+    })?;
+
+    pipeview.set(
+        "log",
+        lua.create_function(|_, msg: String| {
+            tracing::info!("[lua] {}", msg);
+            Ok(())
+        })?,
+    )?;
+
+    lua.globals().set("pipeview", pipeview)?;
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use mlua::Value;
+
+    #[test]
+    fn register_exposes_expected_api() {
+        let lua = Lua::new();
+        register(&lua).unwrap();
+
+        let pipeview: Table = lua.globals().get("pipeview").unwrap();
+        for name in ["open", "list_ports", "sleep", "poll", "log"] {
+            let value: Value = pipeview.get(name).unwrap();
+            assert!(
+                matches!(value, Value::Function(_)),
+                "{name} should be a function"
+            );
+        }
+    }
+
+    #[test]
+    fn register_with_manager_uses_supplied_manager() {
+        let lua = Lua::new();
+        let manager = SessionManager::new();
+        register_with_manager(&lua, manager.clone()).unwrap();
+
+        let pipeview: Table = lua.globals().get("pipeview").unwrap();
+        let open: Value = pipeview.get("open").unwrap();
+        assert!(matches!(open, Value::Function(_)));
+        assert_eq!(manager.count(), 0);
+    }
+}

crates/pipeview-client/src/lua/session_api.rs

@@ -0,0 +1,535 @@
+use std::collections::HashMap;
+use std::sync::{
+    Arc, Mutex,
+    atomic::{AtomicU64, Ordering},
+};
+use std::time::Duration;
+
+use mlua::{Function, Lua, LuaSerdeExt, RegistryKey, UserData, UserDataMethods, Value, Variadic};
+use tokio::sync::mpsc;
+use tokio::task::JoinHandle;
+use pipeview_core::protocol::DecodedData;
+
+use crate::config::SessionConfig;
+use crate::lua::LuaRuntime;
+use crate::session::{SessionEvent, SessionHandle};
+
+pub struct LuaSessionHandle {
+    handle: SessionHandle,
+    runtime: LuaRuntime,
+    callbacks: Arc<CallbackState>,
+}
+
+pub struct CallbackState {
+    queue_tx: mpsc::UnboundedSender<QueuedEvent>,
+    queue_rx: Mutex<mpsc::UnboundedReceiver<QueuedEvent>>,
+    subscriptions: Mutex<HashMap<u64, Subscription>>,
+    next_subscription_id: AtomicU64,
+    relay_task: Mutex<Option<JoinHandle<()>>>,
+}
+
+struct Subscription {
+    callback: Arc<RegistryKey>,
+    filters: Vec<String>,
+}
+
+#[derive(Clone)]
+enum QueuedEvent {
+    Data {
+        pipeline_name: String,
+        data: QueuedData,
+    },
+}
+
+#[derive(Clone)]
+enum QueuedData {
+    Text(String),
+    Hex(String),
+    Binary(Vec<u8>),
+    Plot {
+        channels: Vec<Vec<f64>>,
+        sample_count: usize,
+    },
+}
+
+impl LuaSessionHandle {
+    pub fn new(handle: SessionHandle, runtime: LuaRuntime) -> Self {
+        let callbacks = Arc::new(CallbackState::new(handle.clone()));
+        runtime.register_callback_state(handle.id(), &callbacks);
+        Self {
+            handle,
+            runtime,
+            callbacks,
+        }
+    }
+}
+
+impl Drop for LuaSessionHandle {
+    fn drop(&mut self) {
+        self.runtime.unregister_callback_state(self.handle.id());
+    }
+}
+
+impl CallbackState {
+    pub fn new(handle: SessionHandle) -> Self {
+        let (queue_tx, queue_rx) = mpsc::unbounded_channel();
+        let state = Self {
+            queue_tx,
+            queue_rx: Mutex::new(queue_rx),
+            subscriptions: Mutex::new(HashMap::new()),
+            next_subscription_id: AtomicU64::new(1),
+            relay_task: Mutex::new(None),
+        };
+        state.ensure_relay(handle);
+        state
+    }
+
+    fn ensure_relay(&self, handle: SessionHandle) {
+        let mut relay_task = self
+            .relay_task
+            .lock()
+            .expect("lua relay task mutex poisoned");
+        if relay_task.is_some() {
+            return;
+        }
+
+        let tx = self.queue_tx.clone();
+        let mut rx = handle.subscribe();
+        *relay_task = Some(tokio::spawn(async move {
+            loop {
+                match rx.recv().await {
+                    Ok(SessionEvent::Data(_, entry)) => {
+                        let queued = match entry.data {
+                            DecodedData::Text(text) => QueuedEvent::Data {
+                                pipeline_name: entry.pipeline_name,
+                                data: QueuedData::Text(text),
+                            },
+                            DecodedData::Hex(hex) => QueuedEvent::Data {
+                                pipeline_name: entry.pipeline_name,
+                                data: QueuedData::Hex(hex),
+                            },
+                            DecodedData::Binary(bytes) => QueuedEvent::Data {
+                                pipeline_name: entry.pipeline_name,
+                                data: QueuedData::Binary(bytes),
+                            },
+                            DecodedData::Plot(frame) => QueuedEvent::Data {
+                                pipeline_name: entry.pipeline_name,
+                                data: QueuedData::Plot {
+                                    sample_count: frame.sample_count(),
+                                    channels: frame.channels,
+                                },
+                            },
+                        };
+                        if tx.send(queued).is_err() {
+                            break;
+                        }
+                    }
+                    Ok(SessionEvent::Closed(_)) => break,
+                    Ok(_) => {}
+                    Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => continue,
+                    Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
+                }
+            }
+        }));
+    }
+
+    pub fn add_subscription(&self, callback: RegistryKey, filters: Vec<String>) -> u64 {
+        let id = self.next_subscription_id.fetch_add(1, Ordering::SeqCst);
+        self.subscriptions
+            .lock()
+            .expect("lua callback subscriptions mutex poisoned")
+            .insert(
+                id,
+                Subscription {
+                    callback: Arc::new(callback),
+                    filters,
+                },
+            );
+        id
+    }
+
+    pub fn remove_subscription(&self, lua: &Lua, id: u64) -> mlua::Result<bool> {
+        let removed = self
+            .subscriptions
+            .lock()
+            .expect("lua callback subscriptions mutex poisoned")
+            .remove(&id);
+        if let Some(subscription) = removed {
+            if let Ok(key) = Arc::try_unwrap(subscription.callback) {
+                lua.remove_registry_value(key)?;
+            }
+            Ok(true)
+        } else {
+            Ok(false)
+        }
+    }
+
+    pub async fn pump(&self, lua: &Lua, limit: Option<usize>) -> mlua::Result<usize> {
+        let max = limit.unwrap_or(usize::MAX);
+        let mut drained = Vec::new();
+        {
+            let mut rx = self
+                .queue_rx
+                .lock()
+                .expect("lua callback queue mutex poisoned");
+            while drained.len() < max {
+                match rx.try_recv() {
+                    Ok(event) => drained.push(event),
+                    Err(tokio::sync::mpsc::error::TryRecvError::Empty) => break,
+                    Err(tokio::sync::mpsc::error::TryRecvError::Disconnected) => break,
+                }
+            }
+        }
+
+        let mut called = 0;
+        for event in drained {
+            let callbacks = self.matching_callbacks(&event);
+            for callback in callbacks {
+                let function: Function = lua.registry_value(&callback)?;
+                match &event {
+                    QueuedEvent::Data {
+                        pipeline_name,
+                        data,
+                    } => {
+                        let payload = queued_data_to_value(lua, data)?;
+                        function
+                            .call_async::<()>((pipeline_name.clone(), payload))
+                            .await?;
+                        called += 1;
+                    }
+                }
+            }
+        }
+
+        Ok(called)
+    }
+
+    fn matching_callbacks(&self, event: &QueuedEvent) -> Vec<Arc<RegistryKey>> {
+        let subscriptions = self
+            .subscriptions
+            .lock()
+            .expect("lua callback subscriptions mutex poisoned");
+        subscriptions
+            .values()
+            .filter(|subscription| match event {
+                QueuedEvent::Data { pipeline_name, .. } => {
+                    subscription.filters.is_empty()
+                        || subscription.filters.iter().any(|f| f == pipeline_name)
+                }
+            })
+            .map(|subscription| Arc::clone(&subscription.callback))
+            .collect()
+    }
+}
+
+impl Drop for CallbackState {
+    fn drop(&mut self) {
+        if let Ok(mut relay_task) = self.relay_task.lock()
+            && let Some(task) = relay_task.take()
+        {
+            task.abort();
+        }
+    }
+}
+
+impl UserData for LuaSessionHandle {
+    fn add_methods<M: UserDataMethods<Self>>(methods: &mut M) {
+        methods.add_async_method("send", |_, this, data: mlua::String| async move {
+            this.handle
+                .send(data.as_bytes().to_vec())
+                .await
+                .map_err(mlua::Error::RuntimeError)
+        });
+
+        methods.add_async_method("read", |lua, this, timeout_ms: Option<u64>| async move {
+            let ms = timeout_ms.unwrap_or(1000);
+            match this.handle.read(ms).await {
+                Some(entry) => Ok(Some(decoded_entry_to_table(
+                    &lua,
+                    entry.pipeline_name,
+                    entry.data,
+                )?)),
+                None => Ok(None),
+            }
+        });
+
+        methods.add_async_method(
+            "next_event",
+            |lua, this, timeout_ms: Option<u64>| async move {
+                let mut rx = this.handle.subscribe();
+                let timeout = Duration::from_millis(timeout_ms.unwrap_or(1000));
+                let next = tokio::time::timeout(timeout, async move {
+                    loop {
+                        match rx.recv().await {
+                            Ok(event) => return Some(event),
+                            Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => continue,
+                            Err(tokio::sync::broadcast::error::RecvError::Closed) => return None,
+                        }
+                    }
+                })
+                .await
+                .ok()
+                .flatten();
+
+                match next {
+                    Some(event) => Ok(Some(session_event_to_table(&lua, event)?)),
+                    None => Ok(None),
+                }
+            },
+        );
+
+        methods.add_async_method("close", |_, this, (): ()| async move {
+            this.handle.close().await.map_err(mlua::Error::RuntimeError)
+        });
+
+        methods.add_async_method("reconfigure", |lua, this, config: mlua::Table| async move {
+            let cfg: SessionConfig = lua.from_value(Value::Table(config))?;
+            this.handle
+                .reconfigure(cfg)
+                .await
+                .map_err(mlua::Error::RuntimeError)
+        });
+
+        methods.add_async_method(
+            "on_data",
+            |lua, this, (callback, pipelines): (Function, Variadic<String>)| async move {
+                let filters: Vec<String> = pipelines.into_iter().collect();
+                let key = lua.create_registry_value(callback)?;
+                Ok(this.callbacks.add_subscription(key, filters))
+            },
+        );
+
+        methods.add_method("off", |lua, this, token: u64| {
+            this.callbacks.remove_subscription(lua, token)
+        });
+
+        methods.add_async_method(
+            "pump_events",
+            |lua, this, limit: Option<usize>| async move { this.callbacks.pump(&lua, limit).await },
+        );
+    }
+}
+
+fn decoded_entry_to_table(
+    lua: &Lua,
+    pipeline_name: String,
+    data: DecodedData,
+) -> mlua::Result<mlua::Table> {
+    let table = lua.create_table()?;
+    table.set("type", "data")?;
+    table.set("pipeline", pipeline_name)?;
+    match data {
+        DecodedData::Text(text) => {
+            table.set("kind", "text")?;
+            table.set("data", text)?;
+        }
+        DecodedData::Hex(hex) => {
+            table.set("kind", "hex")?;
+            table.set("data", hex)?;
+        }
+        DecodedData::Binary(bytes) => {
+            table.set("kind", "binary")?;
+            table.set("data", lua.create_string(&bytes)?)?;
+        }
+        DecodedData::Plot(frame) => {
+            table.set("kind", "plot")?;
+            table.set("channels", channels_to_table(lua, &frame.channels)?)?;
+            table.set("sample_count", frame.sample_count())?;
+        }
+    }
+    Ok(table)
+}
+
+fn session_event_to_table(lua: &Lua, event: SessionEvent) -> mlua::Result<mlua::Table> {
+    match event {
+        SessionEvent::Connected(session_id) => {
+            let table = lua.create_table()?;
+            table.set("type", "connected")?;
+            table.set("session_id", session_id)?;
+            Ok(table)
+        }
+        SessionEvent::Disconnected(session_id) => {
+            let table = lua.create_table()?;
+            table.set("type", "disconnected")?;
+            table.set("session_id", session_id)?;
+            Ok(table)
+        }
+        SessionEvent::Closed(session_id) => {
+            let table = lua.create_table()?;
+            table.set("type", "closed")?;
+            table.set("session_id", session_id)?;
+            Ok(table)
+        }
+        SessionEvent::Error(session_id, error) => {
+            let table = lua.create_table()?;
+            table.set("type", "error")?;
+            table.set("session_id", session_id)?;
+            table.set("error", error)?;
+            Ok(table)
+        }
+        SessionEvent::Data(session_id, entry) => {
+            let table = decoded_entry_to_table(lua, entry.pipeline_name, entry.data)?;
+            table.set("session_id", session_id)?;
+            Ok(table)
+        }
+    }
+}
+
+fn queued_data_to_value(lua: &Lua, data: &QueuedData) -> mlua::Result<Value> {
+    match data {
+        QueuedData::Text(text) => Ok(Value::String(lua.create_string(text)?)),
+        QueuedData::Hex(hex) => Ok(Value::String(lua.create_string(hex)?)),
+        QueuedData::Binary(bytes) => Ok(Value::String(lua.create_string(bytes)?)),
+        QueuedData::Plot {
+            channels,
+            sample_count,
+        } => {
+            let table = lua.create_table()?;
+            table.set("kind", "plot")?;
+            table.set("channels", channels_to_table(lua, channels)?)?;
+            table.set("sample_count", *sample_count)?;
+            Ok(Value::Table(table))
+        }
+    }
+}
+
+fn channels_to_table(lua: &Lua, channels: &[Vec<f64>]) -> mlua::Result<mlua::Table> {
+    let table = lua.create_table()?;
+    for (index, channel) in channels.iter().enumerate() {
+        table.set(
+            index + 1,
+            lua.create_sequence_from(channel.iter().copied())?,
+        )?;
+    }
+    Ok(table)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::event::DecodedEntry;
+    use crate::session::SessionId;
+
+    fn test_callback_state() -> Arc<CallbackState> {
+        let (queue_tx, queue_rx) = mpsc::unbounded_channel();
+        Arc::new(CallbackState {
+            queue_tx,
+            queue_rx: Mutex::new(queue_rx),
+            subscriptions: Mutex::new(HashMap::new()),
+            next_subscription_id: AtomicU64::new(1),
+            relay_task: Mutex::new(None),
+        })
+    }
+
+    fn push_data_event(state: &CallbackState, pipeline_name: &str, data: QueuedData) {
+        state
+            .queue_tx
+            .send(QueuedEvent::Data {
+                pipeline_name: pipeline_name.to_string(),
+                data,
+            })
+            .unwrap();
+    }
+
+    #[tokio::test]
+    async fn callback_pump_invokes_matching_subscribers() {
+        let lua = Lua::new();
+        let received = Arc::new(Mutex::new(Vec::<(String, String)>::new()));
+        lua.set_app_data(received.clone());
+
+        let callback = lua
+            .create_function(|lua, (name, data): (String, String)| {
+                let store = lua
+                    .app_data_ref::<Arc<Mutex<Vec<(String, String)>>>>()
+                    .unwrap();
+                store.lock().unwrap().push((name, data));
+                Ok(())
+            })
+            .unwrap();
+
+        let state = test_callback_state();
+        let token = state.add_subscription(lua.create_registry_value(callback).unwrap(), vec![]);
+        assert_eq!(token, 1);
+
+        push_data_event(&state, "text", QueuedData::Text("hello".into()));
+        let called = state.pump(&lua, None).await.unwrap();
+
+        assert_eq!(called, 1);
+        assert_eq!(
+            received.lock().unwrap().as_slice(),
+            &[("text".to_string(), "hello".to_string())]
+        );
+    }
+
+    #[tokio::test]
+    async fn callback_pump_respects_pipeline_filters_and_off() {
+        let lua = Lua::new();
+        let calls = Arc::new(Mutex::new(0usize));
+        lua.set_app_data(calls.clone());
+
+        let callback = lua
+            .create_function(|lua, (_name, _data): (String, String)| {
+                let calls = lua.app_data_ref::<Arc<Mutex<usize>>>().unwrap();
+                *calls.lock().unwrap() += 1;
+                Ok(())
+            })
+            .unwrap();
+
+        let state = test_callback_state();
+        let token = state.add_subscription(
+            lua.create_registry_value(callback).unwrap(),
+            vec!["hex".into()],
+        );
+
+        push_data_event(&state, "text", QueuedData::Text("ignored".into()));
+        assert_eq!(state.pump(&lua, None).await.unwrap(), 0);
+        assert_eq!(*calls.lock().unwrap(), 0);
+
+        assert!(state.remove_subscription(&lua, token).unwrap());
+        push_data_event(&state, "hex", QueuedData::Hex("41".into()));
+        assert_eq!(state.pump(&lua, None).await.unwrap(), 0);
+        assert_eq!(*calls.lock().unwrap(), 0);
+    }
+
+    #[test]
+    fn decoded_entry_to_table_has_expected_shape() {
+        let lua = Lua::new();
+        let table =
+            decoded_entry_to_table(&lua, "text".into(), DecodedData::Text("payload".into()))
+                .unwrap();
+
+        assert_eq!(table.get::<String>("type").unwrap(), "data");
+        assert_eq!(table.get::<String>("pipeline").unwrap(), "text");
+        assert_eq!(table.get::<String>("kind").unwrap(), "text");
+        assert_eq!(table.get::<String>("data").unwrap(), "payload");
+    }
+
+    #[test]
+    fn session_event_to_table_maps_variants() {
+        let lua = Lua::new();
+
+        let error_table =
+            session_event_to_table(&lua, SessionEvent::Error(7 as SessionId, "boom".into()))
+                .unwrap();
+        assert_eq!(error_table.get::<String>("type").unwrap(), "error");
+        assert_eq!(error_table.get::<u64>("session_id").unwrap(), 7);
+        assert_eq!(error_table.get::<String>("error").unwrap(), "boom");
+
+        let data_table = session_event_to_table(
+            &lua,
+            SessionEvent::Data(
+                9,
+                DecodedEntry {
+                    pipeline_name: "hex".into(),
+                    data: DecodedData::Hex("41 42".into()),
+                },
+            ),
+        )
+        .unwrap();
+        assert_eq!(data_table.get::<String>("type").unwrap(), "data");
+        assert_eq!(data_table.get::<u64>("session_id").unwrap(), 9);
+        assert_eq!(data_table.get::<String>("pipeline").unwrap(), "hex");
+        assert_eq!(data_table.get::<String>("kind").unwrap(), "hex");
+        assert_eq!(data_table.get::<String>("data").unwrap(), "41 42");
+    }
+}

crates/pipeview-client/src/manager.rs

@@ -0,0 +1,220 @@
+use std::collections::HashMap;
+use std::sync::{
+    Arc, RwLock,
+    atomic::{AtomicU64, Ordering},
+};
+
+use tokio::sync::broadcast;
+use tracing::info;
+
+use crate::config::SessionConfig;
+use crate::session::{Session, SessionEvent, SessionHandle, SessionId};
+
+#[derive(Clone)]
+pub struct SessionManager {
+    inner: Arc<SessionManagerInner>,
+}
+
+struct SessionManagerInner {
+    sessions: RwLock<HashMap<SessionId, SessionHandle>>,
+    event_tx: broadcast::Sender<SessionEvent>,
+    next_session_id: AtomicU64,
+}
+
+impl SessionManager {
+    pub fn new() -> Self {
+        let (event_tx, _) = broadcast::channel(256);
+        Self {
+            inner: Arc::new(SessionManagerInner {
+                sessions: RwLock::new(HashMap::new()),
+                event_tx,
+                next_session_id: AtomicU64::new(1),
+            }),
+        }
+    }
+
+    pub fn create(&self, config: SessionConfig) -> SessionHandle {
+        let id = self.inner.next_session_id.fetch_add(1, Ordering::SeqCst);
+
+        let pipeline_names: Vec<&str> = config.pipelines.iter().map(|p| p.name.as_str()).collect();
+        info!(
+            session_id = id,
+            transport = ?config.transport,
+            pipelines = ?pipeline_names,
+            history_limit = config.history_limit,
+            auto_reconnect = config.auto_reconnect,
+            "session created"
+        );
+
+        let handle = Session::spawn(id, config);
+        let stored = handle.clone();
+        let mut session_events = handle.subscribe();
+        let relay = self.inner.event_tx.clone();
+
+        tokio::spawn(async move {
+            loop {
+                match session_events.recv().await {
+                    Ok(event) => {
+                        let _ = relay.send(event);
+                    }
+                    Err(broadcast::error::RecvError::Lagged(_)) => continue,
+                    Err(broadcast::error::RecvError::Closed) => break,
+                }
+            }
+        });
+
+        self.inner
+            .sessions
+            .write()
+            .expect("session store poisoned")
+            .insert(id, stored);
+
+        handle
+    }
+
+    pub fn remove(&self, id: SessionId) {
+        let removed = self
+            .inner
+            .sessions
+            .write()
+            .expect("session store poisoned")
+            .remove(&id);
+
+        if let Some(handle) = removed {
+            info!(session_id = id, "session removed");
+            tokio::spawn(async move {
+                let _ = handle.close().await;
+                handle.join().await;
+            });
+        }
+    }
+
+    pub fn get(&self, id: SessionId) -> Option<SessionHandle> {
+        self.inner
+            .sessions
+            .read()
+            .expect("session store poisoned")
+            .get(&id)
+            .cloned()
+    }
+
+    pub fn subscribe(&self) -> broadcast::Receiver<SessionEvent> {
+        self.inner.event_tx.subscribe()
+    }
+
+    pub fn list_ids(&self) -> Vec<SessionId> {
+        self.inner
+            .sessions
+            .read()
+            .expect("session store poisoned")
+            .keys()
+            .copied()
+            .collect()
+    }
+
+    pub fn count(&self) -> usize {
+        self.inner
+            .sessions
+            .read()
+            .expect("session store poisoned")
+            .len()
+    }
+}
+
+impl Default for SessionManager {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[tokio::test]
+    async fn create_keeps_monotonic_ids_after_remove() {
+        let manager = SessionManager::new();
+        let first = manager.create(SessionConfig::default());
+        assert_eq!(first.id(), 1);
+        manager.remove(first.id());
+        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
+        let second = manager.create(SessionConfig::default());
+        assert_eq!(second.id(), 2);
+    }
+
+    #[tokio::test]
+    async fn create_multiple_sessions() {
+        let manager = SessionManager::new();
+        let h1 = manager.create(SessionConfig::default());
+        let h2 = manager.create(SessionConfig::default());
+        let h3 = manager.create(SessionConfig::default());
+        assert_eq!(h1.id(), 1);
+        assert_eq!(h2.id(), 2);
+        assert_eq!(h3.id(), 3);
+    }
+
+    #[tokio::test]
+    async fn list_ids_returns_all() {
+        let manager = SessionManager::new();
+        manager.create(SessionConfig::default());
+        manager.create(SessionConfig::default());
+        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
+        let ids = manager.list_ids();
+        assert_eq!(ids.len(), 2);
+        assert!(ids.contains(&1));
+        assert!(ids.contains(&2));
+    }
+
+    #[tokio::test]
+    async fn get_returns_correct_session() {
+        let manager = SessionManager::new();
+        let h = manager.create(SessionConfig::default());
+        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
+        let found = manager.get(h.id());
+        assert!(found.is_some());
+        assert_eq!(found.unwrap().id(), h.id());
+    }
+
+    #[tokio::test]
+    async fn get_returns_none_for_missing() {
+        let manager = SessionManager::new();
+        assert!(manager.get(999).is_none());
+    }
+
+    #[tokio::test]
+    async fn remove_decrements_count() {
+        let manager = SessionManager::new();
+        let h = manager.create(SessionConfig::default());
+        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
+        manager.remove(h.id());
+        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
+        assert!(manager.get(h.id()).is_none());
+    }
+
+    #[tokio::test]
+    async fn subscribe_receives_events() {
+        let manager = SessionManager::new();
+        let mut rx = manager.subscribe();
+        manager.create(SessionConfig::default());
+        let event = tokio::time::timeout(std::time::Duration::from_secs(10), rx.recv()).await;
+        if let Ok(Ok(event)) = event {
+            assert!(matches!(
+                event,
+                SessionEvent::Connected(_) | SessionEvent::Error(_, _)
+            ));
+        }
+    }
+
+    #[tokio::test]
+    async fn count_starts_at_zero() {
+        let manager = SessionManager::new();
+        assert_eq!(manager.count(), 0);
+        assert!(manager.list_ids().is_empty());
+    }
+
+    #[tokio::test]
+    async fn default_creates_empty_manager() {
+        let manager = SessionManager::default();
+        assert_eq!(manager.count(), 0);
+    }
+}

crates/pipeview-client/src/session.rs

@@ -0,0 +1,702 @@
+use std::sync::{
+    Arc,
+    atomic::{AtomicBool, Ordering},
+};
+use std::time::Duration;
+use std::{future, io};
+
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio::select;
+use tokio::sync::{Mutex, broadcast, mpsc};
+use tokio::task::JoinHandle;
+use tracing::{debug, error, info, warn};
+
+use pipeview_core::pipeline::{MultiPipeline, Pipeline};
+use pipeview_core::transport::Connection;
+
+use crate::cmd::SessionCmd;
+use crate::config::SessionConfig;
+use crate::event::DecodedEntry;
+use crate::history::RingBuffer;
+
+pub type SessionId = u64;
+
+#[derive(Debug, Clone)]
+pub enum SessionEvent {
+    Connected(SessionId),
+    Disconnected(SessionId),
+    Data(SessionId, DecodedEntry),
+    Error(SessionId, String),
+    Closed(SessionId),
+}
+
+struct SessionHandleInner {
+    id: SessionId,
+    cmd_tx: mpsc::Sender<SessionCmd>,
+    event_tx: broadcast::Sender<SessionEvent>,
+    task: Mutex<Option<JoinHandle<()>>>,
+    close_requested: AtomicBool,
+}
+
+impl SessionHandleInner {
+    fn new(
+        id: SessionId,
+        cmd_tx: mpsc::Sender<SessionCmd>,
+        event_tx: broadcast::Sender<SessionEvent>,
+        task: JoinHandle<()>,
+    ) -> Self {
+        Self {
+            id,
+            cmd_tx,
+            event_tx,
+            task: Mutex::new(Some(task)),
+            close_requested: AtomicBool::new(false),
+        }
+    }
+
+    async fn request_close(&self) -> Result<(), String> {
+        self.close_requested.store(true, Ordering::SeqCst);
+        self.cmd_tx
+            .send(SessionCmd::Close)
+            .await
+            .map_err(|_| String::from("session closed"))
+    }
+
+    fn request_close_nonblocking(&self) {
+        self.close_requested.store(true, Ordering::SeqCst);
+        let _ = self.cmd_tx.try_send(SessionCmd::Close);
+    }
+}
+
+#[derive(Clone)]
+pub struct SessionHandle {
+    pub id: SessionId,
+    inner: Arc<SessionHandleInner>,
+}
+
+impl SessionHandle {
+    fn new(inner: Arc<SessionHandleInner>) -> Self {
+        Self {
+            id: inner.id,
+            inner,
+        }
+    }
+
+    pub fn id(&self) -> SessionId {
+        self.id
+    }
+
+    pub async fn send(&self, data: Vec<u8>) -> Result<(), String> {
+        self.inner
+            .cmd_tx
+            .send(SessionCmd::Send(data))
+            .await
+            .map_err(|_| String::from("session closed"))
+    }
+
+    pub async fn close(&self) -> Result<(), String> {
+        self.inner.request_close().await
+    }
+
+    pub async fn reconfigure(&self, config: SessionConfig) -> Result<(), String> {
+        self.inner
+            .cmd_tx
+            .send(SessionCmd::Reconfigure(config))
+            .await
+            .map_err(|_| String::from("session closed"))
+    }
+
+    pub async fn connect(&self) -> Result<(), String> {
+        self.inner
+            .cmd_tx
+            .send(SessionCmd::Connect)
+            .await
+            .map_err(|_| String::from("session closed"))
+    }
+
+    pub async fn disconnect(&self) -> Result<(), String> {
+        self.inner
+            .cmd_tx
+            .send(SessionCmd::Disconnect)
+            .await
+            .map_err(|_| String::from("session closed"))
+    }
+
+    pub async fn reconnect(&self) -> Result<(), String> {
+        self.inner
+            .cmd_tx
+            .send(SessionCmd::Reconnect)
+            .await
+            .map_err(|_| String::from("session closed"))
+    }
+
+    pub async fn set_auto_reconnect(&self, enabled: bool) -> Result<(), String> {
+        self.inner
+            .cmd_tx
+            .send(SessionCmd::SetAutoReconnect(enabled))
+            .await
+            .map_err(|_| String::from("session closed"))
+    }
+
+    pub async fn set_dtr(&self, state: bool) -> Result<(), String> {
+        self.inner
+            .cmd_tx
+            .send(SessionCmd::SetDtr(state))
+            .await
+            .map_err(|_| String::from("session closed"))
+    }
+
+    pub async fn set_rts(&self, state: bool) -> Result<(), String> {
+        self.inner
+            .cmd_tx
+            .send(SessionCmd::SetRts(state))
+            .await
+            .map_err(|_| String::from("session closed"))
+    }
+
+    pub async fn read(&self, timeout_ms: u64) -> Option<DecodedEntry> {
+        let mut rx = self.inner.event_tx.subscribe();
+        let deadline = Duration::from_millis(timeout_ms);
+        tokio::time::timeout(deadline, async {
+            loop {
+                match rx.recv().await {
+                    Ok(SessionEvent::Data(_, entry)) => return Some(entry),
+                    Ok(SessionEvent::Error(_, _) | SessionEvent::Closed(_)) => return None,
+                    Ok(SessionEvent::Connected(_) | SessionEvent::Disconnected(_)) => {}
+                    Err(broadcast::error::RecvError::Lagged(_)) => continue,
+                    Err(broadcast::error::RecvError::Closed) => return None,
+                }
+            }
+        })
+        .await
+        .ok()
+        .flatten()
+    }
+
+    pub fn subscribe(&self) -> broadcast::Receiver<SessionEvent> {
+        self.inner.event_tx.subscribe()
+    }
+
+    pub(crate) async fn join(&self) {
+        let task = {
+            let mut slot = self.inner.task.lock().await;
+            slot.take()
+        };
+        if let Some(task) = task {
+            let _ = task.await;
+        }
+    }
+}
+
+impl Drop for SessionHandle {
+    fn drop(&mut self) {
+        if Arc::strong_count(&self.inner) == 1 {
+            self.inner.request_close_nonblocking();
+            if let Ok(mut task) = self.inner.task.try_lock()
+                && let Some(task) = task.take()
+            {
+                task.abort();
+            }
+        }
+    }
+}
+
+pub struct Session {
+    id: SessionId,
+    config: SessionConfig,
+    conn: Option<Connection>,
+    pipeline: MultiPipeline,
+    pipeline_build_error: Option<String>,
+    history: RingBuffer<DecodedEntry>,
+    cmd_rx: mpsc::Receiver<SessionCmd>,
+    event_tx: broadcast::Sender<SessionEvent>,
+    close_requested: bool,
+    desired_connected: bool,
+}
+
+impl Session {
+    pub fn spawn(id: SessionId, config: SessionConfig) -> SessionHandle {
+        let (cmd_tx, cmd_rx) = mpsc::channel(32);
+        let (event_tx, _) = broadcast::channel(256);
+        let (pipeline, pipeline_build_error) = match Self::build_pipeline(&config) {
+            Ok(pipeline) => (pipeline, None),
+            Err(err) => (
+                MultiPipeline::new(),
+                Some(format!("pipeline build failed: {err}")),
+            ),
+        };
+        let mut session = Self {
+            id,
+            pipeline,
+            pipeline_build_error,
+            history: RingBuffer::new(config.history_limit),
+            config,
+            conn: None,
+            cmd_rx,
+            event_tx: event_tx.clone(),
+            close_requested: false,
+            desired_connected: true,
+        };
+        let task = tokio::spawn(async move { session.run().await });
+        let inner = Arc::new(SessionHandleInner::new(id, cmd_tx, event_tx, task));
+        SessionHandle::new(inner)
+    }
+
+    fn build_pipeline(config: &SessionConfig) -> crate::Result<MultiPipeline> {
+        let mut pipelines = MultiPipeline::new();
+        for pipeline in &config.pipelines {
+            pipelines.add(Pipeline::new(
+                pipeline.name.clone(),
+                pipeline.framer.clone().build()?,
+                pipeline.decoder.clone().build()?,
+            ));
+        }
+        Ok(pipelines)
+    }
+
+    async fn run(&mut self) {
+        info!(session_id = self.id, "session started");
+
+        if let Some(message) = self.pipeline_build_error.take() {
+            self.emit_error(message);
+            self.desired_connected = false;
+        }
+
+        if self.desired_connected
+            && let Err(err) = self.connect().await
+        {
+            warn!(session_id = self.id, error = %err, "initial connect failed");
+            self.emit_error(format!("connect failed: {err}"));
+        }
+
+        loop {
+            let reconnect_timer = self.reconnect_timer();
+            select! {
+                cmd = self.cmd_rx.recv() => {
+                    if !self.handle_command(cmd).await {
+                        break;
+                    }
+                }
+                read = try_read(&mut self.conn, &mut self.pipeline) => {
+                    if !self.handle_read_result(read).await {
+                        break;
+                    }
+                }
+                _ = reconnect_timer => {
+                    if let Err(err) = self.connect().await {
+                        self.emit_error(format!("connect failed: {err}"));
+                    }
+                }
+            }
+        }
+
+        self.disconnect(false).await;
+        self.emit_closed();
+        info!(session_id = self.id, "session ended");
+    }
+
+    async fn handle_command(&mut self, cmd: Option<SessionCmd>) -> bool {
+        match cmd {
+            Some(SessionCmd::Close) => {
+                self.close_requested = true;
+                self.desired_connected = false;
+                false
+            }
+            Some(SessionCmd::Connect) => {
+                self.desired_connected = true;
+                if self.conn.is_none()
+                    && let Err(err) = self.connect().await
+                {
+                    self.emit_error(format!("connect failed: {err}"));
+                }
+                true
+            }
+            Some(SessionCmd::Disconnect) => {
+                self.desired_connected = false;
+                self.disconnect(true).await;
+                true
+            }
+            Some(SessionCmd::Reconnect) => {
+                self.desired_connected = true;
+                self.disconnect(true).await;
+                if let Err(err) = self.connect().await {
+                    self.emit_error(format!("reconnect: {err}"));
+                }
+                true
+            }
+            Some(SessionCmd::SetAutoReconnect(enabled)) => {
+                self.config.auto_reconnect = enabled;
+                true
+            }
+            Some(SessionCmd::Send(data)) => {
+                self.handle_send(data).await;
+                true
+            }
+            Some(SessionCmd::Reconfigure(config)) => {
+                self.handle_reconfigure(config).await;
+                true
+            }
+            Some(SessionCmd::SetDtr(state)) => {
+                self.handle_set_dtr(state);
+                true
+            }
+            Some(SessionCmd::SetRts(state)) => {
+                self.handle_set_rts(state);
+                true
+            }
+            None => false,
+        }
+    }
+
+    async fn handle_send(&mut self, data: Vec<u8>) {
+        debug!(session_id = self.id, len = data.len(), "sending data");
+        match self.conn.as_mut() {
+            Some(conn) => {
+                if let Err(err) = conn.write_all(&data).await {
+                    self.emit_error(format!("write: {err}"));
+                }
+            }
+            None => self.emit_error(String::from("write: session not connected")),
+        }
+    }
+
+    async fn handle_reconfigure(&mut self, config: SessionConfig) {
+        info!(session_id = self.id, "reconfiguring session");
+        let pipeline = match Self::build_pipeline(&config) {
+            Ok(pipeline) => pipeline,
+            Err(err) => {
+                self.emit_error(format!("pipeline build failed: {err}"));
+                return;
+            }
+        };
+        self.disconnect(true).await;
+        self.config = config;
+        self.pipeline = pipeline;
+        self.history = RingBuffer::new(self.config.history_limit);
+
+        if let Err(err) = self.connect().await {
+            self.emit_error(format!("reconnect: {err}"));
+        }
+    }
+
+    fn handle_set_dtr(&mut self, state: bool) {
+        match self.conn.as_mut() {
+            Some(conn) => {
+                if let Err(err) = conn.set_dtr(state) {
+                    self.emit_error(format!("set_dtr: {err}"));
+                }
+            }
+            None => self.emit_error(String::from("set_dtr: session not connected")),
+        }
+    }
+
+    fn handle_set_rts(&mut self, state: bool) {
+        match self.conn.as_mut() {
+            Some(conn) => {
+                if let Err(err) = conn.set_rts(state) {
+                    self.emit_error(format!("set_rts: {err}"));
+                }
+            }
+            None => self.emit_error(String::from("set_rts: session not connected")),
+        }
+    }
+
+    async fn handle_read_result(
+        &mut self,
+        result: Result<Vec<DecodedEntry>, std::io::Error>,
+    ) -> bool {
+        match result {
+            Ok(entries) => {
+                for entry in entries {
+                    self.history.push(entry.clone());
+                    let _ = self.event_tx.send(SessionEvent::Data(self.id, entry));
+                }
+                true
+            }
+            Err(err) => {
+                error!(session_id = self.id, error = %err, "read error");
+                self.emit_error(err.to_string());
+                self.disconnect(true).await;
+                true
+            }
+        }
+    }
+
+    fn reconnect_timer(&self) -> impl std::future::Future<Output = ()> + Send + 'static {
+        let should_retry = self.conn.is_none()
+            && self.desired_connected
+            && self.config.auto_reconnect
+            && !self.close_requested;
+        async move {
+            if should_retry {
+                tokio::time::sleep(Duration::from_secs(1)).await;
+            } else {
+                future::pending::<()>().await;
+            }
+        }
+    }
+
+    async fn connect(&mut self) -> Result<(), io::Error> {
+        if self.conn.is_some() {
+            return Ok(());
+        }
+        let transport = self.config.transport.clone();
+        debug!(session_id = self.id, ?transport, "connecting session");
+        let mut conn = Connection::new(transport);
+        conn.connect().await.map_err(to_io_error)?;
+        self.conn = Some(conn);
+        let _ = self.event_tx.send(SessionEvent::Connected(self.id));
+        info!(session_id = self.id, "session connected");
+        Ok(())
+    }
+
+    async fn disconnect(&mut self, emit_disconnected: bool) {
+        if let Some(mut conn) = self.conn.take() {
+            let _ = conn.disconnect().await;
+            if emit_disconnected {
+                let _ = self.event_tx.send(SessionEvent::Disconnected(self.id));
+            }
+            info!(session_id = self.id, "session disconnected");
+        }
+        self.pipeline.reset();
+    }
+
+    fn emit_error(&self, message: String) {
+        let _ = self.event_tx.send(SessionEvent::Error(self.id, message));
+    }
+
+    fn emit_closed(&self) {
+        let _ = self.event_tx.send(SessionEvent::Closed(self.id));
+    }
+}
+
+async fn try_read(
+    conn: &mut Option<Connection>,
+    pipeline: &mut MultiPipeline,
+) -> Result<Vec<DecodedEntry>, io::Error> {
+    if conn.is_none() {
+        future::pending::<()>().await;
+        unreachable!("pending future should never resolve");
+    }
+
+    let conn = conn.as_mut().expect("connection presence checked");
+
+    let mut buf = [0u8; 4096];
+    match conn.read(&mut buf).await {
+        Ok(0) => Err(io::Error::new(
+            io::ErrorKind::UnexpectedEof,
+            "connection closed",
+        )),
+        Ok(n) => Ok(pipeline
+            .feed(&buf[..n])
+            .into_iter()
+            .map(|result| DecodedEntry {
+                pipeline_name: result.pipeline_name,
+                data: result.data,
+            })
+            .collect()),
+        Err(err) => Err(err),
+    }
+}
+
+fn to_io_error(err: pipeview_core::error::Error) -> std::io::Error {
+    io::Error::other(err.to_string())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::config::{DecoderConfig, FramerConfig, PipelineConfig};
+
+    fn tcp_config() -> SessionConfig {
+        SessionConfig {
+            transport: pipeview_core::transport::TransportConfig::Tcp {
+                addr: "127.0.0.1:1".into(),
+            },
+            pipelines: vec![PipelineConfig {
+                name: "text".into(),
+                framer: FramerConfig::Line {
+                    strip_cr: true,
+                    max_line_len: 65536,
+                },
+                decoder: DecoderConfig::Text {
+                    encoding: pipeview_core::protocol::text::TextEncoding::Utf8,
+                },
+            }],
+            ..Default::default()
+        }
+    }
+
+    #[test]
+    fn session_id_is_u64() {
+        let _id: SessionId = 42;
+    }
+
+    #[test]
+    fn session_event_debug() {
+        let e1 = SessionEvent::Connected(1);
+        let e2 = SessionEvent::Disconnected(1);
+        let e3 = SessionEvent::Closed(1);
+        let e4 = SessionEvent::Error(1, "oops".into());
+        let _ = format!("{:?}", e1);
+        let _ = format!("{:?}", e2);
+        let _ = format!("{:?}", e3);
+        let _ = format!("{:?}", e4);
+    }
+
+    #[test]
+    fn session_event_clone() {
+        let e1 = SessionEvent::Connected(1);
+        let e2 = e1.clone();
+        match (e1, e2) {
+            (SessionEvent::Connected(a), SessionEvent::Connected(b)) => assert_eq!(a, b),
+            _ => panic!(),
+        }
+    }
+
+    #[tokio::test]
+    async fn spawn_and_close_session() {
+        let handle = Session::spawn(1, tcp_config());
+        let mut rx = handle.subscribe();
+        let event = tokio::time::timeout(Duration::from_secs(5), rx.recv())
+            .await
+            .expect("timeout");
+        if let Ok(event) = event {
+            assert!(matches!(
+                event,
+                SessionEvent::Error(_, _) | SessionEvent::Connected(_)
+            ));
+        }
+        let _ = handle.close().await;
+        handle.join().await;
+    }
+
+    #[tokio::test]
+    async fn handle_send_to_dead_session() {
+        let handle = Session::spawn(2, tcp_config());
+        tokio::time::sleep(Duration::from_millis(200)).await;
+        let _ = handle.send(b"data".to_vec()).await;
+        let _ = handle.close().await;
+        handle.join().await;
+    }
+
+    #[tokio::test]
+    async fn handle_close_twice_is_idempotent() {
+        let handle = Session::spawn(3, tcp_config());
+        let _ = handle.close().await;
+        let _ = handle.close().await;
+        handle.join().await;
+    }
+
+    #[tokio::test]
+    async fn handle_read_after_close() {
+        let handle = Session::spawn(4, tcp_config());
+        let _ = handle.close().await;
+        let result = handle.read(100).await;
+        assert!(result.is_none());
+        handle.join().await;
+    }
+
+    #[tokio::test]
+    async fn handle_read_timeout_returns_none() {
+        let handle = Session::spawn(5, tcp_config());
+        let result = handle.read(100).await;
+        assert!(result.is_none());
+        let _ = handle.close().await;
+        handle.join().await;
+    }
+
+    #[tokio::test]
+    async fn handle_subscribe_receives_events() {
+        let handle = Session::spawn(6, tcp_config());
+        let mut rx = handle.subscribe();
+        let event = tokio::time::timeout(Duration::from_secs(5), rx.recv())
+            .await
+            .expect("timeout");
+        if let Ok(event) = event {
+            assert!(matches!(
+                event,
+                SessionEvent::Error(_, _) | SessionEvent::Connected(_)
+            ));
+        }
+        let _ = handle.close().await;
+        handle.join().await;
+    }
+
+    #[tokio::test]
+    async fn handle_id_matches() {
+        let handle = Session::spawn(99, tcp_config());
+        assert_eq!(handle.id(), 99);
+        let _ = handle.close().await;
+        handle.join().await;
+    }
+
+    #[tokio::test]
+    async fn handle_reconfigure_triggers_reconnect() {
+        let handle = Session::spawn(7, tcp_config());
+        let new_cfg = tcp_config();
+        let _ = handle.reconfigure(new_cfg).await;
+        let _ = handle.close().await;
+        handle.join().await;
+    }
+
+    #[tokio::test]
+    async fn try_read_without_connection_stays_pending() {
+        let mut conn = None;
+        let mut pipeline = MultiPipeline::new();
+
+        let result = tokio::time::timeout(
+            Duration::from_millis(50),
+            try_read(&mut conn, &mut pipeline),
+        )
+        .await;
+
+        assert!(result.is_err());
+    }
+
+    #[tokio::test]
+    async fn handle_drop_closes_session() {
+        let handle = Session::spawn(8, tcp_config());
+        let mut rx = handle.subscribe();
+        drop(handle);
+        let mut closed = false;
+        while let Ok(Ok(event)) = tokio::time::timeout(Duration::from_millis(200), rx.recv()).await
+        {
+            if matches!(event, SessionEvent::Closed(_)) {
+                closed = true;
+                break;
+            }
+        }
+        let _ = closed;
+    }
+
+    #[tokio::test]
+    async fn multiple_subscribers_all_receive() {
+        let handle = Session::spawn(9, tcp_config());
+        let mut rx1 = handle.subscribe();
+        let mut rx2 = handle.subscribe();
+
+        let e1 = tokio::time::timeout(Duration::from_secs(5), rx1.recv())
+            .await
+            .unwrap();
+        if let Ok(e1) = e1 {
+            assert!(matches!(
+                e1,
+                SessionEvent::Connected(9) | SessionEvent::Error(9, _)
+            ));
+        }
+
+        let e2 = tokio::time::timeout(Duration::from_secs(1), rx2.recv())
+            .await
+            .unwrap();
+        if let Ok(e2) = e2 {
+            assert!(matches!(
+                e2,
+                SessionEvent::Connected(9) | SessionEvent::Error(9, _)
+            ));
+        }
+
+        let _ = handle.close().await;
+        handle.join().await;
+    }
+}

crates/pipeview-client/tests/lua_tests.rs

@@ -0,0 +1,508 @@
+use std::sync::Once;
+use std::time::Duration;
+use std::{fs, path::PathBuf};
+
+use mlua::Lua;
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio::net::TcpListener;
+
+static INIT: Once = Once::new();
+
+fn init_tracing() {
+    INIT.call_once(|| {
+        tracing_subscriber::fmt()
+            .with_env_filter("pipeview=trace")
+            .try_init()
+            .ok();
+    });
+}
+
+fn write_temp_lua(name: &str, script: &str) -> PathBuf {
+    let nonce = std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .unwrap()
+        .as_nanos();
+    let path = std::env::temp_dir().join(format!("pipeview_{name}_{nonce}.lua"));
+    fs::write(&path, script).unwrap();
+    path
+}
+
+fn text_pipeline_lua() -> &'static str {
+    r#"
+        {
+        name = "text",
+        framer = { Line = {} },
+        decoder = { Text = {} }
+        }
+    "#
+}
+
+fn hex_pipeline_lua() -> &'static str {
+    r#"
+        {
+        name = "hex",
+        framer = { Line = {} },
+        decoder = { Hex = { uppercase = true } }
+        }
+    "#
+}
+
+// ── pipeview.sleep / pipeview.log ──────────────────────────────────
+
+#[tokio::test]
+async fn lua_sleep_and_log() {
+    let lua = Lua::new();
+    pipeview_client::lua::register(&lua).unwrap();
+
+    lua.load(
+        r#"
+        pipeview.log("test start")
+        pipeview.sleep(50)
+        pipeview.log("test end")
+        "#,
+    )
+    .exec_async()
+    .await
+    .unwrap();
+}
+
+// ── pipeview.list_ports ───────────────────────────────────────────
+
+#[tokio::test]
+async fn lua_list_ports() {
+    let lua = Lua::new();
+    pipeview_client::lua::register(&lua).unwrap();
+
+    let result: Vec<String> = lua
+        .load(r#"return pipeview.list_ports()"#)
+        .eval_async()
+        .await
+        .unwrap();
+
+    // Should return a table (possibly empty if no serial ports)
+    // Just verify it doesn't crash
+    let _ = result;
+}
+
+// ── pipeview.open + session:send / session:read / session:close ────
+
+#[tokio::test]
+async fn lua_session_open_send_read_close() {
+    init_tracing();
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr = listener.local_addr().unwrap().to_string();
+
+    let server = tokio::spawn(async move {
+        let (mut stream, _) = listener.accept().await.unwrap();
+        let mut buf = [0u8; 256];
+        let n = stream.read(&mut buf).await.unwrap();
+        assert_eq!(&buf[..n], b"hello\n");
+        stream.write_all(b"world\n").await.unwrap();
+        tokio::time::sleep(Duration::from_millis(200)).await;
+    });
+
+    let lua = Lua::new();
+    pipeview_client::lua::register(&lua).unwrap();
+
+    let script = format!(
+        r#"
+        local sess = pipeview.open({{
+            transport = {{ Tcp = {{ addr = "{}" }} }},
+            pipelines = {{{}}}
+        }})
+
+        sess:send("hello\n")
+
+        local r = sess:read(5000)
+        assert(r ~= nil, "expected data")
+        assert(r.pipeline == "text", "expected text pipeline")
+        assert(r.data == "world", "expected 'world', got " .. tostring(r.data))
+
+        sess:close()
+        "#,
+        addr,
+        text_pipeline_lua()
+    );
+
+    lua.load(&script).exec_async().await.unwrap();
+    server.await.unwrap();
+}
+
+// ── session:read with timeout ────────────────────────────────────
+
+#[tokio::test]
+async fn lua_session_read_timeout() {
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr = listener.local_addr().unwrap().to_string();
+
+    let _server = tokio::spawn(async move {
+        let (_stream, _) = listener.accept().await.unwrap();
+        tokio::time::sleep(std::time::Duration::from_secs(10)).await;
+    });
+
+    let lua = Lua::new();
+    pipeview_client::lua::register(&lua).unwrap();
+
+    let script = format!(
+        r#"
+        local sess = pipeview.open({{
+            transport = {{ Tcp = {{ addr = "{}" }} }},
+            pipelines = {{{}}}
+        }})
+
+        local r = sess:read(100)
+        assert(r == nil, "read should time out")
+
+        sess:close()
+        "#,
+        addr,
+        text_pipeline_lua()
+    );
+
+    lua.load(&script).exec_async().await.unwrap();
+}
+
+// ── pipeview.open with hex decoder ────────────────────────────────
+
+#[tokio::test]
+async fn lua_session_hex_decoder() {
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr = listener.local_addr().unwrap().to_string();
+
+    let server = tokio::spawn(async move {
+        let (mut stream, _) = listener.accept().await.unwrap();
+        stream.write_all(b"ABC\n").await.unwrap();
+        tokio::time::sleep(Duration::from_millis(200)).await;
+    });
+
+    let lua = Lua::new();
+    pipeview_client::lua::register(&lua).unwrap();
+
+    let script = format!(
+        r#"
+        local sess = pipeview.open({{
+            transport = {{ Tcp = {{ addr = "{}" }} }},
+            pipelines = {{{}}}
+        }})
+
+        local r = sess:read(5000)
+        assert(r ~= nil)
+        assert(r.pipeline == "hex")
+        assert(r.kind == "hex")
+        assert(r.data == "41 42 43", "expected hex of ABC, got " .. tostring(r.data))
+
+        sess:close()
+        "#,
+        addr,
+        hex_pipeline_lua()
+    );
+
+    lua.load(&script).exec_async().await.unwrap();
+    server.await.unwrap();
+}
+
+// ── pipeview.open with Lua framer + Lua decoder ───────────────────
+
+#[tokio::test]
+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(
+        "custom_framer",
+        r#"
+        local buffer = ""
+        return {
+            feed = function(bytes)
+                buffer = buffer .. bytes
+                local frames = {}
+                while true do
+                    local i = buffer:find("|", 1, true)
+                    if not i then break end
+                    frames[#frames + 1] = buffer:sub(1, i - 1)
+                    buffer = buffer:sub(i + 1)
+                end
+                return frames
+            end,
+            flush = function()
+                if #buffer == 0 then return nil end
+                local frame = buffer
+                buffer = ""
+                return frame
+            end,
+            reset = function()
+                buffer = ""
+            end,
+            pending_len = function()
+                return #buffer
+            end,
+        }
+        "#,
+    );
+    let decoder_path = write_temp_lua(
+        "custom_decoder",
+        r#"
+        return {
+            decode = function(frame)
+                return { kind = "text", data = string.upper(frame) }
+            end,
+        }
+        "#,
+    );
+
+    let server = tokio::spawn(async move {
+        let (mut stream, _) = listener.accept().await.unwrap();
+        stream.write_all(b"alpha|").await.unwrap();
+        tokio::time::sleep(Duration::from_millis(50)).await;
+        stream.write_all(b"beta|").await.unwrap();
+        tokio::time::sleep(Duration::from_millis(200)).await;
+    });
+
+    let lua = Lua::new();
+    pipeview_client::lua::register(&lua).unwrap();
+
+    let script = format!(
+        r#"
+        local sess = pipeview.open({{
+            transport = {{ Tcp = {{ addr = "{}" }} }},
+            pipelines = {{
+                {{
+                    name = "custom",
+                    framer = {{ Lua = {{ script_path = [[{}]] }} }},
+                    decoder = {{ Lua = {{ script_path = [[{}]] }} }}
+                }}
+            }}
+        }})
+
+        local r1 = sess:read(5000)
+        assert(r1 ~= nil, "expected first custom frame")
+        assert(r1.pipeline == "custom", "expected custom pipeline, got " .. tostring(r1.pipeline))
+        assert(r1.kind == "text", "expected text kind, got " .. tostring(r1.kind))
+        assert(r1.data == "ALPHA", "expected ALPHA, got " .. tostring(r1.data))
+
+        local r2 = sess:read(5000)
+        assert(r2 ~= nil, "expected second custom frame")
+        assert(r2.data == "BETA", "expected BETA, got " .. tostring(r2.data))
+
+        sess:close()
+        "#,
+        addr,
+        framer_path.display(),
+        decoder_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 ─────────────────────────────────────
+
+#[tokio::test]
+async fn lua_session_on_data_callback() {
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr = listener.local_addr().unwrap().to_string();
+
+    let server = tokio::spawn(async move {
+        let (mut stream, _) = listener.accept().await.unwrap();
+        stream.write_all(b"line1\nline2\n").await.unwrap();
+        tokio::time::sleep(Duration::from_millis(200)).await;
+    });
+
+    let lua = Lua::new();
+    pipeview_client::lua::register(&lua).unwrap();
+
+    let script = format!(
+        r#"
+        local received = {{}}
+        local sess = pipeview.open({{
+            transport = {{ Tcp = {{ addr = "{}" }} }},
+            pipelines = {{{}}}
+        }})
+
+        sess:on_data(function(name, data)
+            table.insert(received, {{ name = name, data = data }})
+        end)
+
+        for _ = 1, 50 do
+            if #received < 2 then
+                pipeview.poll(1)
+                pipeview.sleep(10)
+            end
+        end
+
+        assert(#received == 2, "expected 2 callbacks, got " .. #received)
+        assert(received[1].name == "text")
+        assert(received[1].data == "line1")
+        assert(received[2].name == "text")
+        assert(received[2].data == "line2")
+
+        sess:close()
+        "#,
+        addr,
+        text_pipeline_lua()
+    );
+
+    lua.load(&script).exec_async().await.unwrap();
+    server.await.unwrap();
+}
+
+// ── session:reconfigure from Lua ─────────────────────────────────
+
+#[tokio::test]
+async fn lua_session_reconfigure() {
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr = listener.local_addr().unwrap().to_string();
+
+    let server = tokio::spawn(async move {
+        let (mut stream, _) = listener.accept().await.unwrap();
+        stream.write_all(b"test\n").await.unwrap();
+        tokio::time::sleep(Duration::from_millis(300)).await;
+    });
+
+    let listener2 = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr2 = listener2.local_addr().unwrap().to_string();
+
+    let server2 = tokio::spawn(async move {
+        let (mut stream, _) = listener2.accept().await.unwrap();
+        stream.write_all(b"ABC\n").await.unwrap();
+        tokio::time::sleep(Duration::from_millis(300)).await;
+    });
+
+    let lua = Lua::new();
+    pipeview_client::lua::register(&lua).unwrap();
+
+    let script = format!(
+        r#"
+        local sess = pipeview.open({{
+            transport = {{ Tcp = {{ addr = "{}" }} }},
+            pipelines = {{{}}}
+        }})
+
+        local r = sess:read(5000)
+        assert(r.data == "test")
+
+        -- Reconfigure to hex on different address
+        sess:reconfigure({{
+            transport = {{ Tcp = {{ addr = "{}" }} }},
+            pipelines = {{{}}}
+        }})
+
+        local r2 = sess:read(5000)
+        assert(r2.pipeline == "hex")
+        assert(r2.data == "41 42 43")
+
+        sess:close()
+        "#,
+        addr,
+        text_pipeline_lua(),
+        addr2,
+        hex_pipeline_lua()
+    );
+
+    lua.load(&script).exec_async().await.unwrap();
+    server.await.unwrap();
+    server2.await.unwrap();
+}
+
+// ── session:next_event ───────────────────────────────────────────
+
+#[tokio::test]
+async fn lua_session_next_event_stream() {
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr = listener.local_addr().unwrap().to_string();
+
+    let server = tokio::spawn(async move {
+        let (mut stream, _) = listener.accept().await.unwrap();
+        stream.write_all(b"streamed\n").await.unwrap();
+        tokio::time::sleep(Duration::from_millis(200)).await;
+    });
+
+    let lua = Lua::new();
+    pipeview_client::lua::register(&lua).unwrap();
+
+    let script = format!(
+        r#"
+        local sess = pipeview.open({{
+            transport = {{ Tcp = {{ addr = "{}" }} }},
+            pipelines = {{{}}}
+        }})
+
+        local e1 = sess:next_event(5000)
+        assert(e1 ~= nil, "expected connected event")
+        assert(e1.type == "connected", "expected connected, got " .. tostring(e1.type))
+
+        local e2 = sess:next_event(5000)
+        assert(e2 ~= nil, "expected data event")
+        assert(e2.type == "data", "expected data, got " .. tostring(e2.type))
+        assert(e2.pipeline == "text")
+        assert(e2.kind == "text")
+        assert(e2.data == "streamed")
+
+        sess:close()
+        "#,
+        addr,
+        text_pipeline_lua()
+    );
+
+    lua.load(&script).exec_async().await.unwrap();
+    server.await.unwrap();
+}
+
+// ── session:off + pipeview.poll ───────────────────────────────────
+
+#[tokio::test]
+async fn lua_session_off_stops_future_callbacks() {
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr = listener.local_addr().unwrap().to_string();
+
+    let server = tokio::spawn(async move {
+        let (mut stream, _) = listener.accept().await.unwrap();
+        stream.write_all(b"first\n").await.unwrap();
+        tokio::time::sleep(Duration::from_millis(150)).await;
+        stream.write_all(b"second\n").await.unwrap();
+        tokio::time::sleep(Duration::from_millis(200)).await;
+    });
+
+    let lua = Lua::new();
+    pipeview_client::lua::register(&lua).unwrap();
+
+    let script = format!(
+        r#"
+        local received = {{}}
+        local sess = pipeview.open({{
+            transport = {{ Tcp = {{ addr = "{}" }} }},
+            pipelines = {{{}}}
+        }})
+
+        local token = sess:on_data(function(name, data)
+            table.insert(received, name .. ":" .. data)
+        end)
+
+        for _ = 1, 50 do
+            if #received == 0 then
+                pipeview.poll(1)
+                pipeview.sleep(10)
+            end
+        end
+
+        assert(#received == 1, "expected first callback before off, got " .. #received)
+        assert(received[1] == "text:first")
+        assert(sess:off(token) == true)
+
+        pipeview.sleep(250)
+        pipeview.poll(10)
+
+        assert(#received == 1, "callback should not fire after off")
+        assert(sess:off(token) == false)
+
+        sess:close()
+        "#,
+        addr,
+        text_pipeline_lua()
+    );
+
+    lua.load(&script).exec_async().await.unwrap();
+    server.await.unwrap();
+}

crates/pipeview-client/tests/session_lifecycle.rs

@@ -0,0 +1,334 @@
+use std::time::Duration;
+
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio::net::TcpListener;
+
+use pipeview_client::SessionManager;
+use pipeview_client::config::{DecoderConfig, FramerConfig, PipelineConfig, SessionConfig};
+use pipeview_client::session::{Session, SessionEvent};
+use pipeview_core::protocol::DecodedData;
+use pipeview_core::transport::TransportConfig;
+
+fn tcp_config(addr: String) -> SessionConfig {
+    SessionConfig {
+        transport: TransportConfig::Tcp { addr },
+        pipelines: vec![PipelineConfig {
+            name: "text".into(),
+            framer: FramerConfig::Line {
+                strip_cr: true,
+                max_line_len: 1024 * 1024,
+            },
+            decoder: DecoderConfig::Text {
+                encoding: pipeview_core::protocol::text::TextEncoding::Utf8,
+            },
+        }],
+        history_limit: 100,
+        auto_reconnect: false,
+    }
+}
+
+// ── Session spawn + basic send/read ──────────────────────────────
+
+#[tokio::test]
+async fn session_echo_send_and_read() {
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr = listener.local_addr().unwrap().to_string();
+
+    let server = tokio::spawn(async move {
+        let (mut stream, _) = listener.accept().await.unwrap();
+        let mut buf = [0u8; 256];
+        let n = stream.read(&mut buf).await.unwrap();
+        assert_eq!(&buf[..n], b"ping\n");
+        stream.write_all(b"pong\n").await.unwrap();
+    });
+
+    let handle = Session::spawn(1, tcp_config(addr));
+    handle.send(b"ping\n".to_vec()).await.unwrap();
+
+    let entry = handle.read(5000).await.expect("should receive response");
+    match entry.data {
+        DecodedData::Text(s) => assert_eq!(s, "pong"),
+        other => panic!("expected Text, got {:?}", other),
+    }
+
+    handle.close().await.unwrap();
+    server.await.unwrap();
+}
+
+#[tokio::test]
+async fn session_multiple_reads() {
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr = listener.local_addr().unwrap().to_string();
+
+    let server = tokio::spawn(async move {
+        let (mut stream, _) = listener.accept().await.unwrap();
+        stream.write_all(b"a\nb\nc\n").await.unwrap();
+    });
+
+    let handle = Session::spawn(2, tcp_config(addr));
+
+    // Subscribe once and collect all text events
+    let mut rx = handle.subscribe();
+    let mut texts = Vec::new();
+    tokio::time::timeout(Duration::from_secs(5), async {
+        loop {
+            if let SessionEvent::Data(_, entry) = rx.recv().await.unwrap()
+                && let DecodedData::Text(s) = entry.data
+            {
+                texts.push(s);
+                if texts.len() == 3 {
+                    break;
+                }
+            }
+        }
+    })
+    .await
+    .unwrap();
+
+    assert_eq!(texts, vec!["a", "b", "c"]);
+    handle.close().await.unwrap();
+    server.await.unwrap();
+}
+
+// ── Session with multiple pipelines ──────────────────────────────
+
+#[tokio::test]
+async fn session_multi_pipeline_text_and_hex() {
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr = listener.local_addr().unwrap().to_string();
+
+    let server = tokio::spawn(async move {
+        let (mut stream, _) = listener.accept().await.unwrap();
+        stream.write_all(b"hello\n").await.unwrap();
+    });
+
+    let mut config = tcp_config(addr);
+    config.pipelines.push(PipelineConfig {
+        name: "hex".into(),
+        framer: FramerConfig::Line {
+            strip_cr: true,
+            max_line_len: 1024 * 1024,
+        },
+        decoder: DecoderConfig::Hex {
+            uppercase: false,
+            separator: " ".into(),
+            bytes_per_group: 1,
+            endian: pipeview_core::protocol::Endian::Big,
+        },
+    });
+
+    let handle = Session::spawn(3, config);
+
+    // Subscribe and collect both pipeline results
+    let mut rx = handle.subscribe();
+    let mut results = Vec::new();
+    tokio::time::timeout(Duration::from_secs(5), async {
+        loop {
+            if let SessionEvent::Data(_, entry) = rx.recv().await.unwrap() {
+                results.push((entry.pipeline_name.clone(), entry.data.clone()));
+                if results.len() == 2 {
+                    break;
+                }
+            }
+        }
+    })
+    .await
+    .unwrap();
+
+    assert_eq!(results.len(), 2);
+    assert_eq!(results[0].0, "text");
+    assert_eq!(results[1].0, "hex");
+    assert!(matches!(results[1].1, DecodedData::Hex(_)));
+
+    handle.close().await.unwrap();
+    server.await.unwrap();
+}
+
+// ── SessionClose + read timeout ──────────────────────────────────
+
+#[tokio::test]
+async fn session_read_timeout_returns_none() {
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr = listener.local_addr().unwrap().to_string();
+
+    let server = tokio::spawn(async move {
+        let (_stream, _) = listener.accept().await.unwrap();
+        // Accept but send nothing — client should time out
+        tokio::time::sleep(Duration::from_secs(10)).await;
+    });
+
+    let handle = Session::spawn(4, tcp_config(addr));
+
+    // Short timeout — no data sent by server
+    let result = handle.read(200).await;
+    assert!(result.is_none(), "read should time out and return None");
+
+    handle.close().await.unwrap();
+    server.abort();
+}
+
+#[tokio::test]
+async fn session_close_stops_events() {
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr = listener.local_addr().unwrap().to_string();
+
+    let _server = tokio::spawn(async move {
+        let (_stream, _) = listener.accept().await.unwrap();
+        tokio::time::sleep(Duration::from_secs(10)).await;
+    });
+
+    let handle = Session::spawn(5, tcp_config(addr));
+    handle.close().await.unwrap();
+
+    // After close, read should return None immediately
+    let result = handle.read(100).await;
+    assert!(result.is_none());
+}
+
+// ── SessionManager ───────────────────────────────────────────────
+
+#[tokio::test]
+async fn manager_create_and_list() {
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr = listener.local_addr().unwrap().to_string();
+
+    let _server = tokio::spawn(async move {
+        let (_stream, _) = listener.accept().await.unwrap();
+        tokio::time::sleep(Duration::from_secs(10)).await;
+    });
+
+    let mgr = SessionManager::new();
+    assert_eq!(mgr.count(), 0);
+    assert!(mgr.list_ids().is_empty());
+
+    let h1 = mgr.create(tcp_config(addr.clone()));
+    assert_eq!(mgr.count(), 1);
+    assert_eq!(mgr.list_ids(), vec![1]);
+
+    let h2 = mgr.create(tcp_config(addr));
+    assert_eq!(mgr.count(), 2);
+
+    assert!(mgr.get(1).is_some());
+    assert!(mgr.get(2).is_some());
+    assert!(mgr.get(99).is_none());
+
+    mgr.remove(1);
+    assert_eq!(mgr.count(), 1);
+    assert!(mgr.get(1).is_none());
+
+    h1.close().await.unwrap();
+    h2.close().await.unwrap();
+}
+
+#[tokio::test]
+async fn manager_event_subscription() {
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr = listener.local_addr().unwrap().to_string();
+
+    let server = tokio::spawn(async move {
+        let (mut stream, _) = listener.accept().await.unwrap();
+        stream.write_all(b"data\n").await.unwrap();
+    });
+
+    let mgr = SessionManager::new();
+    let mut rx = mgr.subscribe();
+
+    let handle = mgr.create(tcp_config(addr));
+
+    // Wait for Connected event
+    let event = tokio::time::timeout(Duration::from_secs(5), rx.recv())
+        .await
+        .unwrap()
+        .unwrap();
+    assert!(matches!(event, SessionEvent::Connected(1)));
+
+    // Wait for Data event
+    let event = tokio::time::timeout(Duration::from_secs(5), rx.recv())
+        .await
+        .unwrap()
+        .unwrap();
+    assert!(matches!(event, SessionEvent::Data(1, _)));
+
+    handle.close().await.unwrap();
+    server.await.unwrap();
+}
+
+// ── Reconfigure ──────────────────────────────────────────────────
+
+#[tokio::test]
+async fn session_reconfigure_changes_pipeline() {
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr = listener.local_addr().unwrap().to_string();
+
+    let server = tokio::spawn(async move {
+        let (mut stream, _) = listener.accept().await.unwrap();
+        stream.write_all(b"old\n").await.unwrap();
+    });
+
+    let handle = Session::spawn(6, tcp_config(addr));
+    let entry = handle.read(5000).await.unwrap();
+    assert!(matches!(entry.data, DecodedData::Text(s) if s == "old"));
+    server.await.unwrap();
+
+    // Reconfigure to hex pipeline on a new address (don't close first)
+    let listener2 = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr2 = listener2.local_addr().unwrap().to_string();
+
+    let server2 = tokio::spawn(async move {
+        let (mut stream, _) = listener2.accept().await.unwrap();
+        stream.write_all(b"ABC\n").await.unwrap();
+    });
+
+    let mut new_cfg = tcp_config(addr2);
+    new_cfg.pipelines[0].name = "hex".into();
+    new_cfg.pipelines[0].decoder = DecoderConfig::Hex {
+        uppercase: true,
+        separator: " ".into(),
+        bytes_per_group: 1,
+        endian: pipeview_core::protocol::Endian::Big,
+    };
+
+    handle.reconfigure(new_cfg).await.unwrap();
+    let entry = handle.read(5000).await.unwrap();
+    assert!(matches!(entry.data, DecodedData::Hex(s) if s == "41 42 43"));
+
+    handle.close().await.unwrap();
+    server2.await.unwrap();
+}
+
+// ── SessionHandle event subscription ─────────────────────────────
+
+#[tokio::test]
+async fn session_handle_subscribe_sees_events() {
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr = listener.local_addr().unwrap().to_string();
+
+    let server = tokio::spawn(async move {
+        let (mut stream, _) = listener.accept().await.unwrap();
+        stream.write_all(b"event_test\n").await.unwrap();
+    });
+
+    let handle = Session::spawn(7, tcp_config(addr));
+    let mut rx = handle.subscribe();
+
+    // Should see Connected then Data
+    let e1 = tokio::time::timeout(Duration::from_secs(5), rx.recv())
+        .await
+        .unwrap()
+        .unwrap();
+    assert!(matches!(e1, SessionEvent::Connected(7)));
+
+    let e2 = tokio::time::timeout(Duration::from_secs(5), rx.recv())
+        .await
+        .unwrap()
+        .unwrap();
+    match e2 {
+        SessionEvent::Data(7, entry) => {
+            assert_eq!(entry.pipeline_name, "text");
+        }
+        other => panic!("expected Data, got {:?}", other),
+    }
+
+    handle.close().await.unwrap();
+    server.await.unwrap();
+}