feature: atproto-tap crate and tools

Cargo.lock

···
 ]
 
 [[package]]
+name = "atproto-tap"
+version = "0.13.0"
+dependencies = [
+ "atproto-client",
+ "atproto-identity",
+ "base64",
+ "clap",
+ "compact_str",
+ "futures",
+ "http",
+ "itoa",
+ "reqwest",
+ "serde",
+ "serde_json",
+ "thiserror 2.0.17",
+ "tokio",
+ "tokio-stream",
+ "tokio-websockets",
+ "tracing",
+ "tracing-subscriber",
+]
+
+[[package]]
 name = "atproto-xrpcs"
 version = "0.13.0"
 dependencies = [
···
 checksum = "d71b6127be86fdcfddb610f7182ac57211d4b18a3e9c82eb2d17662f2227ad6a"
 
 [[package]]
+name = "castaway"
+version = "0.2.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "dec551ab6e7578819132c713a93c022a05d60159dc86e7a7050223577484c55a"
+dependencies = [
+ "rustversion",
+]
+
+[[package]]
 name = "cbor4ii"
 version = "0.2.14"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
 version = "1.0.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b05b61dc5112cbb17e4b6cd61790d9845d13888356391624cbe7e41efeac1e75"
+
+[[package]]
+name = "compact_str"
+version = "0.8.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3b79c4069c6cad78e2e0cdfcbd26275770669fb39fd308a752dc110e83b9af32"
+dependencies = [
+ "castaway",
+ "cfg-if",
+ "itoa",
+ "rustversion",
+ "ryu",
+ "serde",
+ "static_assertions",
+]
 
 [[package]]
 name = "const-oid"
···
 checksum = "6ce2be8dc25455e1f91df71bfa12ad37d7af1092ae736f3a6cd0e37bc7810596"
 
 [[package]]
+name = "static_assertions"
+version = "1.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a2eb9349b6444b326872e140eb1cf5e7c522154d69e7a0ffb0fb81c06b37543f"
+
+[[package]]
 name = "strsim"
 version = "0.11.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
 checksum = "1729aa945f29d91ba541258c8df89027d5792d85a8841fb65e8bf0f4ede4ef61"
 dependencies = [
  "rustls",
+ "tokio",
+]
+
+[[package]]
+name = "tokio-stream"
+version = "0.1.17"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "eca58d7bba4a75707817a2c44174253f9236b2d5fbd055602e9d5c07c139a047"
+dependencies = [
+ "futures-core",
+ "pin-project-lite",
  "tokio",
 ]
 

Cargo.toml

···
     "crates/atproto-oauth-axum",
     "crates/atproto-oauth",
     "crates/atproto-record",
+    "crates/atproto-tap",
     "crates/atproto-xrpcs-helloworld",
     "crates/atproto-xrpcs",
     "crates/atproto-lexicon",
···
 atproto-oauth-aip = { version = "0.13.0", path = "crates/atproto-oauth-aip" }
 atproto-oauth-axum = { version = "0.13.0", path = "crates/atproto-oauth-axum" }
 atproto-record = { version = "0.13.0", path = "crates/atproto-record" }
+atproto-tap = { version = "0.13.0", path = "crates/atproto-tap" }
 atproto-xrpcs = { version = "0.13.0", path = "crates/atproto-xrpcs" }
 
 ammonia = "4.0"

crates/atproto-jetstream/src/consumer.rs

···
                     break;
                 },
                 () = &mut sleeper => {
-                        // consumer_control_insert(&self.pool, &self.config.jetstream_hostname, time_usec).await?;
-
                         sleeper.as_mut().reset(Instant::now() + interval);
                 },
                 item = client.next() => {

crates/atproto-tap/Cargo.toml

···
+[package]
+name = "atproto-tap"
+version = "0.13.0"
+description = "AT Protocol TAP (Trusted Attestation Protocol) service consumer"
+readme = "README.md"
+homepage = "https://tangled.sh/@smokesignal.events/atproto-identity-rs"
+documentation = "https://docs.rs/atproto-tap"
+
+edition.workspace = true
+rust-version.workspace = true
+authors.workspace = true
+repository.workspace = true
+license.workspace = true
+keywords.workspace = true
+categories.workspace = true
+
+[dependencies]
+tokio = { workspace = true, features = ["sync", "time"] }
+tokio-stream = "0.1"
+tokio-websockets = { workspace = true }
+futures = { workspace = true }
+reqwest = { workspace = true }
+serde = { workspace = true }
+serde_json = { workspace = true }
+thiserror = { workspace = true }
+tracing = { workspace = true }
+http = { workspace = true }
+base64 = { workspace = true }
+atproto-identity.workspace = true
+atproto-client = { workspace = true, optional = true }
+
+# Memory efficiency
+compact_str = { version = "0.8", features = ["serde"] }
+itoa = "1.0"
+
+# Optional for CLI
+clap = { workspace = true, optional = true }
+tracing-subscriber = { version = "0.3", features = ["env-filter"], optional = true }
+
+[features]
+default = []
+cli = ["dep:clap", "dep:tracing-subscriber", "dep:atproto-client", "tokio/rt-multi-thread", "tokio/macros", "tokio/signal"]
+
+[[bin]]
+name = "atproto-tap-client"
+required-features = ["cli"]
+
+[[bin]]
+name = "atproto-tap-extras"
+required-features = ["cli"]
+
+[lints]
+workspace = true

crates/atproto-tap/src/bin/atproto-tap-client.rs

···
+//! Command-line client for TAP services.
+//!
+//! This tool provides commands for consuming TAP events and managing tracked repositories.
+//!
+//! # Usage
+//!
+//! ```bash
+//! # Stream events from a TAP service
+//! cargo run --features cli --bin atproto-tap-client -- localhost:2480 read
+//!
+//! # Stream with authentication and filters
+//! cargo run --features cli --bin atproto-tap-client -- localhost:2480 -p secret read --live-only
+//!
+//! # Add repositories to track
+//! cargo run --features cli --bin atproto-tap-client -- localhost:2480 -p secret repos add did:plc:xyz did:plc:abc
+//!
+//! # Remove repositories from tracking
+//! cargo run --features cli --bin atproto-tap-client -- localhost:2480 -p secret repos remove did:plc:xyz
+//!
+//! # Resolve a DID to its DID document
+//! cargo run --features cli --bin atproto-tap-client -- localhost:2480 resolve did:plc:xyz
+//!
+//! # Resolve a DID and only output the handle
+//! cargo run --features cli --bin atproto-tap-client -- localhost:2480 resolve did:plc:xyz --handle-only
+//!
+//! # Get repository tracking info
+//! cargo run --features cli --bin atproto-tap-client -- localhost:2480 info did:plc:xyz
+//! ```
+
+use atproto_tap::{TapClient, TapConfig, TapEvent, connect};
+use clap::{Parser, Subcommand};
+use std::time::Duration;
+use tokio_stream::StreamExt;
+
+/// TAP service client for consuming events and managing repositories.
+#[derive(Parser)]
+#[command(
+    name = "atproto-tap-client",
+    version,
+    about = "TAP service client for AT Protocol",
+    long_about = "Connect to a TAP service to stream repository/identity events or manage tracked repositories.\n\n\
+                  Events are printed to stdout as JSON, one per line.\n\
+                  Use Ctrl+C to gracefully stop the consumer."
+)]
+struct Args {
+    /// TAP service hostname (e.g., localhost:2480)
+    hostname: String,
+
+    /// Admin password for authentication
+    #[arg(short, long, global = true)]
+    password: Option<String>,
+
+    #[command(subcommand)]
+    command: Command,
+}
+
+#[derive(Subcommand)]
+enum Command {
+    /// Connect to TAP and stream events as JSON
+    Read {
+        /// Disable acknowledgments
+        #[arg(long)]
+        no_acks: bool,
+
+        /// Maximum reconnection attempts (0 = unlimited)
+        #[arg(long, default_value = "0")]
+        max_reconnects: u32,
+
+        /// Print debug information to stderr
+        #[arg(short, long)]
+        debug: bool,
+
+        /// Filter to specific collections (comma-separated)
+        #[arg(long)]
+        collections: Option<String>,
+
+        /// Only show live events (skip backfill)
+        #[arg(long)]
+        live_only: bool,
+    },
+
+    /// Manage tracked repositories
+    Repos {
+        #[command(subcommand)]
+        action: ReposAction,
+    },
+
+    /// Resolve a DID to its DID document
+    Resolve {
+        /// DID to resolve (e.g., did:plc:xyz123)
+        did: String,
+
+        /// Only output the handle (instead of full DID document)
+        #[arg(long)]
+        handle_only: bool,
+    },
+
+    /// Get tracking info for a repository
+    Info {
+        /// DID to get info for (e.g., did:plc:xyz123)
+        did: String,
+    },
+}
+
+#[derive(Subcommand)]
+enum ReposAction {
+    /// Add repositories to track
+    Add {
+        /// DIDs to add (e.g., did:plc:xyz123)
+        #[arg(required = true)]
+        dids: Vec<String>,
+    },
+
+    /// Remove repositories from tracking
+    Remove {
+        /// DIDs to remove
+        #[arg(required = true)]
+        dids: Vec<String>,
+    },
+}
+
+#[tokio::main]
+async fn main() {
+    let args = Args::parse();
+
+    match args.command {
+        Command::Read {
+            no_acks,
+            max_reconnects,
+            debug,
+            collections,
+            live_only,
+        } => {
+            run_read(
+                &args.hostname,
+                args.password,
+                no_acks,
+                max_reconnects,
+                debug,
+                collections,
+                live_only,
+            )
+            .await;
+        }
+        Command::Repos { action } => {
+            run_repos(&args.hostname, args.password, action).await;
+        }
+        Command::Resolve { did, handle_only } => {
+            run_resolve(&args.hostname, args.password, &did, handle_only).await;
+        }
+        Command::Info { did } => {
+            run_info(&args.hostname, args.password, &did).await;
+        }
+    }
+}
+
+async fn run_read(
+    hostname: &str,
+    password: Option<String>,
+    no_acks: bool,
+    max_reconnects: u32,
+    debug: bool,
+    collections: Option<String>,
+    live_only: bool,
+) {
+    // Initialize tracing if debug mode
+    if debug {
+        tracing_subscriber::fmt()
+            .with_env_filter("atproto_tap=debug")
+            .with_writer(std::io::stderr)
+            .init();
+    }
+
+    // Build configuration
+    let mut config_builder = TapConfig::builder()
+        .hostname(hostname)
+        .send_acks(!no_acks);
+
+    if let Some(password) = password {
+        config_builder = config_builder.admin_password(password);
+    }
+
+    if max_reconnects > 0 {
+        config_builder = config_builder.max_reconnect_attempts(Some(max_reconnects));
+    }
+
+    // Set reasonable defaults for CLI usage
+    config_builder = config_builder
+        .initial_reconnect_delay(Duration::from_secs(1))
+        .max_reconnect_delay(Duration::from_secs(30));
+
+    let config = config_builder.build();
+
+    eprintln!("Connecting to TAP service at {}...", hostname);
+
+    let mut stream = connect(config);
+
+    // Parse collection filters
+    let collection_filters: Vec<String> = collections
+        .map(|c| c.split(',').map(|s| s.trim().to_string()).collect())
+        .unwrap_or_default();
+
+    // Handle Ctrl+C
+    let ctrl_c = tokio::signal::ctrl_c();
+    tokio::pin!(ctrl_c);
+
+    loop {
+        tokio::select! {
+            Some(result) = stream.next() => {
+                match result {
+                    Ok(event) => {
+                        // Apply filters
+                        let should_print = match event.as_ref() {
+                            TapEvent::Record { record, .. } => {
+                                // Filter by live flag
+                                if live_only && !record.live {
+                                    false
+                                }
+                                // Filter by collection
+                                else if !collection_filters.is_empty() {
+                                    collection_filters.iter().any(|c| record.collection.as_ref() == c)
+                                } else {
+                                    true
+                                }
+                            }
+                            TapEvent::Identity { .. } => !live_only, // Always show identity unless live_only
+                        };
+
+                        if should_print {
+                            // Print as JSON to stdout
+                            match serde_json::to_string(event.as_ref()) {
+                                Ok(json) => println!("{}", json),
+                                Err(e) => {
+                                    eprintln!("Failed to serialize event: {}", e);
+                                }
+                            }
+                        }
+                    }
+                    Err(e) => {
+                        eprintln!("Error: {}", e);
+
+                        // Exit on fatal errors
+                        if e.is_fatal() {
+                            eprintln!("Fatal error, exiting");
+                            std::process::exit(1);
+                        }
+                    }
+                }
+            }
+            _ = &mut ctrl_c => {
+                eprintln!("\nReceived Ctrl+C, shutting down...");
+                stream.close().await;
+                break;
+            }
+        }
+    }
+
+    eprintln!("Client stopped");
+}
+
+async fn run_repos(hostname: &str, password: Option<String>, action: ReposAction) {
+    let client = TapClient::new(hostname, password);
+
+    match action {
+        ReposAction::Add { dids } => {
+            let did_refs: Vec<&str> = dids.iter().map(|s| s.as_str()).collect();
+
+            match client.add_repos(&did_refs).await {
+                Ok(()) => {
+                    eprintln!("Added {} repository(ies) to tracking", dids.len());
+                    for did in &dids {
+                        println!("{}", did);
+                    }
+                }
+                Err(e) => {
+                    eprintln!("Failed to add repositories: {}", e);
+                    std::process::exit(1);
+                }
+            }
+        }
+        ReposAction::Remove { dids } => {
+            let did_refs: Vec<&str> = dids.iter().map(|s| s.as_str()).collect();
+
+            match client.remove_repos(&did_refs).await {
+                Ok(()) => {
+                    eprintln!("Removed {} repository(ies) from tracking", dids.len());
+                    for did in &dids {
+                        println!("{}", did);
+                    }
+                }
+                Err(e) => {
+                    eprintln!("Failed to remove repositories: {}", e);
+                    std::process::exit(1);
+                }
+            }
+        }
+    }
+}
+
+async fn run_resolve(hostname: &str, password: Option<String>, did: &str, handle_only: bool) {
+    let client = TapClient::new(hostname, password);
+
+    match client.resolve(did).await {
+        Ok(doc) => {
+            if handle_only {
+                // Use the handles() method from atproto_identity::model::Document
+                match doc.handles() {
+                    Some(handle) => println!("{}", handle),
+                    None => {
+                        eprintln!("No handle found in DID document");
+                        std::process::exit(1);
+                    }
+                }
+            } else {
+                // Print full DID document as JSON
+                match serde_json::to_string_pretty(&doc) {
+                    Ok(json) => println!("{}", json),
+                    Err(e) => {
+                        eprintln!("Failed to serialize DID document: {}", e);
+                        std::process::exit(1);
+                    }
+                }
+            }
+        }
+        Err(e) => {
+            eprintln!("Failed to resolve DID: {}", e);
+            std::process::exit(1);
+        }
+    }
+}
+
+async fn run_info(hostname: &str, password: Option<String>, did: &str) {
+    let client = TapClient::new(hostname, password);
+
+    match client.info(did).await {
+        Ok(info) => {
+            // Print as JSON for easy parsing
+            match serde_json::to_string_pretty(&info) {
+                Ok(json) => println!("{}", json),
+                Err(e) => {
+                    eprintln!("Failed to serialize info: {}", e);
+                    std::process::exit(1);
+                }
+            }
+        }
+        Err(e) => {
+            eprintln!("Failed to get repository info: {}", e);
+            std::process::exit(1);
+        }
+    }
+}

