feat: implement core capture engine with CLI and output protocols

9 files changed, 834 insertions, 52 deletions

diff --git a/src/args.rs b/src/args.rs
index fcf224c..52dddfc 100644
--- a/src/args.rs
+++ b/src/args.rs
@@ -1,12 +1,51 @@
 use std::path::PathBuf;
 
+/// Whether verbose debug output is enabled.
+/// Set by `DMVOPArguments::verbose`.
+pub static VERBOSE: std::sync::atomic::AtomicBool = std::sync::atomic::AtomicBool::new(false);
+
+/// Print a debug message only when `--verbose` is set.
+#[macro_export]
+macro_rules! debug_log {
+    ($($arg:tt)*) => {
+        if $crate::args::VERBOSE.load(std::sync::atomic::Ordering::Relaxed) {
+            eprintln!($($arg)*);
+        }
+    };
+}
+
 #[derive(clap::Parser)]
-#[command(
-    no_binary_name = true,
-    disable_help_flag = true,
-    disable_version_flag = true
-)]
+#[command(name = "dmvop", disable_help_flag = true, disable_version_flag = true)]
 pub struct DMVOPArguments {
+    // Verbose output (show debug messages)
+    #[arg(long = "verbose", short = 'V')]
+    pub verbose: bool,
+
+    // List all available input devices and exit
+    #[arg(long = "list-devices", alias = "list", short = 'L')]
+    pub list_devices: bool,
+
+    // List all available Whisper models and exit
+    #[arg(long = "list-models")]
+    pub list_models: bool,
+
+    // Download a specific Whisper model and exit
+    #[arg(long = "download-model", alias = "get-model", require_equals = true)]
+    pub download_model: Option<String>,
+
+    // Language hint for Whisper (e.g. en, zh, ja). Auto-detects if not set.
+    #[arg(long = "lang", require_equals = true)]
+    pub lang: Option<String>,
+
+    // Whisper model to use (e.g. tiny, base, small, medium, large-v3)
+    #[arg(
+        long = "model",
+        short = 'm',
+        default_value = "base_en",
+        require_equals = true
+    )]
+    pub model: String,
+
     // Devices (unix/linux device or WASAPI name)
     #[arg(
         long = "device",
@@ -14,7 +53,7 @@ pub struct DMVOPArguments {
         allow_hyphen_values = true,
         require_equals = true
     )]
-    device_name: String,
+    pub device_name: Option<String>,
 
     // Format
     // Use %{param} to represent a parameter
@@ -30,24 +69,25 @@ pub struct DMVOPArguments {
         default_value = "%{vol},%{word}",
         require_equals = true
     )]
-    format_pattern: Option<String>,
+    pub format_pattern: Option<String>,
+
+    #[arg(long = "format-file", short = 'S', require_equals = true)]
+    pub format_file: Option<PathBuf>,
 
+    // Output (can be specified multiple times)
     #[arg(
-        long = "format-file",
-        short = 'S',
-        alias = "fmt",
-        require_equals = true
+        long,
+        short = 'O',
+        require_equals = true,
+        default_value = "stdout",
+        num_args = 1
     )]
-    format_file: Option<PathBuf>,
-
-    // Output
-    #[arg(long, short = 'O', require_equals = true, default_value = "stdout")]
-    output: OutputMode,
+    pub output: Vec<OutputMode>,
 
     // MISC
     // Port (default: 5117)
     #[arg(long, short = 'p', default_value_t = 5117, require_equals = true)]
-    port: u16,
+    pub port: u16,
 
     // Socket file (default: ./dmvop.sock in current directory)
     #[arg(
@@ -56,7 +96,7 @@ pub struct DMVOPArguments {
         default_value = "./dmvop.sock",
         require_equals = true
     )]
-    socket_file: PathBuf,
+    pub socket_file: PathBuf,
 
     // Subnet mask for UDP broadcast (default: only last octet, e.g., "255.255.255.0")
     #[arg(
@@ -65,10 +105,10 @@ pub struct DMVOPArguments {
         default_value = "255.255.255.0",
         require_equals = true
     )]
-    subnet_mask: String,
+    pub subnet_mask: String,
 }
 
-#[derive(Clone)]
+#[derive(Clone, Debug)]
 #[allow(non_camel_case_types)]
 pub enum OutputMode {
     /// Establish a basic TCP service, broadcasting output to all connected sockets
@@ -103,3 +143,43 @@ impl std::str::FromStr for OutputMode {
         }
     }
 }