crates/atproto-tap/src/bin/atproto-tap-extras.rs

···
+//! Additional TAP client utilities for AT Protocol.
+//!
+//! This tool provides extra commands for managing TAP tracked repositories
+//! based on social graph data.
+//!
+//! # Usage
+//!
+//! ```bash
+//! # Add all accounts followed by a DID to TAP tracking
+//! cargo run --features cli --bin atproto-tap-extras -- localhost:2480 repos-add-followers did:plc:xyz
+//!
+//! # With authentication
+//! cargo run --features cli --bin atproto-tap-extras -- localhost:2480 -p secret repos-add-followers did:plc:xyz
+//! ```
+
+use atproto_client::client::Auth;
+use atproto_client::com::atproto::repo::{ListRecordsParams, list_records};
+use atproto_identity::plc::query as plc_query;
+use atproto_tap::TapClient;
+use clap::{Parser, Subcommand};
+use serde::Deserialize;
+
+/// TAP extras utility for managing tracked repositories.
+#[derive(Parser)]
+#[command(
+    name = "atproto-tap-extras",
+    version,
+    about = "TAP extras utility for AT Protocol",
+    long_about = "Additional utilities for managing TAP tracked repositories based on social graph data."
+)]
+struct Args {
+    /// TAP service hostname (e.g., localhost:2480)
+    hostname: String,
+
+    /// Admin password for TAP authentication
+    #[arg(short, long, global = true)]
+    password: Option<String>,
+
+    /// PLC directory hostname for DID resolution
+    #[arg(long, default_value = "plc.directory", global = true)]
+    plc_hostname: String,
+
+    #[command(subcommand)]
+    command: Command,
+}
+
+#[derive(Subcommand)]
+enum Command {
+    /// Add accounts followed by a DID to TAP tracking.
+    ///
+    /// Fetches all app.bsky.graph.follow records from the specified DID's repository
+    /// and adds the followed DIDs to TAP for tracking.
+    ReposAddFollowers {
+        /// DID to read followers from (e.g., did:plc:xyz123)
+        did: String,
+
+        /// Batch size for adding repos to TAP
+        #[arg(long, default_value = "100")]
+        batch_size: usize,
+
+        /// Dry run - print DIDs without adding to TAP
+        #[arg(long)]
+        dry_run: bool,
+    },
+}
+
+/// Follow record structure from app.bsky.graph.follow.
+#[derive(Debug, Deserialize)]
+struct FollowRecord {
+    /// The DID of the account being followed.
+    subject: String,
+}
+
+#[tokio::main]
+async fn main() {
+    let args = Args::parse();
+
+    match args.command {
+        Command::ReposAddFollowers {
+            did,
+            batch_size,
+            dry_run,
+        } => {
+            run_repos_add_followers(
+                &args.hostname,
+                args.password,
+                &args.plc_hostname,
+                &did,
+                batch_size,
+                dry_run,
+            )
+            .await;
+        }
+    }
+}
+
+async fn run_repos_add_followers(
+    tap_hostname: &str,
+    tap_password: Option<String>,
+    plc_hostname: &str,
+    did: &str,
+    batch_size: usize,
+    dry_run: bool,
+) {
+    let http_client = reqwest::Client::new();
+
+    // Resolve the DID to get the PDS endpoint
+    eprintln!("Resolving DID: {}", did);
+    let document = match plc_query(&http_client, plc_hostname, did).await {
+        Ok(doc) => doc,
+        Err(e) => {
+            eprintln!("Failed to resolve DID: {}", e);
+            std::process::exit(1);
+        }
+    };
+
+    let pds_endpoints = document.pds_endpoints();
+    if pds_endpoints.is_empty() {
+        eprintln!("No PDS endpoint found in DID document");
+        std::process::exit(1);
+    }
+    let pds_url = pds_endpoints[0];
+    eprintln!("Using PDS: {}", pds_url);
+
+    // Collect all followed DIDs
+    let mut followed_dids: Vec<String> = Vec::new();
+    let mut cursor: Option<String> = None;
+    let collection = "app.bsky.graph.follow".to_string();
+
+    eprintln!("Fetching follow records...");
+
+    loop {
+        let params = if let Some(c) = cursor.take() {
+            ListRecordsParams::new().limit(100).cursor(c)
+        } else {
+            ListRecordsParams::new().limit(100)
+        };
+
+        let response = match list_records::<FollowRecord>(
+            &http_client,
+            &Auth::None,
+            pds_url,
+            did.to_string(),
+            collection.clone(),
+            params,
+        )
+        .await
+        {
+            Ok(resp) => resp,
+            Err(e) => {
+                eprintln!("Failed to list records: {}", e);
+                std::process::exit(1);
+            }
+        };
+
+        for record in &response.records {
+            followed_dids.push(record.value.subject.clone());
+        }
+
+        eprintln!(
+            "  Fetched {} records (total: {})",
+            response.records.len(),
+            followed_dids.len()
+        );
+
+        match response.cursor {
+            Some(c) if !response.records.is_empty() => {
+                cursor = Some(c);
+            }
+            _ => break,
+        }
+    }
+
+    if followed_dids.is_empty() {
+        eprintln!("No follow records found");
+        return;
+    }
+
+    eprintln!("Found {} followed accounts", followed_dids.len());
+
+    if dry_run {
+        eprintln!("\nDry run - would add these DIDs to TAP:");
+        for did in &followed_dids {
+            println!("{}", did);
+        }
+        return;
+    }
+
+    // Add to TAP in batches
+    let tap_client = TapClient::new(tap_hostname, tap_password);
+    let mut added = 0;
+
+    for chunk in followed_dids.chunks(batch_size) {
+        let did_refs: Vec<&str> = chunk.iter().map(|s| s.as_str()).collect();
+
+        match tap_client.add_repos(&did_refs).await {
+            Ok(()) => {
+                added += chunk.len();
+                eprintln!("Added {} DIDs to TAP (total: {})", chunk.len(), added);
+            }
+            Err(e) => {
+                eprintln!("Failed to add repos to TAP: {}", e);
+                std::process::exit(1);
+            }
+        }
+    }
+
+    eprintln!("Successfully added {} DIDs to TAP", added);
+
+    // Print all added DIDs
+    for did in &followed_dids {
+        println!("{}", did);
+    }
+}