+
+/// Parse a format pattern string and produce a formatted output string
+/// from the provided values.
+///
+/// Supported placeholders:
+/// - `%{vol}` — volume 0–100
+/// - `%{word}` — transcribed word/text
+/// - `%{confid}` / `%{confidence}` — confidence score
+pub fn format_output(pattern: &str, word: &str, confidence: f32, volume: f32) -> String {
+    let mut result = pattern.to_string();
+
+    // Volume: clamp dB range (~-60 to 0) to 0–100 scale
+    // Typical speech is around -30 dB to -12 dB
+    let vol_normalized = ((volume + 60.0).clamp(0.0, 60.0) / 60.0 * 100.0) as u32;
+
+    // Replace known placeholders
+    result = result.replace("%{vol}", &vol_normalized.to_string());
+    result = result.replace("%{word}", word);
+    result = result.replace("%{confid}", &format!("{:.1}", confidence));
+    result = result.replace("%{confidence}", &format!("{:.1}", confidence));
+
+    result
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_format_output() {
+        let s = format_output("%{vol},%{word},%{confid}", "hello", 95.0, -12.0);
+        assert_eq!(s, "80,hello,95.0");
+    }
+
+    #[test]
+    fn test_format_output_no_volume() {
+        let s = format_output("text: %{word}", "world", 0.0, -60.0);
+        assert_eq!(s, "text: world");
+    }
+}
diff --git a/src/main.rs b/src/main.rs
index cc8d946..5b2bb0a 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -2,8 +2,430 @@ mod args;
 pub use args::*;
 
 mod output_protocol;
-pub use output_protocol::*;
+use clap::Parser;
+use output_protocol::OutputProtocol;
+use std::path::PathBuf;
+use std::sync::Arc;
+use vtx_engine::EngineBuilder;
 