crates/atproto-tap/src/client.rs

···
+//! HTTP client for TAP management API.
+//!
+//! This module provides [`TapClient`] for interacting with the TAP service's
+//! HTTP management endpoints, including adding/removing tracked repositories.
+
+use crate::errors::TapError;
+use atproto_identity::model::Document;
+use base64::Engine;
+use base64::engine::general_purpose::STANDARD as BASE64;
+use reqwest::header::{AUTHORIZATION, CONTENT_TYPE, HeaderMap, HeaderValue};
+use serde::{Deserialize, Serialize};
+
+/// HTTP client for TAP management API.
+///
+/// Provides methods for managing which repositories the TAP service tracks,
+/// checking service health, and querying repository status.
+///
+/// # Example
+///
+/// ```ignore
+/// use atproto_tap::TapClient;
+///
+/// let client = TapClient::new("localhost:2480", Some("admin_password".to_string()));
+///
+/// // Add repositories to track
+/// client.add_repos(&["did:plc:xyz123", "did:plc:abc456"]).await?;
+///
+/// // Check health
+/// if client.health().await? {
+///     println!("TAP service is healthy");
+/// }
+/// ```
+#[derive(Debug, Clone)]
+pub struct TapClient {
+    http_client: reqwest::Client,
+    base_url: String,
+    auth_header: Option<HeaderValue>,
+}
+
+impl TapClient {
+    /// Create a new TAP management client.
+    ///
+    /// # Arguments
+    ///
+    /// * `hostname` - TAP service hostname (e.g., "localhost:2480")
+    /// * `admin_password` - Optional admin password for authentication
+    pub fn new(hostname: &str, admin_password: Option<String>) -> Self {
+        let auth_header = admin_password.map(|password| {
+            let credentials = format!("admin:{}", password);
+            let encoded = BASE64.encode(credentials.as_bytes());
+            HeaderValue::from_str(&format!("Basic {}", encoded))
+                .expect("Invalid auth header value")
+        });
+
+        Self {
+            http_client: reqwest::Client::new(),
+            base_url: format!("http://{}", hostname),
+            auth_header,
+        }
+    }
+
+    /// Create default headers for requests.
+    fn default_headers(&self) -> HeaderMap {
+        let mut headers = HeaderMap::new();
+        headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));
+        if let Some(auth) = &self.auth_header {
+            headers.insert(AUTHORIZATION, auth.clone());
+        }
+        headers
+    }
+
+    /// Add repositories to track.
+    ///
+    /// Sends a POST request to `/repos/add` with the list of DIDs.
+    ///
+    /// # Arguments
+    ///
+    /// * `dids` - Slice of DID strings to track
+    ///
+    /// # Example
+    ///
+    /// ```ignore
+    /// client.add_repos(&[
+    ///     "did:plc:z72i7hdynmk6r22z27h6tvur",
+    ///     "did:plc:ewvi7nxzyoun6zhxrhs64oiz",
+    /// ]).await?;
+    /// ```
+    pub async fn add_repos(&self, dids: &[&str]) -> Result<(), TapError> {
+        let url = format!("{}/repos/add", self.base_url);
+        let body = AddReposRequest {
+            dids: dids.iter().map(|s| s.to_string()).collect(),
+        };
+
+        let response = self
+            .http_client
+            .post(&url)
+            .headers(self.default_headers())
+            .json(&body)
+            .send()
+            .await?;
+
+        if response.status().is_success() {
+            tracing::debug!(count = dids.len(), "Added repositories to TAP");
+            Ok(())
+        } else {
+            let status = response.status().as_u16();
+            let message = response.text().await.unwrap_or_default();
+            Err(TapError::HttpResponseError { status, message })
+        }
+    }
+
+    /// Remove repositories from tracking.
+    ///
+    /// Sends a POST request to `/repos/remove` with the list of DIDs.
+    ///
+    /// # Arguments
+    ///
+    /// * `dids` - Slice of DID strings to stop tracking
+    pub async fn remove_repos(&self, dids: &[&str]) -> Result<(), TapError> {
+        let url = format!("{}/repos/remove", self.base_url);
+        let body = AddReposRequest {
+            dids: dids.iter().map(|s| s.to_string()).collect(),
+        };
+
+        let response = self
+            .http_client
+            .post(&url)
+            .headers(self.default_headers())
+            .json(&body)
+            .send()
+            .await?;
+
+        if response.status().is_success() {
+            tracing::debug!(count = dids.len(), "Removed repositories from TAP");
+            Ok(())
+        } else {
+            let status = response.status().as_u16();
+            let message = response.text().await.unwrap_or_default();
+            Err(TapError::HttpResponseError { status, message })
+        }
+    }
+
+    /// Check service health.
+    ///
+    /// Sends a GET request to `/health`.
+    ///
+    /// # Returns
+    ///
+    /// `true` if the service is healthy, `false` otherwise.
+    pub async fn health(&self) -> Result<bool, TapError> {
+        let url = format!("{}/health", self.base_url);
+
+        let response = self
+            .http_client
+            .get(&url)
+            .headers(self.default_headers())
+            .send()
+            .await?;
+
+        Ok(response.status().is_success())
+    }
+
+    /// Resolve a DID to its DID document.
+    ///
+    /// Sends a GET request to `/resolve/:did`.
+    ///
+    /// # Arguments
+    ///
+    /// * `did` - The DID to resolve
+    ///
+    /// # Returns
+    ///
+    /// The DID document for the identity.
+    pub async fn resolve(&self, did: &str) -> Result<Document, TapError> {
+        let url = format!("{}/resolve/{}", self.base_url, did);
+
+        let response = self
+            .http_client
+            .get(&url)
+            .headers(self.default_headers())
+            .send()
+            .await?;
+
+        if response.status().is_success() {
+            let doc: Document = response.json().await?;
+            Ok(doc)
+        } else {
+            let status = response.status().as_u16();
+            let message = response.text().await.unwrap_or_default();
+            Err(TapError::HttpResponseError { status, message })
+        }
+    }
+
+    /// Get info about a tracked repository.
+    ///
+    /// Sends a GET request to `/info/:did`.
+    ///
+    /// # Arguments
+    ///
+    /// * `did` - The DID to get info for
+    ///
+    /// # Returns
+    ///
+    /// Repository tracking information.
+    pub async fn info(&self, did: &str) -> Result<RepoInfo, TapError> {
+        let url = format!("{}/info/{}", self.base_url, did);
+
+        let response = self
+            .http_client
+            .get(&url)
+            .headers(self.default_headers())
+            .send()
+            .await?;
+
+        if response.status().is_success() {
+            let info: RepoInfo = response.json().await?;
+            Ok(info)
+        } else {
+            let status = response.status().as_u16();
+            let message = response.text().await.unwrap_or_default();
+            Err(TapError::HttpResponseError { status, message })
+        }
+    }
+}
+
+/// Request body for adding/removing repositories.
+#[derive(Debug, Serialize)]
+struct AddReposRequest {
+    dids: Vec<String>,
+}
+
+/// Repository tracking information.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct RepoInfo {
+    /// The repository DID.
+    pub did: Box<str>,
+    /// Current sync state.
+    pub state: RepoState,
+    /// The handle for the repository.
+    #[serde(default)]
+    pub handle: Option<Box<str>>,
+    /// Number of records in the repository.
+    #[serde(default)]
+    pub records: u64,
+    /// Current repository revision.
+    #[serde(default)]
+    pub rev: Option<Box<str>>,
+    /// Number of retries for syncing.
+    #[serde(default)]
+    pub retries: u32,
+    /// Error message if any.
+    #[serde(default)]
+    pub error: Option<Box<str>>,
+    /// Additional fields may be present depending on TAP version.
+    #[serde(flatten)]
+    pub extra: serde_json::Value,
+}
+
+/// Repository sync state.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
+#[serde(rename_all = "lowercase")]
+pub enum RepoState {
+    /// Repository is active and synced.
+    Active,
+    /// Repository is currently syncing.
+    Syncing,
+    /// Repository is fully synced.
+    Synced,
+    /// Sync failed for this repository.
+    Failed,
+    /// Repository is queued for sync.
+    Queued,
+    /// Unknown state.
+    #[serde(other)]
+    Unknown,
+}
+
+/// Deprecated alias for RepoState.
+#[deprecated(since = "0.13.0", note = "Use RepoState instead")]
+pub type RepoStatus = RepoState;
+
+impl std::fmt::Display for RepoState {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            RepoState::Active => write!(f, "active"),
+            RepoState::Syncing => write!(f, "syncing"),
+            RepoState::Synced => write!(f, "synced"),
+            RepoState::Failed => write!(f, "failed"),
+            RepoState::Queued => write!(f, "queued"),
+            RepoState::Unknown => write!(f, "unknown"),
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_client_creation() {
+        let client = TapClient::new("localhost:2480", None);
+        assert_eq!(client.base_url, "http://localhost:2480");
+        assert!(client.auth_header.is_none());
+
+        let client = TapClient::new("localhost:2480", Some("secret".to_string()));
+        assert!(client.auth_header.is_some());
+    }
+
+    #[test]
+    fn test_repo_state_display() {
+        assert_eq!(RepoState::Active.to_string(), "active");
+        assert_eq!(RepoState::Syncing.to_string(), "syncing");
+        assert_eq!(RepoState::Synced.to_string(), "synced");
+        assert_eq!(RepoState::Failed.to_string(), "failed");
+        assert_eq!(RepoState::Queued.to_string(), "queued");
+        assert_eq!(RepoState::Unknown.to_string(), "unknown");
+    }
+
+    #[test]
+    fn test_repo_state_deserialize() {
+        let json = r#""active""#;
+        let state: RepoState = serde_json::from_str(json).unwrap();
+        assert_eq!(state, RepoState::Active);
+
+        let json = r#""syncing""#;
+        let state: RepoState = serde_json::from_str(json).unwrap();
+        assert_eq!(state, RepoState::Syncing);
+
+        let json = r#""some_new_state""#;
+        let state: RepoState = serde_json::from_str(json).unwrap();
+        assert_eq!(state, RepoState::Unknown);
+    }
+
+    #[test]
+    fn test_repo_info_deserialize() {
+        let json = r#"{"did":"did:plc:cbkjy5n7bk3ax2wplmtjofq2","error":"","handle":"ngerakines.me","records":21382,"retries":0,"rev":"3mam4aazabs2m","state":"active"}"#;
+        let info: RepoInfo = serde_json::from_str(json).unwrap();
+        assert_eq!(&*info.did, "did:plc:cbkjy5n7bk3ax2wplmtjofq2");
+        assert_eq!(info.state, RepoState::Active);
+        assert_eq!(info.handle.as_deref(), Some("ngerakines.me"));
+        assert_eq!(info.records, 21382);
+        assert_eq!(info.retries, 0);
+        assert_eq!(info.rev.as_deref(), Some("3mam4aazabs2m"));
+        // Empty string deserializes as Some("")
+        assert_eq!(info.error.as_deref(), Some(""));
+    }
+
+    #[test]
+    fn test_repo_info_deserialize_minimal() {
+        // Test with only required fields
+        let json = r#"{"did":"did:plc:test","state":"syncing"}"#;
+        let info: RepoInfo = serde_json::from_str(json).unwrap();
+        assert_eq!(&*info.did, "did:plc:test");
+        assert_eq!(info.state, RepoState::Syncing);
+        assert_eq!(info.handle, None);
+        assert_eq!(info.records, 0);
+        assert_eq!(info.retries, 0);
+        assert_eq!(info.rev, None);
+        assert_eq!(info.error, None);
+    }
+
+    #[test]
+    fn test_add_repos_request_serialize() {
+        let req = AddReposRequest {
+            dids: vec!["did:plc:xyz".to_string(), "did:plc:abc".to_string()],
+        };
+        let json = serde_json::to_string(&req).unwrap();
+        assert!(json.contains("dids"));
+        assert!(json.contains("did:plc:xyz"));
+    }
+}