-fn main() {
-    println!("Hello, world!");
+/// Output channel enum — wraps each OutputProtocol implementation so we can
+/// store a heterogeneous collection and dispatch `send` without trait objects.
+enum OutputChannel {
+    Stdout(Arc<output_protocol::StandardOutputProtocol>),
+    Stderr(Arc<output_protocol::StandardErrorProtocol>),
+    Tcp(Arc<output_protocol::TCPOutputProtocol>),
+    Udp(Arc<output_protocol::UDPOutputProtocol>),
+    UdpBroadcast(Arc<output_protocol::UDPBroadcastOutputProtocol>),
+    #[cfg(unix)]
+    Ipc(Arc<output_protocol::IPCOutputProtocol>),
+}
+
+impl OutputChannel {
+    async fn init(&self) {
+        match self {
+            OutputChannel::Stdout(p) => p.init().await,
+            OutputChannel::Stderr(p) => p.init().await,
+            OutputChannel::Tcp(p) => p.init().await,
+            OutputChannel::Udp(p) => p.init().await,
+            OutputChannel::UdpBroadcast(p) => p.init().await,
+            #[cfg(unix)]
+            OutputChannel::Ipc(p) => p.init().await,
+        }
+    }
+
+    async fn send(&self, message: &str) {
+        match self {
+            OutputChannel::Stdout(p) => p.clone().send(message).await,
+            OutputChannel::Stderr(p) => p.clone().send(message).await,
+            OutputChannel::Tcp(p) => p.clone().send(message).await,
+            OutputChannel::Udp(p) => p.clone().send(message).await,
+            OutputChannel::UdpBroadcast(p) => p.clone().send(message).await,
+            #[cfg(unix)]
+            OutputChannel::Ipc(p) => p.clone().send(message).await,
+        }
+    }
+}
+
+#[tokio::main]
+async fn main() {
+    // Set up tracing so we can see vtx-engine logs (including transcription errors)
+    tracing_subscriber::fmt()
+        .with_env_filter(
+            tracing_subscriber::EnvFilter::builder()
+                .with_default_directive(tracing_subscriber::filter::LevelFilter::WARN.into())
+                .from_env_lossy(),
+        )
+        .with_target(false)
+        .init();
+
+    let args = DMVOPArguments::parse();
+
+    // Set global verbose flag
+    VERBOSE.store(args.verbose, std::sync::atomic::Ordering::Relaxed);
+
+    // ---------------------------------------------------------------
+    // --list-models: show available models and exit
+    // ---------------------------------------------------------------
+    if args.list_models {
+        list_models();
+        return;
+    }
+
+    // ---------------------------------------------------------------
+    // --download-model: download a specific model and exit
+    // ---------------------------------------------------------------
+    if let Some(ref model_name) = args.download_model {
+        download_model_cli(model_name).await;
+        return;
+    }
+
+    // ---------------------------------------------------------------
+    // 1. Build the vtx-engine (needed for both listing and capture)
+    // ---------------------------------------------------------------
+    eprintln!("[dmvop] Initializing voice engine...");
+
+    let model = vtx_engine::WhisperModel::parse_identifier(&args.model).unwrap_or_else(|| {
+        eprintln!(
+            "[dmvop] Unknown model '{}'. Use --list-models to see available models.",
+            args.model
+        );
+        std::process::exit(1);
+    });
+
+    eprintln!(
+        "[dmvop] Using model: {} ({})",
+        model.config_key(),
+        model.display_name()
+    );
+
+    let mut builder = EngineBuilder::new().app_name("dmvop").model(model);
+
+    if let Some(ref lang) = args.lang {
+        eprintln!("[dmvop] Language hint: {}", lang);
+        builder = builder.language(lang.as_str());
+    }
+
+    let (engine, mut rx) = builder.build().await.expect("Failed to build vtx-engine");
+
+    // Disable PTT mode so VAD drives automatic segmentation
+    engine.set_ptt_mode(false);
+
+    // ---------------------------------------------------------------
+    // Check model availability and download if needed
+    // ---------------------------------------------------------------
+    let model_status = engine.check_model_status();
+    if !model_status.available {
+        eprintln!("[dmvop] Model not found at: {}", model_status.path);
+        eprintln!("[dmvop] Downloading model, please wait...");
+        match engine.download_model().await {
+            Ok(_) => eprintln!("[dmvop] Model downloaded successfully"),
+            Err(e) => {
+                eprintln!("[dmvop] Failed to download model: {}", e);
+                eprintln!("[dmvop] You can manually download a model from:");
+                eprintln!("[dmvop]   https://huggingface.co/ggerganov/whisper.cpp/tree/main");
+                eprintln!("[dmvop]   Place it at: {}", model_status.path);
+                std::process::exit(1);
+            }
+        }
+    } else {
+        eprintln!("[dmvop] Model found: {}", model_status.path);
+    }
+
+    // ---------------------------------------------------------------
+    // 2. List devices and exit?
+    // ---------------------------------------------------------------
+    let devices = engine.list_input_devices();
+
+    if args.list_devices {
+        if devices.is_empty() {
+            eprintln!("[dmvop] No input devices found.");
+        } else {
+            println!("Available input devices:");
+            for (i, dev) in devices.iter().enumerate() {
+                println!(
+                    "  [{}] {} (id: {}, type: {:?})",
+                    i, dev.name, dev.id, dev.source_type
+                );
+            }
+        }
+        return;
+    }
+
+    // ---------------------------------------------------------------
+    // 3. Resolve the format pattern
+    // ---------------------------------------------------------------
+    let pattern = resolve_format_pattern(args.format_pattern.as_deref(), args.format_file.as_ref());
+
+    // ---------------------------------------------------------------
+    // 4. Create and initialize output channels
+    // ---------------------------------------------------------------
+    let mut channels: Vec<OutputChannel> = Vec::new();
+
+    for mode in &args.output {
+        match create_output_channel(mode, args.port, args.socket_file.clone(), &args.subnet_mask) {
+            Some(ch) => {
+                ch.init().await;
+                channels.push(ch);
+            }
+            None => debug_log!(
+                "[dmvop] Warning: failed to create output channel {:?}",
+                mode
+            ),
+        }
+    }
+
+    if channels.is_empty() {
+        eprintln!("[dmvop] No output channels available. Exiting.");
+        std::process::exit(1);
+    }
+
+    // ---------------------------------------------------------------
+    // 5. Find the requested device and start capture
+    // ---------------------------------------------------------------
+    let device_name = match &args.device_name {
+        Some(n) => n.as_str(),
+        None => {
+            eprintln!(
+                "[dmvop] No device specified. Use --device=<name> or --list-devices to see available devices."
+            );
+            std::process::exit(1);
+        }
+    };
+
+    let device = devices
+        .iter()
+        .find(|d| d.id == device_name || d.name == device_name)
+        .or_else(|| devices.first());
+
+    match &device {
+        Some(d) => {
+            eprintln!("[dmvop] Using input device: {} (id: {})", d.name, d.id);
+        }
+        None => {
+            eprintln!(
+                "[dmvop] Device '{}' not found and no fallback available.",
+                device_name
+            );
+            std::process::exit(1);
+        }
+    }
+
+    engine
+        .start_capture(device.map(|d| d.id.clone()), None)
+        .await
+        .expect("Failed to start audio capture");
+
+    eprintln!("[dmvop] Capture started. Waiting for speech...");
+
+    // ---------------------------------------------------------------
+    // 5. Event loop — listen for transcription & audio level events
+    // ---------------------------------------------------------------
+    let mut last_volume_db: f32 = -60.0;
+
+    loop {
+        match rx.recv().await {
+            Ok(event) => match event {
+                vtx_engine::EngineEvent::TranscriptionComplete(result) => {
+                    let formatted = format_output(&pattern, &result.text, 0.0, last_volume_db);
+
+                    for ch in &channels {
+                        ch.send(&formatted).await;
+                    }
+                }
+                vtx_engine::EngineEvent::TranscriptionSegment(segment) => {
+                    let formatted = format_output(&pattern, &segment.text, 0.0, last_volume_db);
+
+                    for ch in &channels {
+                        ch.send(&formatted).await;
+                    }
+                }
+                vtx_engine::EngineEvent::VisualizationData(viz) => {
+                    if let Some(ref metrics) = viz.speech_metrics {
+                        last_volume_db = metrics.amplitude_db;
+                    }
+                }
+                vtx_engine::EngineEvent::SpeechStarted => {
+                    debug_log!("[dmvop] Speech started");
+                }
+                vtx_engine::EngineEvent::SpeechEnded { duration_ms } => {
+                    debug_log!("[dmvop] Speech ended ({}ms)", duration_ms);
+                }
+                vtx_engine::EngineEvent::CaptureStateChanged { capturing, error } => {
+                    if !capturing {
+                        eprintln!(
+                            "[dmvop] Capture stopped: {}",
+                            error.unwrap_or_else(|| "unknown".to_string())
+                        );
+                        break;
+                    }
+                }
+                vtx_engine::EngineEvent::ModelDownloadProgress { percent } => {
+                    debug_log!("[dmvop] Downloading model: {}%", percent);
+                }
+                vtx_engine::EngineEvent::ModelDownloadComplete { success } => {
+                    debug_log!(
+                        "[dmvop] Model download {}",
+                        if success { "complete" } else { "failed" }
+                    );
+                }
+                _ => {}
+            },
+            Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => {
+                debug_log!("[dmvop] Warning: missed {} events", n);
+            }
+            Err(tokio::sync::broadcast::error::RecvError::Closed) => {
+                eprintln!("[dmvop] Engine event stream closed");
+                break;
+            }
+        }
+    }
+
+    eprintln!("[dmvop] Shutting down.");
+}
+
+/// Resolve the format pattern from either the command-line `--format` value
+/// or the contents of `--format-file`.
+fn resolve_format_pattern(pattern: Option<&str>, file: Option<&PathBuf>) -> String {
+    if let Some(path) = file {
+        match std::fs::read_to_string(path) {
+            Ok(content) => {
+                let trimmed = content.trim().to_string();
+                if !trimmed.is_empty() {
+                    return trimmed;
+                }
+                eprintln!(
+                    "[dmvop] Warning: format file {} is empty, using default pattern",
+                    path.display()
+                );
+            }
+            Err(e) => {
+                eprintln!(
+                    "[dmvop] Warning: could not read format file {}: {}",
+                    path.display(),
+                    e
+                );
+            }
+        }
+    }
+
+    pattern
+        .filter(|s| !s.is_empty())
+        .map(|s| s.to_string())
+        .unwrap_or_else(|| "%{vol},%{word}".to_string())
+}
+
+/// Create an output channel from an [`OutputMode`].
+fn create_output_channel(
+    mode: &OutputMode,
+    port: u16,
+    socket_file: PathBuf,
+    subnet_mask: &str,
+) -> Option<OutputChannel> {
+    match mode {
+        OutputMode::STDOUT => Some(OutputChannel::Stdout(Arc::new(
+            output_protocol::StandardOutputProtocol,
+        ))),
+        OutputMode::STDERR => Some(OutputChannel::Stderr(Arc::new(
+            output_protocol::StandardErrorProtocol,
+        ))),
+        OutputMode::TCP => Some(OutputChannel::Tcp(Arc::new(
+            output_protocol::TCPOutputProtocol::new(port),
+        ))),
+        OutputMode::UDP => Some(OutputChannel::Udp(Arc::new(
+            output_protocol::UDPOutputProtocol::new(port),
+        ))),
+        OutputMode::UDP_BROADCAST => Some(OutputChannel::UdpBroadcast(Arc::new(
+            output_protocol::UDPBroadcastOutputProtocol::new(port, subnet_mask),
+        ))),
+        OutputMode::IPC => {
+            #[cfg(unix)]
+            {
+                Some(OutputChannel::Ipc(Arc::new(
+                    output_protocol::IPCOutputProtocol::new(socket_file),
+                )))
+            }
+            #[cfg(not(unix))]
+            {
+                let _ = socket_file;
+                eprintln!("[dmvop] IPC (Unix domain socket) is not supported on this platform");
+                None
+            }
+        }
+    }
+}
+
+/// Print all available Whisper models and their sizes.
+fn list_models() {
+    println!("Available Whisper models:");
+    for model in vtx_engine::WhisperModel::all_in_size_order() {
+        let size = model.size_mb();
+        let size_str = if size >= 1024 {
+            format!("{:.1} GB", size as f64 / 1024.0)
+        } else {
+            format!("{} MB", size)
+        };
+        println!(
+            "  {:20}  {}  ({})",
+            model.config_key(),
+            size_str,
+            model.display_name()
+        );
+    }
+}
+
+/// Download a specific Whisper model by identifier.
+async fn download_model_cli(model_name: &str) {
+    let model = match vtx_engine::WhisperModel::parse_identifier(model_name) {
+        Some(m) => m,
+        None => {
+            eprintln!(
+                "[dmvop] Unknown model '{}'. Use --list-models to see available models.",
+                model_name
+            );
+            std::process::exit(1);
+        }
+    };
+
+    eprintln!(
+        "[dmvop] Building engine with model '{}'...",
+        model.config_key()
+    );
+
+    let (engine, _rx) = match EngineBuilder::new()
+        .app_name("dmvop")
+        .model(model)
+        .build()
+        .await
+    {
+        Ok(e) => e,
+        Err(e) => {
+            eprintln!("[dmvop] Failed to build engine: {}", e);
+            std::process::exit(1);
+        }
+    };
+
+    let status = engine.check_model_status();
+    if status.available {
+        eprintln!("[dmvop] Model already exists at: {}", status.path);
+        return;
+    }
+
+    eprintln!(
+        "[dmvop] Downloading {} ({} MB)...",
+        model.config_key(),
+        model.size_mb()
+    );
+
+    match engine.download_model().await {
+        Ok(_) => {
+            eprintln!("[dmvop] Model downloaded to: {}", status.path);
+        }
+        Err(e) => {
+            eprintln!("[dmvop] Failed to download model: {}", e);
+            eprintln!("[dmvop] You can manually download from:");
+            eprintln!("[dmvop]   {}", model.download_url());
+            eprintln!("[dmvop]   Place it at: {}", status.path);
+            std::process::exit(1);
+        }
+    }
 }
diff --git a/src/output_protocol.rs b/src/output_protocol.rs
index 87e2400..6a25d2c 100644
--- a/src/output_protocol.rs
+++ b/src/output_protocol.rs
@@ -1,5 +1,6 @@
 use std::sync::Arc;
 
+#[cfg(unix)]
 pub mod ipc;
 pub mod stderr;
 pub mod stdout;
@@ -7,6 +8,7 @@ pub mod tcp;
 pub mod udp;
 pub mod udp_broadcast;
 
+#[cfg(unix)]
 pub use ipc::*;
 pub use stderr::*;
 pub use stdout::*;
@@ -26,13 +28,12 @@ pub use udp_broadcast::*;
 pub trait OutputProtocol {
     /// Initializes the output channel.
     ///
-    /// This method consumes `self`, indicating it should be called once during setup
-    /// to prepare the output resource (e.g., opening a file, establishing a connection).
+    /// Prepares the output resource (e.g., opening a file, establishing a connection).
     ///
     /// # Returns
     ///
     /// A future that resolves when initialization is complete.
-    fn init(self) -> impl Future<Output = ()> + Send + Sync;
+    fn init(&self) -> impl Future<Output = ()> + Send + Sync;
 
     /// Sends a message string through the output channel.
     ///
diff --git a/src/output_protocol/ipc.rs b/src/output_protocol/ipc.rs
index c9c88c1..fb6df80 100644
--- a/src/output_protocol/ipc.rs
+++ b/src/output_protocol/ipc.rs
@@ -1,14 +1,70 @@
 use crate::OutputProtocol;
+use crate::debug_log;
+use std::path::PathBuf;
+use std::sync::Arc;
+use tokio::io::AsyncWriteExt;
+use tokio::net::UnixStream;
 
-#[derive(Debug, Default)]
-pub struct IPCOutputProtocol {}
+/// IPC output protocol.
+///
+/// Connects to a Unix domain socket and sends messages to it.
+/// On Windows, this will fail with a clear error message since
+/// Unix domain sockets are not supported.
+#[derive(Debug)]
+pub struct IPCOutputProtocol {
+    socket_path: PathBuf,
+    stream: tokio::sync::Mutex<Option<UnixStream>>,
+}
+
+impl IPCOutputProtocol {
+    pub fn new(socket_path: PathBuf) -> Self {
+        Self {
+            socket_path,
+            stream: tokio::sync::Mutex::new(None),
+        }
+    }
+}
 
 impl OutputProtocol for IPCOutputProtocol {
-    async fn init(self) {
-        todo!()
+    async fn init(&self) {
+        let path = self.socket_path.clone();
+        let stream_lock = &self.stream;
+
+        match UnixStream::connect(&path).await {
+            Ok(stream) => {
+                debug_log!("[IPC] Connected to {}", path.display());
+                *stream_lock.lock().await = Some(stream);
+            }
+            Err(e) => {
+                debug_log!("[IPC] Failed to connect to {}: {}", path.display(), e);
+            }
+        }
     }
 
-    async fn send(self: std::sync::Arc<Self>, _str: &str) {
-        todo!()
+    async fn send(self: Arc<Self>, message: &str) {
+        let mut guard = self.stream.lock().await;
+        if let Some(ref mut stream) = *guard {
+            let bytes = format!("{}\n", message);
+            if stream.write_all(bytes.as_bytes()).await.is_err() {
+                debug_log!("[IPC] Write error, reconnecting...");
+                *guard = None;
+                // Try to reconnect
+                if let Ok(new_stream) = UnixStream::connect(&self.socket_path).await {
+                    debug_log!("[IPC] Reconnected to {}", self.socket_path.display());
+                    *guard = Some(new_stream);
+                }
+            }
+        } else {
+            // Try to connect
+            if let Ok(stream) = UnixStream::connect(&self.socket_path).await {
+                debug_log!("[IPC] Connected to {}", self.socket_path.display());
+                let bytes = format!("{}\n", message);
+                let _ = stream.writable().await;
+                // We can't use the stream directly here since we need to store it
+                // for future sends. Let's store and send.
+                let _ = stream.try_write(bytes.as_bytes());
+                *guard = Some(stream);
+            }
+        }
     }
 }
diff --git a/src/output_protocol/stderr.rs b/src/output_protocol/stderr.rs
index fabd390..68e94a7 100644
--- a/src/output_protocol/stderr.rs
+++ b/src/output_protocol/stderr.rs
@@ -4,7 +4,7 @@ use crate::OutputProtocol;
 pub struct StandardErrorProtocol;
 
 impl OutputProtocol for StandardErrorProtocol {
-    async fn init(self) {
+    async fn init(&self) {
         // No initialization needed
     }
 
diff --git a/src/output_protocol/stdout.rs b/src/output_protocol/stdout.rs
index 3640507..3458da8 100644
--- a/src/output_protocol/stdout.rs
+++ b/src/output_protocol/stdout.rs
@@ -4,7 +4,7 @@ use crate::OutputProtocol;
 pub struct StandardOutputProtocol;
 
 impl OutputProtocol for StandardOutputProtocol {
-    async fn init(self) {
+    async fn init(&self) {
         // No initialization needed
     }
 
diff --git a/src/output_protocol/tcp.rs b/src/output_protocol/tcp.rs
index c2490e6..f8e7787 100644
--- a/src/output_protocol/tcp.rs
+++ b/src/output_protocol/tcp.rs
@@ -1,14 +1,74 @@
 use crate::OutputProtocol;
+use crate::debug_log;
+use std::sync::Arc;
+use tokio::io::AsyncWriteExt;
+use tokio::net::TcpListener;
+use tokio::sync::Mutex;
 
-#[derive(Debug, Default)]
-pub struct TCPOutputProtocol {}
+#[derive(Debug)]
+pub struct TCPOutputProtocol {
+    port: u16,
+    clients: Arc<Mutex<Vec<tokio::net::tcp::OwnedWriteHalf>>>,
+}
+
+impl TCPOutputProtocol {
+    pub fn new(port: u16) -> Self {
+        Self {
+            port,
+            clients: Arc::new(Mutex::new(Vec::new())),
+        }
+    }
+}
 
 impl OutputProtocol for TCPOutputProtocol {
-    async fn init(self) {
-        todo!()
+    async fn init(&self) {
+        let port = self.port;
+        let clients = self.clients.clone();
+
+        tokio::spawn(async move {
+            let addr = format!("0.0.0.0:{}", port);
+            let listener = match TcpListener::bind(&addr).await {
+                Ok(l) => l,
+                Err(e) => {
+                    eprintln!("[TCP] Failed to bind to {}: {}", addr, e);
+                    return;
+                }
+            };
+
+            debug_log!("[TCP] Listening on {}", addr);
+
+            loop {
+                match listener.accept().await {
+                    Ok((stream, peer)) => {
+                        debug_log!("[TCP] Client connected: {}", peer);
+                        let (_, write_half) = stream.into_split();
+                        clients.lock().await.push(write_half);
+                    }
+                    Err(e) => {
+                        debug_log!("[TCP] Accept error: {}", e);
+                    }
+                }
+            }
+        });
     }
 
-    async fn send(self: std::sync::Arc<Self>, _str: &str) {
-        todo!()
+    async fn send(self: Arc<Self>, message: &str) {
+        let mut clients = self.clients.lock().await;
+        let mut i = 0;
+        while i < clients.len() {
+            let mut write_half = clients.remove(i);
+            let bytes = format!("{}\n", message);
+            match write_half.write_all(bytes.as_bytes()).await {
+                Ok(_) => {
+                    // Re-insert at the end if successful
+                    clients.insert(i, write_half);
+                    i += 1;
+                }
+                Err(_) => {
+                    // Client disconnected, drop it
+                    debug_log!("[TCP] Client disconnected, removing");
+                }
+            }
+        }
     }
 }
diff --git a/src/output_protocol/udp.rs b/src/output_protocol/udp.rs
index f1e5876..8c6c9b9 100644
--- a/src/output_protocol/udp.rs
+++ b/src/output_protocol/udp.rs
@@ -1,14 +1,83 @@
+use crate::debug_log;
 use crate::OutputProtocol;
+use std::net::SocketAddr;
+use std::sync::Arc;
+use tokio::net::UdpSocket;
+use tokio::sync::Mutex;
 
-#[derive(Debug, Default)]
-pub struct UDPOutputProtocol {}
+/// UDP output protocol.
+///
+/// Binds to `0.0.0.0:{port}`. The first incoming datagram determines the
+/// target address; all subsequent `send` calls deliver messages to that peer.
+#[derive(Debug)]
+pub struct UDPOutputProtocol {
+    inner: Arc<UDPInner>,
+}
+
+#[derive(Debug)]
+struct UDPInner {
+    socket: Mutex<Option<Arc<UdpSocket>>>,
+    target: Mutex<Option<SocketAddr>>,
+    target_learned: tokio::sync::watch::Sender<bool>,
+}
+
+impl UDPOutputProtocol {
+    pub fn new(port: u16) -> Self {
+        let (tx, _rx) = tokio::sync::watch::channel(false);
+        let inner = Arc::new(UDPInner {
+            socket: Mutex::new(None),
+            target: Mutex::new(None),
+            target_learned: tx,
+        });
+
+        let inner_clone = inner.clone();
+        tokio::spawn(async move {
+            let addr = format!("0.0.0.0:{}", port);
+            let socket = match UdpSocket::bind(&addr).await {
+                Ok(s) => {
+                    debug_log!("[UDP] Listening on {}", addr);
+                    Arc::new(s)
+                }
+                Err(e) => {
+                    debug_log!("[UDP] Failed to bind: {}", e);
+                    return;
+                }
+            };
+
+            // Store the socket
+            *inner_clone.socket.lock().await = Some(socket.clone());
+
+            // Learn target from first incoming packet
+            let mut buf = [0u8; 4096];
+            let (len, src) = match socket.recv_from(&mut buf).await {
+                Ok(r) => r,
+                Err(e) => {
+                    debug_log!("[UDP] Recv error: {}", e);
+                    return;
+                }
+            };
+            debug_log!("[UDP] Learned target address: {} (got {} bytes)", src, len);
+            *inner_clone.target.lock().await = Some(src);
+            let _ = inner_clone.target_learned.send(true);
+        });
+
+        Self { inner }
+    }
+}
 
 impl OutputProtocol for UDPOutputProtocol {
-    async fn init(self) {
-        todo!()
+    async fn init(&self) {
+        // Everything is set up in new() — nothing more to do.
     }
 
-    async fn send(self: std::sync::Arc<Self>, _str: &str) {
-        todo!()
+    async fn send(self: Arc<Self>, message: &str) {
+        let socket_opt = self.inner.socket.lock().await;
+        if let Some(ref socket) = *socket_opt {
+            let target_opt = self.inner.target.lock().await;
+            if let Some(target) = *target_opt {
+                let bytes = format!("{}\n", message);
+                let _ = socket.send_to(bytes.as_bytes(), target).await;
+            }
+        }
     }
 }
diff --git a/src/output_protocol/udp_broadcast.rs b/src/output_protocol/udp_broadcast.rs
index 5a5aa57..dae77fe 100644
--- a/src/output_protocol/udp_broadcast.rs
+++ b/src/output_protocol/udp_broadcast.rs
@@ -1,14 +1,108 @@
+use crate::debug_log;
 use crate::OutputProtocol;
 
-#[derive(Debug, Default)]
-pub struct UDPBroadcastOutputProtocol {}
+use std::sync::Arc;
+use tokio::net::UdpSocket;
+use tokio::sync::Mutex;
+
+/// UDP broadcast output protocol.
+///
+/// Broadcasts messages to the subnet defined by `port` and `subnet_mask`.
+/// The broadcast address is computed as `device_ip | (~subnet_mask)`.
+#[derive(Debug)]
+pub struct UDPBroadcastOutputProtocol {
+    inner: Arc<UDPBroadcastInner>,
+}
+
+#[derive(Debug)]
+struct UDPBroadcastInner {
+    socket: Mutex<Option<Arc<UdpSocket>>>,
+    broadcast_addr: Mutex<Option<String>>,
+}
+
+impl UDPBroadcastOutputProtocol {
+    pub fn new(port: u16, subnet_mask: &str) -> Self {
+        let inner = Arc::new(UDPBroadcastInner {
+            socket: Mutex::new(None),
+            broadcast_addr: Mutex::new(None),
+        });
+
+        let inner_clone = inner.clone();
+        let mask = subnet_mask.to_string();
+
+        tokio::spawn(async move {
+            let socket = match UdpSocket::bind("0.0.0.0:0").await {
+                Ok(s) => {
+                    s.set_broadcast(true).ok();
+                    s
+                }
+                Err(e) => {
+                    debug_log!("[UDP-Broadcast] Failed to create socket: {}", e);
+                    return;
+                }
+            };
+
+            // Compute broadcast address
+            let broadcast = compute_broadcast_address(&mask);
+            let addr = format!("{}:{}", broadcast, port);
+            debug_log!("[UDP-Broadcast] Broadcasting to {}", addr);
+
+            *inner_clone.socket.lock().await = Some(Arc::new(socket));
+            *inner_clone.broadcast_addr.lock().await = Some(addr);
+        });
+
+        Self { inner }
+    }
+}
 
 impl OutputProtocol for UDPBroadcastOutputProtocol {
-    async fn init(self) {
-        todo!()
+    async fn init(&self) {
+        // Everything set up in new()
     }
 
-    async fn send(self: std::sync::Arc<Self>, _str: &str) {
-        todo!()
+    async fn send(self: Arc<Self>, message: &str) {
+        let socket_guard = self.inner.socket.lock().await;
+        let addr_guard = self.inner.broadcast_addr.lock().await;
+        if let Some(ref socket) = *socket_guard {
+            if let Some(ref addr) = *addr_guard {
+                let bytes = format!("{}\n", message);
+                let _ = socket.send_to(bytes.as_bytes(), addr).await;
+            }
+        }
+    }
+}
+
+/// Compute the broadcast address from the local machine's IP and subnet mask.
+fn compute_broadcast_address(subnet_mask: &str) -> String {
+    // Try to find a local IPv4 address
+    if let Ok(addr) = get_local_ipv4() {
+        let mask: u32 = subnet_mask
+            .split('.')
+            .filter_map(|o| o.parse::<u32>().ok())
+            .fold(0u32, |acc, o| (acc << 8) | o);
+
+        let ip_parts: Vec<u32> = addr
+            .split('.')
+            .filter_map(|o| o.parse::<u32>().ok())
+            .collect();
+
+        if ip_parts.len() == 4 && mask != 0 {
+            let ip_int = ip_parts.iter().fold(0u32, |acc, o| (acc << 8) | o);
+            let broadcast_int = ip_int | !mask;
+            let b = broadcast_int.to_be_bytes();
+            return format!("{}.{}.{}.{}", b[0], b[1], b[2], b[3]);
+        }
     }
+
+    // Fallback
+    "255.255.255.255".to_string()
+}
+
+/// Get the local machine's primary IPv4 address.
+fn get_local_ipv4() -> Result<String, std::io::Error> {
+    // Use UDP connect trick to find the local IP
+    let socket = std::net::UdpSocket::bind("0.0.0.0:0")?;
+    socket.connect("8.8.8.8:80")?;
+    let local = socket.local_addr()?;
+    Ok(local.ip().to_string())
 }