crates/atproto-tap/src/config.rs

···
+//! Configuration for TAP stream connections.
+//!
+//! This module provides the [`TapConfig`] struct for configuring TAP stream
+//! connections, including hostname, authentication, and reconnection behavior.
+
+use std::time::Duration;
+
+/// Configuration for a TAP stream connection.
+///
+/// Use [`TapConfig::builder()`] for ergonomic construction with defaults.
+///
+/// # Example
+///
+/// ```
+/// use atproto_tap::TapConfig;
+/// use std::time::Duration;
+///
+/// let config = TapConfig::builder()
+///     .hostname("localhost:2480")
+///     .admin_password("secret")
+///     .send_acks(true)
+///     .max_reconnect_attempts(Some(10))
+///     .build();
+/// ```
+#[derive(Debug, Clone)]
+pub struct TapConfig {
+    /// TAP service hostname (e.g., "localhost:2480").
+    ///
+    /// The WebSocket URL is constructed as `ws://{hostname}/channel`.
+    pub hostname: String,
+
+    /// Optional admin password for authentication.
+    ///
+    /// If set, HTTP Basic Auth is used with username "admin".
+    pub admin_password: Option<String>,
+
+    /// Whether to send acknowledgments for received messages.
+    ///
+    /// Default: `true`. Set to `false` if the TAP service has acks disabled.
+    pub send_acks: bool,
+
+    /// User-Agent header value for WebSocket connections.
+    pub user_agent: String,
+
+    /// Maximum reconnection attempts before giving up.
+    ///
+    /// `None` means unlimited reconnection attempts (default).
+    pub max_reconnect_attempts: Option<u32>,
+
+    /// Initial delay before first reconnection attempt.
+    ///
+    /// Default: 1 second.
+    pub initial_reconnect_delay: Duration,
+
+    /// Maximum delay between reconnection attempts.
+    ///
+    /// Default: 60 seconds.
+    pub max_reconnect_delay: Duration,
+
+    /// Multiplier for exponential backoff between reconnections.
+    ///
+    /// Default: 2.0 (doubles the delay each attempt).
+    pub reconnect_backoff_multiplier: f64,
+}
+
+impl Default for TapConfig {
+    fn default() -> Self {
+        Self {
+            hostname: "localhost:2480".to_string(),
+            admin_password: None,
+            send_acks: true,
+            user_agent: format!("atproto-tap/{}", env!("CARGO_PKG_VERSION")),
+            max_reconnect_attempts: None,
+            initial_reconnect_delay: Duration::from_secs(1),
+            max_reconnect_delay: Duration::from_secs(60),
+            reconnect_backoff_multiplier: 2.0,
+        }
+    }
+}
+
+impl TapConfig {
+    /// Create a new configuration builder with defaults.
+    pub fn builder() -> TapConfigBuilder {
+        TapConfigBuilder::default()
+    }
+
+    /// Create a minimal configuration for the given hostname.
+    pub fn new(hostname: impl Into<String>) -> Self {
+        Self {
+            hostname: hostname.into(),
+            ..Default::default()
+        }
+    }
+
+    /// Returns the WebSocket URL for the TAP channel.
+    pub fn ws_url(&self) -> String {
+        format!("ws://{}/channel", self.hostname)
+    }
+
+    /// Returns the HTTP base URL for the TAP management API.
+    pub fn http_base_url(&self) -> String {
+        format!("http://{}", self.hostname)
+    }
+}
+
+/// Builder for [`TapConfig`].
+#[derive(Debug, Clone, Default)]
+pub struct TapConfigBuilder {
+    config: TapConfig,
+}
+
+impl TapConfigBuilder {
+    /// Set the TAP service hostname.
+    pub fn hostname(mut self, hostname: impl Into<String>) -> Self {
+        self.config.hostname = hostname.into();
+        self
+    }
+
+    /// Set the admin password for authentication.
+    pub fn admin_password(mut self, password: impl Into<String>) -> Self {
+        self.config.admin_password = Some(password.into());
+        self
+    }
+
+    /// Set whether to send acknowledgments.
+    pub fn send_acks(mut self, send_acks: bool) -> Self {
+        self.config.send_acks = send_acks;
+        self
+    }
+
+    /// Set the User-Agent header value.
+    pub fn user_agent(mut self, user_agent: impl Into<String>) -> Self {
+        self.config.user_agent = user_agent.into();
+        self
+    }
+
+    /// Set the maximum reconnection attempts.
+    ///
+    /// `None` means unlimited attempts.
+    pub fn max_reconnect_attempts(mut self, max: Option<u32>) -> Self {
+        self.config.max_reconnect_attempts = max;
+        self
+    }
+
+    /// Set the initial reconnection delay.
+    pub fn initial_reconnect_delay(mut self, delay: Duration) -> Self {
+        self.config.initial_reconnect_delay = delay;
+        self
+    }
+
+    /// Set the maximum reconnection delay.
+    pub fn max_reconnect_delay(mut self, delay: Duration) -> Self {
+        self.config.max_reconnect_delay = delay;
+        self
+    }
+
+    /// Set the reconnection backoff multiplier.
+    pub fn reconnect_backoff_multiplier(mut self, multiplier: f64) -> Self {
+        self.config.reconnect_backoff_multiplier = multiplier;
+        self
+    }
+
+    /// Build the configuration.
+    pub fn build(self) -> TapConfig {
+        self.config
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_default_config() {
+        let config = TapConfig::default();
+        assert_eq!(config.hostname, "localhost:2480");
+        assert!(config.admin_password.is_none());
+        assert!(config.send_acks);
+        assert!(config.max_reconnect_attempts.is_none());
+        assert_eq!(config.initial_reconnect_delay, Duration::from_secs(1));
+        assert_eq!(config.max_reconnect_delay, Duration::from_secs(60));
+        assert!((config.reconnect_backoff_multiplier - 2.0).abs() < f64::EPSILON);
+    }
+
+    #[test]
+    fn test_builder() {
+        let config = TapConfig::builder()
+            .hostname("tap.example.com:2480")
+            .admin_password("secret123")
+            .send_acks(false)
+            .max_reconnect_attempts(Some(5))
+            .initial_reconnect_delay(Duration::from_millis(500))
+            .max_reconnect_delay(Duration::from_secs(30))
+            .reconnect_backoff_multiplier(1.5)
+            .build();
+
+        assert_eq!(config.hostname, "tap.example.com:2480");
+        assert_eq!(config.admin_password, Some("secret123".to_string()));
+        assert!(!config.send_acks);
+        assert_eq!(config.max_reconnect_attempts, Some(5));
+        assert_eq!(config.initial_reconnect_delay, Duration::from_millis(500));
+        assert_eq!(config.max_reconnect_delay, Duration::from_secs(30));
+        assert!((config.reconnect_backoff_multiplier - 1.5).abs() < f64::EPSILON);
+    }
+
+    #[test]
+    fn test_ws_url() {
+        let config = TapConfig::new("localhost:2480");
+        assert_eq!(config.ws_url(), "ws://localhost:2480/channel");
+
+        let config = TapConfig::new("tap.example.com:8080");
+        assert_eq!(config.ws_url(), "ws://tap.example.com:8080/channel");
+    }
+
+    #[test]
+    fn test_http_base_url() {
+        let config = TapConfig::new("localhost:2480");
+        assert_eq!(config.http_base_url(), "http://localhost:2480");
+    }
+}

crates/atproto-tap/src/connection.rs

···
+//! WebSocket connection management for TAP streams.
+//!
+//! This module handles the low-level WebSocket connection to a TAP service,
+//! including authentication and message sending/receiving.
+
+use crate::config::TapConfig;
+use crate::errors::TapError;
+use base64::Engine;
+use base64::engine::general_purpose::STANDARD as BASE64;
+use futures::{SinkExt, StreamExt};
+use http::Uri;
+use std::str::FromStr;
+use tokio_websockets::{ClientBuilder, Message, WebSocketStream};
+use tokio_websockets::MaybeTlsStream;
+use tokio::net::TcpStream;
+
+/// WebSocket connection to a TAP service.
+pub(crate) struct TapConnection {
+    /// The underlying WebSocket stream.
+    ws: WebSocketStream<MaybeTlsStream<TcpStream>>,
+    /// Pre-allocated buffer for acknowledgment messages.
+    ack_buffer: Vec<u8>,
+}
+
+impl TapConnection {
+    /// Establish a new WebSocket connection to the TAP service.
+    pub async fn connect(config: &TapConfig) -> Result<Self, TapError> {
+        let uri = Uri::from_str(&config.ws_url())
+            .map_err(|e| TapError::InvalidUrl(e.to_string()))?;
+
+        let mut builder = ClientBuilder::from_uri(uri);
+
+        // Add User-Agent header
+        builder = builder
+            .add_header(
+                http::header::USER_AGENT,
+                http::HeaderValue::from_str(&config.user_agent)
+                    .map_err(|e| TapError::ConnectionFailed(format!("Invalid user agent: {}", e)))?,
+            )
+            .map_err(|e| TapError::ConnectionFailed(format!("Failed to add header: {}", e)))?;
+
+        // Add Basic Auth header if password is configured
+        if let Some(password) = &config.admin_password {
+            let credentials = format!("admin:{}", password);
+            let encoded = BASE64.encode(credentials.as_bytes());
+            let auth_value = format!("Basic {}", encoded);
+
+            builder = builder
+                .add_header(
+                    http::header::AUTHORIZATION,
+                    http::HeaderValue::from_str(&auth_value)
+                        .map_err(|e| TapError::ConnectionFailed(format!("Invalid auth header: {}", e)))?,
+                )
+                .map_err(|e| TapError::ConnectionFailed(format!("Failed to add auth header: {}", e)))?;
+        }
+
+        // Connect
+        let (ws, _response) = builder
+            .connect()
+            .await
+            .map_err(|e| TapError::ConnectionFailed(e.to_string()))?;
+
+        tracing::debug!(hostname = %config.hostname, "Connected to TAP service");
+
+        Ok(Self {
+            ws,
+            ack_buffer: Vec::with_capacity(48), // {"type":"ack","id":18446744073709551615} is 40 bytes max
+        })
+    }
+
+    /// Receive the next message from the WebSocket.
+    ///
+    /// Returns `None` if the connection was closed cleanly.
+    pub async fn recv(&mut self) -> Result<Option<String>, TapError> {
+        match self.ws.next().await {
+            Some(Ok(msg)) => {
+                if msg.is_text() {
+                    msg.as_text()
+                        .map(|s| Some(s.to_string()))
+                        .ok_or_else(|| TapError::ParseError("Failed to get text from message".into()))
+                } else if msg.is_close() {
+                    tracing::debug!("Received close frame from TAP service");
+                    Ok(None)
+                } else {
+                    // Ignore ping/pong and binary messages
+                    tracing::trace!("Received non-text message, ignoring");
+                    // Recurse to get the next text message
+                    Box::pin(self.recv()).await
+                }
+            }
+            Some(Err(e)) => Err(TapError::ConnectionFailed(e.to_string())),
+            None => {
+                tracing::debug!("WebSocket stream ended");
+                Ok(None)
+            }
+        }
+    }
+
+    /// Send an acknowledgment for the given event ID.
+    ///
+    /// Uses a pre-allocated buffer and itoa for allocation-free formatting.
+    /// Format: `{"type":"ack","id":12345}`
+    pub async fn send_ack(&mut self, id: u64) -> Result<(), TapError> {
+        self.ack_buffer.clear();
+        self.ack_buffer.extend_from_slice(b"{\"type\":\"ack\",\"id\":");
+        let mut itoa_buf = itoa::Buffer::new();
+        self.ack_buffer.extend_from_slice(itoa_buf.format(id).as_bytes());
+        self.ack_buffer.push(b'}');
+
+        // All bytes are ASCII so this is always valid UTF-8
+        let msg = std::str::from_utf8(&self.ack_buffer)
+            .expect("ack buffer contains only ASCII");
+
+        self.ws
+            .send(Message::text(msg.to_string()))
+            .await
+            .map_err(|e| TapError::AckFailed(e.to_string()))?;
+
+        // Flush to ensure the ack is sent immediately
+        self.ws
+            .flush()
+            .await
+            .map_err(|e| TapError::AckFailed(format!("Failed to flush ack: {}", e)))?;
+
+        tracing::trace!(id, "Sent ack");
+        Ok(())
+    }
+
+    /// Close the WebSocket connection gracefully.
+    pub async fn close(&mut self) -> Result<(), TapError> {
+        self.ws
+            .close()
+            .await
+            .map_err(|e| TapError::ConnectionFailed(format!("Failed to close: {}", e)))?;
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    #[test]
+    fn test_ack_buffer_format() {
+        // Test that our manual JSON formatting is correct
+        // Format: {"type":"ack","id":12345}
+        let mut buffer = Vec::with_capacity(64);
+
+        let id: u64 = 12345;
+        buffer.clear();
+        buffer.extend_from_slice(b"{\"type\":\"ack\",\"id\":");
+        let mut itoa_buf = itoa::Buffer::new();
+        buffer.extend_from_slice(itoa_buf.format(id).as_bytes());
+        buffer.push(b'}');
+
+        let result = std::str::from_utf8(&buffer).unwrap();
+        assert_eq!(result, r#"{"type":"ack","id":12345}"#);
+
+        // Test max u64
+        let id: u64 = u64::MAX;
+        buffer.clear();
+        buffer.extend_from_slice(b"{\"type\":\"ack\",\"id\":");
+        buffer.extend_from_slice(itoa_buf.format(id).as_bytes());
+        buffer.push(b'}');
+
+        let result = std::str::from_utf8(&buffer).unwrap();
+        assert_eq!(result, r#"{"type":"ack","id":18446744073709551615}"#);
+        assert!(buffer.len() <= 64); // Fits in our pre-allocated buffer
+    }
+}

crates/atproto-tap/src/errors.rs

···
+//! Error types for TAP operations.
+//!
+//! This module defines the error types returned by TAP stream and client operations.
+
+use thiserror::Error;
+
+/// Errors that can occur during TAP operations.
+#[derive(Debug, Error)]
+pub enum TapError {
+    /// WebSocket connection failed.
+    #[error("error-atproto-tap-connection-1 WebSocket connection failed: {0}")]
+    ConnectionFailed(String),
+
+    /// Connection was closed unexpectedly.
+    #[error("error-atproto-tap-connection-2 Connection closed unexpectedly")]
+    ConnectionClosed,
+
+    /// Maximum reconnection attempts exceeded.
+    #[error("error-atproto-tap-connection-3 Maximum reconnection attempts exceeded after {0} attempts")]
+    MaxReconnectAttemptsExceeded(u32),
+
+    /// Authentication failed.
+    #[error("error-atproto-tap-auth-1 Authentication failed: {0}")]
+    AuthenticationFailed(String),
+
+    /// Failed to parse a message from the server.
+    #[error("error-atproto-tap-parse-1 Failed to parse message: {0}")]
+    ParseError(String),
+
+    /// Failed to send an acknowledgment.
+    #[error("error-atproto-tap-ack-1 Failed to send acknowledgment: {0}")]
+    AckFailed(String),
+
+    /// HTTP request failed.
+    #[error("error-atproto-tap-http-1 HTTP request failed: {0}")]
+    HttpError(String),
+
+    /// HTTP response indicated an error.
+    #[error("error-atproto-tap-http-2 HTTP error response: {status} - {message}")]
+    HttpResponseError {
+        /// HTTP status code.
+        status: u16,
+        /// Error message from response.
+        message: String,
+    },
+
+    /// Invalid URL.
+    #[error("error-atproto-tap-url-1 Invalid URL: {0}")]
+    InvalidUrl(String),
+
+    /// I/O error.
+    #[error("error-atproto-tap-io-1 I/O error: {0}")]
+    IoError(#[from] std::io::Error),
+
+    /// JSON serialization/deserialization error.
+    #[error("error-atproto-tap-json-1 JSON error: {0}")]
+    JsonError(#[from] serde_json::Error),
+
+    /// Stream has been closed and cannot be used.
+    #[error("error-atproto-tap-stream-1 Stream is closed")]
+    StreamClosed,
+
+    /// Operation timed out.
+    #[error("error-atproto-tap-timeout-1 Operation timed out")]
+    Timeout,
+}
+
+impl TapError {
+    /// Returns true if this error indicates a connection issue that may be recoverable.
+    pub fn is_connection_error(&self) -> bool {
+        matches!(
+            self,
+            TapError::ConnectionFailed(_)
+                | TapError::ConnectionClosed
+                | TapError::IoError(_)
+                | TapError::Timeout
+        )
+    }
+
+    /// Returns true if this error is a parse error that doesn't affect connection state.
+    pub fn is_parse_error(&self) -> bool {
+        matches!(self, TapError::ParseError(_) | TapError::JsonError(_))
+    }
+
+    /// Returns true if this error is fatal and the stream should not attempt recovery.
+    pub fn is_fatal(&self) -> bool {
+        matches!(
+            self,
+            TapError::MaxReconnectAttemptsExceeded(_)
+                | TapError::AuthenticationFailed(_)
+                | TapError::StreamClosed
+        )
+    }
+}
+
+impl From<reqwest::Error> for TapError {
+    fn from(err: reqwest::Error) -> Self {
+        if err.is_timeout() {
+            TapError::Timeout
+        } else if err.is_connect() {
+            TapError::ConnectionFailed(err.to_string())
+        } else {
+            TapError::HttpError(err.to_string())
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_error_classification() {
+        assert!(TapError::ConnectionFailed("test".into()).is_connection_error());
+        assert!(TapError::ConnectionClosed.is_connection_error());
+        assert!(TapError::Timeout.is_connection_error());
+
+        assert!(TapError::ParseError("test".into()).is_parse_error());
+        assert!(TapError::JsonError(serde_json::from_str::<()>("invalid").unwrap_err()).is_parse_error());
+
+        assert!(TapError::MaxReconnectAttemptsExceeded(5).is_fatal());
+        assert!(TapError::AuthenticationFailed("test".into()).is_fatal());
+        assert!(TapError::StreamClosed.is_fatal());
+
+        // Non-fatal errors
+        assert!(!TapError::ConnectionFailed("test".into()).is_fatal());
+        assert!(!TapError::ParseError("test".into()).is_fatal());
+    }
+
+    #[test]
+    fn test_error_display() {
+        let err = TapError::ConnectionFailed("refused".to_string());
+        assert!(err.to_string().contains("error-atproto-tap-connection-1"));
+        assert!(err.to_string().contains("refused"));
+
+        let err = TapError::HttpResponseError {
+            status: 404,
+            message: "Not Found".to_string(),
+        };
+        assert!(err.to_string().contains("404"));
+        assert!(err.to_string().contains("Not Found"));
+    }
+}

crates/atproto-tap/src/events.rs

···
+//! TAP event types for AT Protocol record and identity events.
+//!
+//! This module defines the event structures received from a TAP service.
+//! Events are optimized for memory efficiency using:
+//! - `CompactString` for small strings (SSO for ≤24 bytes)
+//! - `Box<str>` for immutable strings (no capacity overhead)
+//! - `serde_json::Value` for record payloads (allows lazy access)
+
+use compact_str::CompactString;
+use serde::{Deserialize, Serialize, de::DeserializeOwned};
+
+/// A TAP event received from the stream.
+///
+/// TAP delivers two types of events:
+/// - `Record`: Repository record changes (create, update, delete)
+/// - `Identity`: Identity/handle changes for accounts
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(tag = "type", rename_all = "lowercase")]
+pub enum TapEvent {
+    /// A repository record event (create, update, or delete).
+    Record {
+        /// Sequential event identifier.
+        id: u64,
+        /// The record event data.
+        record: RecordEvent,
+    },
+    /// An identity change event.
+    Identity {
+        /// Sequential event identifier.
+        id: u64,
+        /// The identity event data.
+        identity: IdentityEvent,
+    },
+}
+
+impl TapEvent {
+    /// Returns the event ID.
+    pub fn id(&self) -> u64 {
+        match self {
+            TapEvent::Record { id, .. } => *id,
+            TapEvent::Identity { id, .. } => *id,
+        }
+    }
+}
+
+/// A repository record event from TAP.
+///
+/// Contains information about a record change in a user's repository,
+/// including the action taken and the record data (for creates/updates).
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct RecordEvent {
+    /// True if from live firehose, false if from backfill/resync.
+    ///
+    /// During initial sync or recovery, TAP delivers historical events
+    /// with `live: false`. Once caught up, live events have `live: true`.
+    pub live: bool,
+
+    /// Repository revision identifier.
+    ///
+    /// Typically 13 characters, stored inline via CompactString SSO.
+    pub rev: CompactString,
+
+    /// Actor DID (e.g., "did:plc:xyz123").
+    pub did: Box<str>,
+
+    /// Collection NSID (e.g., "app.bsky.feed.post").
+    pub collection: Box<str>,
+
+    /// Record key within the collection.
+    ///
+    /// Typically a TID (13 characters), stored inline via CompactString SSO.
+    pub rkey: CompactString,
+
+    /// The action performed on the record.
+    pub action: RecordAction,
+
+    /// Content identifier (CID) of the record.
+    ///
+    /// Present for create and update actions, absent for delete.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub cid: Option<CompactString>,
+
+    /// Record data as JSON value.
+    ///
+    /// Present for create and update actions, absent for delete.
+    /// Use [`parse_record`](Self::parse_record) to deserialize on demand.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub record: Option<serde_json::Value>,
+}
+
+impl RecordEvent {
+    /// Parse the record payload into a typed structure.
+    ///
+    /// This method deserializes the raw JSON on demand, avoiding
+    /// unnecessary allocations when the record data isn't needed.
+    ///
+    /// # Errors
+    ///
+    /// Returns an error if the record is absent (delete events) or
+    /// if deserialization fails.
+    ///
+    /// # Example
+    ///
+    /// ```ignore
+    /// use serde::Deserialize;
+    ///
+    /// #[derive(Deserialize)]
+    /// struct Post {
+    ///     text: String,
+    ///     #[serde(rename = "createdAt")]
+    ///     created_at: String,
+    /// }
+    ///
+    /// let post: Post = record_event.parse_record()?;
+    /// println!("Post text: {}", post.text);
+    /// ```
+    pub fn parse_record<T: DeserializeOwned>(&self) -> Result<T, serde_json::Error> {
+        match &self.record {
+            Some(value) => serde_json::from_value(value.clone()),
+            None => Err(serde::de::Error::custom("no record data (delete event)")),
+        }
+    }
+
+    /// Returns the record as a JSON Value reference, if present.
+    pub fn record_value(&self) -> Option<&serde_json::Value> {
+        self.record.as_ref()
+    }
+
+    /// Returns true if this is a delete event.
+    pub fn is_delete(&self) -> bool {
+        self.action == RecordAction::Delete
+    }
+
+    /// Returns the AT-URI for this record.
+    ///
+    /// Format: `at://{did}/{collection}/{rkey}`
+    pub fn at_uri(&self) -> String {
+        format!("at://{}/{}/{}", self.did, self.collection, self.rkey)
+    }
+}
+
+/// The action performed on a record.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
+#[serde(rename_all = "lowercase")]
+pub enum RecordAction {
+    /// A new record was created.
+    Create,
+    /// An existing record was updated.
+    Update,
+    /// A record was deleted.
+    Delete,
+}
+
+impl std::fmt::Display for RecordAction {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            RecordAction::Create => write!(f, "create"),
+            RecordAction::Update => write!(f, "update"),
+            RecordAction::Delete => write!(f, "delete"),
+        }
+    }
+}
+
+/// An identity change event from TAP.
+///
+/// Contains information about handle or account status changes.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct IdentityEvent {
+    /// Actor DID.
+    pub did: Box<str>,
+
+    /// Current handle for the account.
+    pub handle: Box<str>,
+
+    /// Whether the account is currently active.
+    #[serde(default)]
+    pub is_active: bool,
+
+    /// Account status.
+    #[serde(default)]
+    pub status: IdentityStatus,
+}
+
+/// Account status in an identity event.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Default, Serialize, Deserialize)]
+#[serde(rename_all = "lowercase")]
+pub enum IdentityStatus {
+    /// Account is active and in good standing.
+    #[default]
+    Active,
+    /// Account has been deactivated by the user.
+    Deactivated,
+    /// Account has been suspended.
+    Suspended,
+    /// Account has been deleted.
+    Deleted,
+    /// Account has been taken down.
+    Takendown,
+}
+
+impl std::fmt::Display for IdentityStatus {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            IdentityStatus::Active => write!(f, "active"),
+            IdentityStatus::Deactivated => write!(f, "deactivated"),
+            IdentityStatus::Suspended => write!(f, "suspended"),
+            IdentityStatus::Deleted => write!(f, "deleted"),
+            IdentityStatus::Takendown => write!(f, "takendown"),
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_parse_record_event() {
+        let json = r#"{
+            "id": 12345,
+            "type": "record",
+            "record": {
+                "live": true,
+                "rev": "3lyileto4q52k",
+                "did": "did:plc:z72i7hdynmk6r22z27h6tvur",
+                "collection": "app.bsky.feed.post",
+                "rkey": "3lyiletddxt2c",
+                "action": "create",
+                "cid": "bafyreigroo6vhxt62ufcndhaxzas6btq4jmniuz4egszbwuqgiyisqwqoy",
+                "record": {"$type": "app.bsky.feed.post", "text": "Hello world!", "createdAt": "2025-01-01T00:00:00Z"}
+            }
+        }"#;
+
+        let event: TapEvent = serde_json::from_str(json).expect("Failed to parse");
+
+        match event {
+            TapEvent::Record { id, record } => {
+                assert_eq!(id, 12345);
+                assert!(record.live);
+                assert_eq!(record.rev.as_str(), "3lyileto4q52k");
+                assert_eq!(&*record.did, "did:plc:z72i7hdynmk6r22z27h6tvur");
+                assert_eq!(&*record.collection, "app.bsky.feed.post");
+                assert_eq!(record.rkey.as_str(), "3lyiletddxt2c");
+                assert_eq!(record.action, RecordAction::Create);
+                assert!(record.cid.is_some());
+                assert!(record.record.is_some());
+
+                // Test lazy parsing
+                #[derive(Deserialize)]
+                struct Post {
+                    text: String,
+                }
+                let post: Post = record.parse_record().expect("Failed to parse record");
+                assert_eq!(post.text, "Hello world!");
+            }
+            _ => panic!("Expected Record event"),
+        }
+    }
+
+    #[test]
+    fn test_parse_delete_event() {
+        let json = r#"{
+            "id": 12346,
+            "type": "record",
+            "record": {
+                "live": true,
+                "rev": "3lyileto4q52k",
+                "did": "did:plc:z72i7hdynmk6r22z27h6tvur",
+                "collection": "app.bsky.feed.post",
+                "rkey": "3lyiletddxt2c",
+                "action": "delete"
+            }
+        }"#;
+
+        let event: TapEvent = serde_json::from_str(json).expect("Failed to parse");
+
+        match event {
+            TapEvent::Record { id, record } => {
+                assert_eq!(id, 12346);
+                assert_eq!(record.action, RecordAction::Delete);
+                assert!(record.is_delete());
+                assert!(record.cid.is_none());
+                assert!(record.record.is_none());
+            }
+            _ => panic!("Expected Record event"),
+        }
+    }
+
+    #[test]
+    fn test_parse_identity_event() {
+        let json = r#"{
+            "id": 12347,
+            "type": "identity",
+            "identity": {
+                "did": "did:plc:z72i7hdynmk6r22z27h6tvur",
+                "handle": "user.bsky.social",
+                "is_active": true,
+                "status": "active"
+            }
+        }"#;
+
+        let event: TapEvent = serde_json::from_str(json).expect("Failed to parse");
+
+        match event {
+            TapEvent::Identity { id, identity } => {
+                assert_eq!(id, 12347);
+                assert_eq!(&*identity.did, "did:plc:z72i7hdynmk6r22z27h6tvur");
+                assert_eq!(&*identity.handle, "user.bsky.social");
+                assert!(identity.is_active);
+                assert_eq!(identity.status, IdentityStatus::Active);
+            }
+            _ => panic!("Expected Identity event"),
+        }
+    }
+
+    #[test]
+    fn test_record_action_display() {
+        assert_eq!(RecordAction::Create.to_string(), "create");
+        assert_eq!(RecordAction::Update.to_string(), "update");
+        assert_eq!(RecordAction::Delete.to_string(), "delete");
+    }
+
+    #[test]
+    fn test_identity_status_display() {
+        assert_eq!(IdentityStatus::Active.to_string(), "active");
+        assert_eq!(IdentityStatus::Deactivated.to_string(), "deactivated");
+        assert_eq!(IdentityStatus::Suspended.to_string(), "suspended");
+        assert_eq!(IdentityStatus::Deleted.to_string(), "deleted");
+        assert_eq!(IdentityStatus::Takendown.to_string(), "takendown");
+    }
+
+    #[test]
+    fn test_at_uri() {
+        let record = RecordEvent {
+            live: true,
+            rev: "3lyileto4q52k".into(),
+            did: "did:plc:xyz".into(),
+            collection: "app.bsky.feed.post".into(),
+            rkey: "abc123".into(),
+            action: RecordAction::Create,
+            cid: None,
+            record: None,
+        };
+
+        assert_eq!(record.at_uri(), "at://did:plc:xyz/app.bsky.feed.post/abc123");
+    }
+
+    #[test]
+    fn test_event_id() {
+        let record_event = TapEvent::Record {
+            id: 100,
+            record: RecordEvent {
+                live: true,
+                rev: "rev".into(),
+                did: "did".into(),
+                collection: "col".into(),
+                rkey: "rkey".into(),
+                action: RecordAction::Create,
+                cid: None,
+                record: None,
+            },
+        };
+        assert_eq!(record_event.id(), 100);
+
+        let identity_event = TapEvent::Identity {
+            id: 200,
+            identity: IdentityEvent {
+                did: "did".into(),
+                handle: "handle".into(),
+                is_active: true,
+                status: IdentityStatus::Active,
+            },
+        };
+        assert_eq!(identity_event.id(), 200);
+    }
+}

crates/atproto-tap/src/lib.rs

···
+//! TAP (Trusted Attestation Protocol) service consumer for AT Protocol.
+//!
+//! This crate provides a client for consuming events from a TAP service,
+//! which delivers filtered, verified AT Protocol repository events.
+//!
+//! # Overview
+//!
+//! TAP is a single-tenant service that subscribes to an AT Protocol Relay and
+//! outputs filtered, verified events. Key features include:
+//!
+//! - **Verified Events**: MST integrity checks and signature verification
+//! - **Automatic Backfill**: Historical events delivered with `live: false`
+//! - **Repository Filtering**: Track specific DIDs or collections
+//! - **Acknowledgment Protocol**: At-least-once delivery semantics
+//!
+//! # Quick Start
+//!
+//! ```ignore
+//! use atproto_tap::{connect_to, TapEvent};
+//! use tokio_stream::StreamExt;
+//!
+//! #[tokio::main]
+//! async fn main() {
+//!     let mut stream = connect_to("localhost:2480");
+//!
+//!     while let Some(result) = stream.next().await {
+//!         match result {
+//!             Ok(event) => match event.as_ref() {
+//!                 TapEvent::Record { record, .. } => {
+//!                     println!("{} {} {}", record.action, record.collection, record.did);
+//!                 }
+//!                 TapEvent::Identity { identity, .. } => {
+//!                     println!("Identity: {} = {}", identity.did, identity.handle);
+//!                 }
+//!             },
+//!             Err(e) => eprintln!("Error: {}", e),
+//!         }
+//!     }
+//! }
+//! ```
+//!
+//! # Using with `tokio::select!`
+//!
+//! The stream integrates naturally with Tokio's select macro:
+//!
+//! ```ignore
+//! use atproto_tap::{connect, TapConfig};
+//! use tokio_stream::StreamExt;
+//! use tokio::signal;
+//!
+//! #[tokio::main]
+//! async fn main() {
+//!     let config = TapConfig::builder()
+//!         .hostname("localhost:2480")
+//!         .admin_password("secret")
+//!         .build();
+//!
+//!     let mut stream = connect(config);
+//!
+//!     loop {
+//!         tokio::select! {
+//!             Some(result) = stream.next() => {
+//!                 // Process event
+//!             }
+//!             _ = signal::ctrl_c() => {
+//!                 break;
+//!             }
+//!         }
+//!     }
+//! }
+//! ```
+//!
+//! # Management API
+//!
+//! Use [`TapClient`] to manage tracked repositories:
+//!
+//! ```ignore
+//! use atproto_tap::TapClient;
+//!
+//! let client = TapClient::new("localhost:2480", Some("password".to_string()));
+//!
+//! // Add repositories to track
+//! client.add_repos(&["did:plc:xyz123"]).await?;
+//!
+//! // Check service health
+//! if client.health().await? {
+//!     println!("TAP service is healthy");
+//! }
+//! ```
+//!
+//! # Memory Efficiency
+//!
+//! This crate is optimized for high-throughput event processing:
+//!
+//! - **Arc-wrapped events**: Events are shared via `Arc` for zero-cost sharing
+//! - **CompactString**: Small strings use inline storage (no heap allocation)
+//! - **Box<str>**: Immutable strings without capacity overhead
+//! - **RawValue**: Record payloads are lazily parsed on demand
+//! - **Pre-allocated buffers**: Ack messages avoid per-message allocations
+
+#![forbid(unsafe_code)]
+#![warn(missing_docs)]
+
+mod client;
+mod config;
+mod connection;
+mod errors;
+mod events;
+mod stream;
+
+// Re-export public types
+pub use atproto_identity::model::{Document, Service, VerificationMethod};
+pub use client::{RepoInfo, RepoState, TapClient};
+#[allow(deprecated)]
+pub use client::RepoStatus;
+pub use config::{TapConfig, TapConfigBuilder};
+pub use errors::TapError;
+pub use events::{IdentityEvent, IdentityStatus, RecordAction, RecordEvent, TapEvent};
+pub use stream::{TapStream, connect, connect_to};

crates/atproto-tap/src/stream.rs

···
+//! TAP event stream implementation.
+//!
+//! This module provides [`TapStream`], an async stream that yields TAP events
+//! with automatic connection management and reconnection handling.
+//!
+//! # Design
+//!
+//! The stream encapsulates all connection logic, allowing consumers to simply
+//! iterate over events using standard stream combinators or `tokio::select!`.
+//!
+//! Reconnection is handled automatically with exponential backoff. Parse errors
+//! are yielded as `Err` items but don't affect connection state - only connection
+//! errors trigger reconnection attempts.
+
+use crate::config::TapConfig;
+use crate::connection::TapConnection;
+use crate::errors::TapError;
+use crate::events::TapEvent;
+use futures::Stream;
+use std::pin::Pin;
+use std::sync::Arc;
+use std::task::{Context, Poll};
+use std::time::Duration;
+use tokio::sync::mpsc;
+
+/// An async stream of TAP events with automatic reconnection.
+///
+/// `TapStream` implements [`Stream`] and yields `Result<Arc<TapEvent>, TapError>`.
+/// Events are wrapped in `Arc` for efficient zero-cost sharing across consumers.
+///
+/// # Connection Management
+///
+/// The stream automatically:
+/// - Connects on first poll
+/// - Reconnects with exponential backoff on connection errors
+/// - Sends acknowledgments after parsing each message (if enabled)
+/// - Yields parse errors without affecting connection state
+///
+/// # Example
+///
+/// ```ignore
+/// use atproto_tap::{TapConfig, TapStream};
+/// use tokio_stream::StreamExt;
+///
+/// let config = TapConfig::builder()
+///     .hostname("localhost:2480")
+///     .build();
+///
+/// let mut stream = TapStream::new(config);
+///
+/// while let Some(result) = stream.next().await {
+///     match result {
+///         Ok(event) => println!("Event: {:?}", event),
+///         Err(e) => eprintln!("Error: {}", e),
+///     }
+/// }
+/// ```
+pub struct TapStream {
+    /// Receiver for events from the background task.
+    receiver: mpsc::Receiver<Result<Arc<TapEvent>, TapError>>,
+    /// Handle to request stream closure.
+    close_sender: Option<mpsc::Sender<()>>,
+    /// Whether the stream has been closed.
+    closed: bool,
+}
+
+impl TapStream {
+    /// Create a new TAP stream with the given configuration.
+    ///
+    /// The stream will start connecting immediately in a background task.
+    pub fn new(config: TapConfig) -> Self {
+        // Channel for events - buffer a few to handle bursts
+        let (event_tx, event_rx) = mpsc::channel(32);
+        // Channel for close signal
+        let (close_tx, close_rx) = mpsc::channel(1);
+
+        // Spawn background task to manage connection
+        tokio::spawn(connection_task(config, event_tx, close_rx));
+
+        Self {
+            receiver: event_rx,
+            close_sender: Some(close_tx),
+            closed: false,
+        }
+    }
+
+    /// Close the stream and release resources.
+    ///
+    /// After calling this, the stream will yield `None` on the next poll.
+    pub async fn close(&mut self) {
+        if let Some(sender) = self.close_sender.take() {
+            // Signal the background task to close
+            let _ = sender.send(()).await;
+        }
+        self.closed = true;
+    }
+
+    /// Returns true if the stream is closed.
+    pub fn is_closed(&self) -> bool {
+        self.closed
+    }
+}
+
+impl Stream for TapStream {
+    type Item = Result<Arc<TapEvent>, TapError>;
+
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        if self.closed {
+            return Poll::Ready(None);
+        }
+
+        self.receiver.poll_recv(cx)
+    }
+}
+
+impl Drop for TapStream {
+    fn drop(&mut self) {
+        // Drop the close_sender to signal the background task
+        self.close_sender.take();
+        tracing::debug!("TapStream dropped");
+    }
+}
+
+/// Background task that manages the WebSocket connection.
+async fn connection_task(
+    config: TapConfig,
+    event_tx: mpsc::Sender<Result<Arc<TapEvent>, TapError>>,
+    mut close_rx: mpsc::Receiver<()>,
+) {
+    let mut current_reconnect_delay = config.initial_reconnect_delay;
+    let mut attempt: u32 = 0;
+
+    loop {
+        // Check for close signal
+        if close_rx.try_recv().is_ok() {
+            tracing::debug!("Connection task received close signal");
+            break;
+        }
+
+        // Try to connect
+        tracing::debug!(attempt, hostname = %config.hostname, "Connecting to TAP service");
+        let conn_result = TapConnection::connect(&config).await;
+
+        match conn_result {
+            Ok(mut conn) => {
+                tracing::info!(hostname = %config.hostname, "TAP stream connected");
+                // Reset reconnection state on successful connect
+                current_reconnect_delay = config.initial_reconnect_delay;
+                attempt = 0;
+
+                // Event loop for this connection
+                loop {
+                    tokio::select! {
+                        biased;
+
+                        _ = close_rx.recv() => {
+                            tracing::debug!("Connection task received close signal during receive");
+                            let _ = conn.close().await;
+                            return;
+                        }
+
+                        recv_result = conn.recv() => {
+                            match recv_result {
+                                Ok(Some(msg)) => {
+                                    // Parse the message
+                                    match serde_json::from_str::<TapEvent>(&msg) {
+                                        Ok(event) => {
+                                            let event_id = event.id();
+
+                                            // Send ack if enabled (before sending event to channel)
+                                            if config.send_acks
+                                                && let Err(err) = conn.send_ack(event_id).await
+                                            {
+                                                tracing::warn!(error = %err, "Failed to send ack");
+                                                // Don't break connection for ack errors
+                                            }
+
+                                            // Send event to channel
+                                            let event = Arc::new(event);
+                                            if event_tx.send(Ok(event)).await.is_err() {
+                                                // Receiver dropped, exit task
+                                                tracing::debug!("Event receiver dropped, closing connection");
+                                                let _ = conn.close().await;
+                                                return;
+                                            }
+                                        }
+                                        Err(err) => {
+                                            // Parse errors don't affect connection
+                                            tracing::warn!(error = %err, "Failed to parse TAP message");
+                                            if event_tx.send(Err(TapError::ParseError(err.to_string()))).await.is_err() {
+                                                tracing::debug!("Event receiver dropped, closing connection");
+                                                let _ = conn.close().await;
+                                                return;
+                                            }
+                                        }
+                                    }
+                                }
+                                Ok(None) => {
+                                    // Connection closed by server
+                                    tracing::debug!("TAP connection closed by server");
+                                    break;
+                                }
+                                Err(err) => {
+                                    // Connection error
+                                    tracing::warn!(error = %err, "TAP connection error");
+                                    break;
+                                }
+                            }
+                        }
+                    }
+                }
+            }
+            Err(err) => {
+                tracing::warn!(error = %err, attempt, "Failed to connect to TAP service");
+            }
+        }
+
+        // Increment attempt counter
+        attempt += 1;
+
+        // Check if we've exceeded max attempts
+        if let Some(max) = config.max_reconnect_attempts
+            && attempt >= max
+        {
+            tracing::error!(attempts = attempt, "Max reconnection attempts exceeded");
+            let _ = event_tx
+                .send(Err(TapError::MaxReconnectAttemptsExceeded(attempt)))
+                .await;
+            break;
+        }
+
+        // Wait before reconnecting with exponential backoff
+        tracing::debug!(
+            delay_ms = current_reconnect_delay.as_millis(),
+            attempt,
+            "Waiting before reconnection"
+        );
+
+        tokio::select! {
+            _ = close_rx.recv() => {
+                tracing::debug!("Connection task received close signal during backoff");
+                return;
+            }
+            _ = tokio::time::sleep(current_reconnect_delay) => {
+                // Update delay for next attempt
+                current_reconnect_delay = Duration::from_secs_f64(
+                    (current_reconnect_delay.as_secs_f64() * config.reconnect_backoff_multiplier)
+                        .min(config.max_reconnect_delay.as_secs_f64()),
+                );
+            }
+        }
+    }
+
+    tracing::debug!("Connection task exiting");
+}
+
+/// Create a new TAP stream with the given configuration.
+pub fn connect(config: TapConfig) -> TapStream {
+    TapStream::new(config)
+}
+
+/// Create a new TAP stream connected to the given hostname.
+///
+/// Uses default configuration values.
+pub fn connect_to(hostname: &str) -> TapStream {
+    TapStream::new(TapConfig::new(hostname))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_stream_initial_state() {
+        // Note: This test doesn't actually poll the stream, just checks initial state
+        // Creating a TapStream requires a tokio runtime for the spawn
+    }
+
+    #[tokio::test]
+    async fn test_stream_close() {
+        let mut stream = TapStream::new(TapConfig::new("localhost:9999"));
+        assert!(!stream.is_closed());
+        stream.close().await;
+        assert!(stream.is_closed());
+    }
+
+    #[test]
+    fn test_connect_functions() {
+        // These just create configs, actual connection happens in background task
+        // We can't test without a runtime, so just verify the types compile
+        let _ = TapConfig::new("localhost:2480");
+    }
+
+    #[test]
+    fn test_reconnect_delay_calculation() {
+        // Test the delay calculation logic
+        let initial = Duration::from_secs(1);
+        let max = Duration::from_secs(10);
+        let multiplier = 2.0;
+
+        let mut delay = initial;
+        assert_eq!(delay, Duration::from_secs(1));
+
+        delay = Duration::from_secs_f64((delay.as_secs_f64() * multiplier).min(max.as_secs_f64()));
+        assert_eq!(delay, Duration::from_secs(2));
+
+        delay = Duration::from_secs_f64((delay.as_secs_f64() * multiplier).min(max.as_secs_f64()));
+        assert_eq!(delay, Duration::from_secs(4));
+
+        delay = Duration::from_secs_f64((delay.as_secs_f64() * multiplier).min(max.as_secs_f64()));
+        assert_eq!(delay, Duration::from_secs(8));
+
+        delay = Duration::from_secs_f64((delay.as_secs_f64() * multiplier).min(max.as_secs_f64()));
+        assert_eq!(delay, Duration::from_secs(10)); // Capped at max
+    }
+}