v0.13.0 doc update, refactor, http/tcp chain source and sink added

2026-07-17 05:58:44 -04:00
parent fa7f41c059
commit a48514a2eb
35 changed files with 2211 additions and 1292 deletions
@@ -1,13 +1,14 @@
 .idea
-data
+data/
-dev
+dev/
-log
+log/
-logs
+logs/
-cert
+cert/
-bin
+bin/
-script
+script/
-build
+build/
 *.log
 *.toml
 build.sh
 catalog.txt
 combined.txt
@@ -1,372 +0,0 @@
 ###############################################################################
 ### LogWisp Configuration
 ### Default location: ~/.config/logwisp/logwisp.toml
 ### Configuration Precedence: CLI flags > Environment > File > Defaults
 ### Default values shown - uncommented lines represent active configuration
 ###############################################################################
 ###############################################################################
 ### Global Settings
 ###############################################################################
 quiet = false                           # Enable quiet mode, suppress console output
 status_reporter = true                  # Enable periodic status logging
 auto_reload = false                     # Enable config auto-reload on file change
 ###############################################################################
 ### Logging Configuration (LogWisp's internal operational logging)
 ###############################################################################
 [logging]
 output = "stdout"                       # file|stdout|stderr|split|all|none
 level = "info"                          # debug|info|warn|error
 # [logging.file]
 # directory = "./log"                   # Log directory path
 # name = "logwisp"                      # Base filename
 # max_size_mb = 100                     # Rotation threshold
 # max_total_size_mb = 1000              # Total size limit
 # retention_hours = 168.0               # Delete logs older than (7 days)
 [logging.console]
 target = "stdout"                       # stdout|stderr|split
 format = "txt"                          # txt|json
 ###############################################################################
 ### Pipeline Configuration
 ### Each pipeline: sources -> rate_limit -> filters -> format -> sinks
 ###############################################################################
 [[pipelines]]
 name = "default"                        # Pipeline identifier
 ###============================================================================
 ### Rate Limiting (Pipeline-level)
 ###============================================================================
 # [pipelines.rate_limit]
 # rate = 1000.0                         # Entries per second (0=disabled)
 # burst = 2000.0                        # Burst capacity (defaults to rate)
 # policy = "drop"                       # pass|drop
 # max_entry_size_bytes = 0              # Max entry size (0=unlimited)
 ###============================================================================
 ### Filters (Sequential pattern matching)
 ###============================================================================
 ### ⚠️ Example: Include only ERROR and WARN logs
 ## [[pipelines.filters]]
 ## type = "include"                     # include|exclude
 ## logic = "or"                         # or|and
 ## patterns = [".*ERROR.*", ".*WARN.*"]
 ### ⚠️ Example: Exclude debug logs
 ## [[pipelines.filters]]
 ## type = "exclude"
 ## patterns = [".*DEBUG.*"]
 ###============================================================================
 ### Format (Log transformation)
 ###============================================================================
 # [pipelines.format]
 # type = "raw"                          # raw|json|txt
 ## JSON formatting
 # [pipelines.format.json]
 # pretty = false                        # Pretty-print JSON
 # timestamp_field = "timestamp"         # Field name for timestamp
 # level_field = "level"                 # Field name for log level
 # message_field = "message"             # Field name for message
 # source_field = "source"               # Field name for source
 ## Text templating
 # [pipelines.format.txt]
 # template = "{{.Timestamp | FmtTime}} [{{.Level}}] {{.Message}}"
 # timestamp_format = "2006-01-02 15:04:05"
 ## Raw templating
 # [pipelines.format.raw]
 # add_new_line = true                   # Preserve new line delimiter between log entries
 ###============================================================================
 ### SOURCES (Inputs)
 ### Architecture: Pipeline can have multiple sources
 ###============================================================================
 ###----------------------------------------------------------------------------
 ### File Source (File monitoring)
 [[pipelines.sources]]
 type = "file"
 [pipelines.sources.file]
 directory = "./"                        # Directory to monitor
 pattern = "*.log"                       # Glob pattern
 check_interval_ms = 100                 # File check interval
 recursive = false                       # Recursive monitoring (TODO)
 ###----------------------------------------------------------------------------
 ### Console Source
 # [[pipelines.sources]]
 # type = "console"
 # [pipelines.sources.console]
 # buffer_size = 1000
 ###----------------------------------------------------------------------------
 ### HTTP Source (Server mode - receives logs via HTTP POST)
 # [[pipelines.sources]]
 # type = "http"
 # [pipelines.sources.http]
 # host = "0.0.0.0"                      # Listen interface
 # port = 8081                           # Listen port
 # ingest_path = "/ingest"               # Ingestion endpoint
 # buffer_size = 1000
 # max_body_size = 1048576               # 1MB
 # read_timeout_ms = 10000
 # write_timeout_ms = 10000
 ### Network access control
 # [pipelines.sources.http.acl]
 # enabled = false
 # max_connections_per_ip = 10           # Max simultaneous connections from a single IP
 # max_connections_total = 100           # Max simultaneous connections for this component
 # requests_per_second = 100.0           # Per-IP request rate limit
 # burst_size = 200                      # Per-IP request burst limit
 # response_message = "Rate limit exceeded"
 # response_code = 429
 # ip_whitelist = ["192.168.1.0/24"]
 # ip_blacklist = ["10.0.0.100"]
 ### TLS configuration (mTLS support)
 # [pipelines.sources.http.tls]
 # enabled = false
 # cert_file = "/path/to/server.pem"     # Server certificate
 # key_file = "/path/to/server.key"      # Server private key
 # client_auth = false                   # Enable mTLS
 # client_ca_file = "/path/to/ca.pem"    # CA for client verification
 # verify_client_cert = true             # Verify client certificates
 # min_version = "TLS1.2"                # TLS1.0|TLS1.1|TLS1.2|TLS1.3
 # max_version = "TLS1.3"
 # cipher_suites = ""                    # Comma-separated cipher list
 ###----------------------------------------------------------------------------
 ### TCP Source (Server mode - receives logs via TCP)
 # [[pipelines.sources]]
 # type = "tcp"
 # [pipelines.sources.tcp]
 # host = "0.0.0.0"
 # port = 9091
 # buffer_size = 1000
 # read_timeout_ms = 10000
 # keep_alive = true
 # keep_alive_period_ms = 30000
 ### Network access control
 # [pipelines.sources.tcp.acl]
 # enabled = false
 # max_connections_per_ip = 10           # Max simultaneous connections from a single IP
 # max_connections_total = 100           # Max simultaneous connections for this component
 # requests_per_second = 100.0           # Per-IP request rate limit
 # burst_size = 200                      # Per-IP request burst limit
 # response_message = "Rate limit exceeded"
 # response_code = 429
 # ip_whitelist = ["192.168.1.0/24"]
 # ip_blacklist = ["10.0.0.100"]
 ### ⚠️ IMPORTANT: TCP does NOT support TLS/mTLS (gnet limitation)
 ### Use HTTP Source with TLS for encrypted transport
 ###============================================================================
 ### SINKS (Outputs)
 ### Architecture: Pipeline can have multiple sinks (fan-out)
 ###============================================================================
 ###----------------------------------------------------------------------------
 ### Console Sink
 # [[pipelines.sinks]]
 # type = "console"
 # [pipelines.sinks.console]
 # target = "stdout"                     # stdout|stderr|split
 # colorize = false                      # Colorized output
 # buffer_size = 100
 ###----------------------------------------------------------------------------
 ### File Sink (Rotating logs)
 # [[pipelines.sinks]]
 # type = "file"
 # [pipelines.sinks.file]
 # directory = "./logs"
 # name = "output"
 # max_size_mb = 100
 # max_total_size_mb = 1000
 # min_disk_free_mb = 100
 # retention_hours = 168.0               # 7 days
 # buffer_size = 1000
 # flush_interval_ms = 1000
 ###----------------------------------------------------------------------------
 ### HTTP Sink (Server mode - SSE streaming for clients)
 [[pipelines.sinks]]
 type = "http"
 [pipelines.sinks.http]
 host = "0.0.0.0"
 port = 8080
 stream_path = "/stream"                 # SSE streaming endpoint
 status_path = "/status"                 # Status endpoint
 buffer_size = 1000
 write_timeout_ms = 10000
 ### Heartbeat configuration (keep connections alive)
 [pipelines.sinks.http.heartbeat]
 enabled = true
 interval_ms = 30000                     # 30 seconds
 include_timestamp = true
 include_stats = false
 format = "comment"                      # comment|event|json
 ### Network access control
 # [pipelines.sinks.http.acl]
 # enabled = false
 # max_connections_per_ip = 10           # Max simultaneous connections from a single IP
 # max_connections_total = 100           # Max simultaneous connections for this component
 # requests_per_second = 100.0           # Per-IP request rate limit
 # burst_size = 200                      # Per-IP request burst limit
 # response_message = "Rate limit exceeded"
 # response_code = 429
 # ip_whitelist = ["192.168.1.0/24"]
 # ip_blacklist = ["10.0.0.100"]
 ### TLS configuration (mTLS support)
 # [pipelines.sinks.http.tls]
 # enabled = false
 # cert_file = "/path/to/server.pem"     # Server certificate
 # key_file = "/path/to/server.key"      # Server private key
 # client_auth = false                   # Enable mTLS
 # client_ca_file = "/path/to/ca.pem"    # CA for client verification
 # verify_client_cert = true             # Verify client certificates
 # min_version = "TLS1.2"                # TLS1.0|TLS1.1|TLS1.2|TLS1.3
 # max_version = "TLS1.3"
 # cipher_suites = ""                    # Comma-separated cipher list
 ###----------------------------------------------------------------------------
 ### TCP Sink (Server mode - TCP streaming for clients)
 # [[pipelines.sinks]]
 # type = "tcp"
 # [pipelines.sinks.tcp]
 # host = "0.0.0.0"
 # port = 9090
 # buffer_size = 1000
 # write_timeout_ms = 10000
 # keep_alive = true
 # keep_alive_period_ms = 30000
 ### Heartbeat configuration
 # [pipelines.sinks.tcp.heartbeat]
 # enabled = false
 # interval_ms = 30000
 # include_timestamp = true
 # include_stats = false
 # format = "json"                       # json|txt
 ### Network access control
 # [pipelines.sinks.tcp.acl]
 # enabled = false
 # max_connections_per_ip = 10           # Max simultaneous connections from a single IP
 # max_connections_total = 100           # Max simultaneous connections for this component
 # requests_per_second = 100.0           # Per-IP request rate limit
 # burst_size = 200                      # Per-IP request burst limit
 # response_message = "Rate limit exceeded"
 # response_code = 429
 # ip_whitelist = ["192.168.1.0/24"]
 # ip_blacklist = ["10.0.0.100"]
 ### ⚠️ IMPORTANT: TCP does NOT support TLS/mTLS (gnet limitation)
 ### Use HTTP Sink with TLS for encrypted transport
 ###----------------------------------------------------------------------------
 ### HTTP Client Sink (Forward to remote HTTP endpoint)
 # [[pipelines.sinks]]
 # type = "http_client"
 # [pipelines.sinks.http_client]
 # url = "https://logs.example.com/ingest"
 # buffer_size = 1000
 # batch_size = 100                      # Entries per batch
 # batch_delay_ms = 1000                 # Max wait before sending
 # timeout_seconds = 30
 # max_retries = 3
 # retry_delay_ms = 1000
 # retry_backoff = 2.0                   # Exponential backoff multiplier
 # insecure_skip_verify = false          # Skip TLS verification
 ### TLS configuration for client
 # [pipelines.sinks.http_client.tls]
 # enabled = false                       # Enable TLS for the outgoing connection
 # server_ca_file = "/path/to/ca.pem"    # CA for verifying the remote server's certificate
 # server_name = "logs.example.com"      # For server certificate validation (SNI)
 # insecure_skip_verify = false          # Skip server verification, use with caution
 # client_cert_file = "/path/to/client.pem"  # Client's certificate to present to the server for mTLS
 # client_key_file = "/path/to/client.key"   # Client's private key for mTLS
 # min_version = "TLS1.2"
 # max_version = "TLS1.3"
 # cipher_suites = ""
 ### ⚠️ Example: HTTP Client Sink → HTTP Source with mTLS
 ## HTTP Source with mTLS:
 ## [pipelines.sources.http.tls]
 ## enabled = true
 ## cert_file = "/path/to/server.pem"
 ## key_file = "/path/to/server.key"
 ## client_auth = true                   # Enable client cert verification
 ## client_ca_file = "/path/to/ca.pem"
 ## verify_client_cert = true
 ## HTTP Client with client cert:
 ## [pipelines.sinks.http_client.tls]
 ## enabled = true
 ## server_ca_file = "/path/to/ca.pem"   # Verify server
 ## client_cert_file = "/path/to/client.pem"  # Client certificate
 ## client_key_file = "/path/to/client.key"
 ###----------------------------------------------------------------------------
 ### TCP Client Sink (Forward to remote TCP endpoint)
 # [[pipelines.sinks]]
 # type = "tcp_client"
 # [pipelines.sinks.tcp_client]
 # host = "logs.example.com"
 # port = 9090
 # buffer_size = 1000
 # dial_timeout_seconds = 10            # Connection timeout
 # write_timeout_seconds = 30           # Write timeout
 # read_timeout_seconds = 10            # Read timeout
 # keep_alive_seconds = 30              # TCP keep-alive
 # reconnect_delay_ms = 1000            # Initial reconnect delay
 # max_reconnect_delay_ms = 30000       # Max reconnect delay
 # reconnect_backoff = 1.5              # Exponential backoff
 ### ⚠️ WARNING: TCP Client has NO TLS support
 ### Use HTTP Client with TLS for encrypted transport
 ###############################################################################
 ### Common Usage Patterns
 ###############################################################################
 ### Pattern 1: Log Aggregation (Client → Server)
 ### - HTTP Client Sink → HTTP Source (with optional TLS/mTLS)
 ### - TCP Client Sink → TCP Source (unencrypted only)
 ### Pattern 2: Live Monitoring
 ### - HTTP Sink: Browser-based SSE streaming (https://host:8080/stream)
 ### - TCP Sink: Debug interface (telnet/netcat to port 9090)
 ### Pattern 3: Log Collection & Distribution
 ### - File Source → Multiple Sinks (fan-out)
 ### - Multiple Sources → Single Pipeline → Multiple Sinks
@@ -6,24 +6,20 @@ A pipeline-based log transport and processing system built in Go. LogWisp provid
 ### Core Capabilities
 - **Pipeline Architecture**: Independent processing pipelines with source(s) → filter → format → sink(s) flow
- **Multiple Input Sources**: Directory monitoring, stdin, HTTP, TCP
+- **Multiple Input Sources**: File monitoring, console (stdin), random log generation, null
- **Flexible Output Sinks**: Console, file, HTTP SSE, TCP streaming, HTTP/TCP forwarding
+- **Flexible Output Sinks**: Console, file, HTTP SSE, TCP streaming, null
 - **Real-time Processing**: Sub-millisecond latency with configurable buffering
 - **Hot Configuration Reload**: Update pipelines without service restart
 - **Session Management**: Built-in session tracking for multiple client connections
 ### Data Processing
 - **Pattern-based Filtering**: Chainable include/exclude filters with regex support
- **Multiple Formatters**: Raw, JSON, and template-based text formatting
+- **Multiple Formatters**: Raw, JSON, and text formatting with integrated sanitizer policies
 - **Rate Limiting**: Pipeline rate controls
 - **Heartbeat Generation**: Flow-level heartbeat events for keep-alives
 ### Security & Reliability  
 - **Authentication**: mTLS support
 - **Access Control**: IP whitelisting/blacklisting, connection limits
 - **TLS Encryption**: Full TLS 1.2/1.3 support for HTTP connections
 - **Automatic Reconnection**: Resilient client connections with exponential backoff
 - **File Rotation**: Size-based rotation with retention policies
 ### Operational Features
 - **Status Monitoring**: Real-time statistics and health endpoints
 - **Signal Handling**: Graceful shutdown and configuration reload via signals
 - **Background Mode**: Daemon operation with proper signal handling
@@ -38,8 +34,7 @@ A pipeline-based log transport and processing system built in Go. LogWisp provid
 - [Output Sinks](sinks.md) - Sink types and output options
 - [Filters](filters.md) - Pattern-based log filtering
 - [Formatters](formatters.md) - Log formatting and transformation
- [Security](security.md) - IP-based access control configuration and mTLS
+- [Networking & Security](networking.md) - Network features (Note: TLS and Auth are currently placeholders in the new architecture)
 - [Networking](networking.md) - TLS, rate limiting, and network features
 - [Command Line Interface](cli.md) - CLI flags and subcommands
 - [Operations Guide](operations.md) - Running and maintaining LogWisp
@@ -51,15 +46,17 @@ Install LogWisp and create a basic configuration:
 [[pipelines]]
 name = "default"
-[[pipelines.sources]]
+[[pipelines.plugin_sources]]
-type = "directory"
+id = "default_source"
-[pipelines.sources.directory]
+type = "file"
-path = "./"
+[pipelines.plugin_sources.config]
 directory = "./"
 pattern = "*.log"
-[[pipelines.sinks]]
+[[pipelines.plugin_sinks]]
 id = "default_sink"
 type = "console"
-[pipelines.sinks.console]
+[pipelines.plugin_sinks.config]
 target = "stdout"
 ```
@@ -13,13 +13,15 @@ Each pipeline operates independently with a source → filter → format → sin
 ```
 Service (Main Process)
 ├── Pipeline 1
-│   ├── Sources (1 or more)
+│   ├── Plugin Sources (1 or more)
-│   ├── Rate Limiter (optional)
+│   ├── Flow
-│   ├── Filter Chain (optional)
+│   │   ├── Heartbeat Generator (optional)
-│   ├── Formatter (optional)
+│   │   ├── Rate Limiter (optional)
-│   └── Sinks (1 or more)
+│   │   ├── Filter Chain (optional)
 │   │   └── Formatter (optional)
 │   └── Plugin Sinks (1 or more)
 ├── Pipeline 2
-│   └── [Same structure]
+│   └── [Similar structure]
 └── Status Reporter (optional)
 ```
@@ -27,11 +29,11 @@ Service (Main Process)
 ### Processing Stages
-1. **Source Stage**: Sources monitor inputs and generate log entries
+1. **Source Stage**: Plugin sources monitor inputs and generate log entries
-2. **Rate Limiting**: Optional pipeline-level rate control
+2. **Flow - Rate Limiting**: Optional pipeline-level rate control
-3. **Filtering**: Pattern-based inclusion/exclusion
+3. **Flow - Filtering**: Pattern-based inclusion/exclusion
-4. **Formatting**: Transform entries to desired output format
+4. **Flow - Formatting**: Transform entries to desired output format with sanitization
-5. **Distribution**: Fan-out to multiple sinks
+5. **Distribution**: Fan-out to multiple plugin sinks
 ### Entry Lifecycle
@@ -50,14 +52,16 @@ Each component maintains internal buffers to handle burst traffic:
 - Sinks: Independent buffers per sink
 - Network components: Additional TCP/HTTP buffers
 *"Sink dispatch uses non-blocking sends. When a sink's input buffer is full, the event is dropped for that sink only and counted in pipeline statistics (`total_dropped_by_sink`). The policy is uniform and not configurable; a slow sink does not stall the pipeline or other sinks."*
 ## Component Types
 ### Sources (Input)
- **Directory Source**: File system monitoring with rotation detection
+- **File Source**: File system directory monitoring with rotation detection
- **Stdin Source**: Standard input processing
+- **Console Source**: Standard input processing (stdin)
- **HTTP Source**: REST endpoint for log ingestion
+- **Random Source**: Generates random log entries for testing
- **TCP Source**: Raw TCP socket listener
+- **Null Source**: Discards logs, used for testing
 ### Sinks (Output)
@@ -65,14 +69,13 @@ Each component maintains internal buffers to handle burst traffic:
 - **File Sink**: Rotating file writer
 - **HTTP Sink**: Server-Sent Events (SSE) streaming
 - **TCP Sink**: TCP server for client connections
- **HTTP Client Sink**: Forward to remote HTTP endpoints
+- **Null Sink**: Discards all received events
 - **TCP Client Sink**: Forward to remote TCP servers
 ### Processing Components
 - **Rate Limiter**: Token bucket algorithm for flow control
 - **Filter Chain**: Sequential pattern matching
- **Formatters**: Raw, JSON, or template-based text transformation
+- **Formatters**: Raw, JSON, or text transformation with sanitizer policies
 ## Concurrency Model
@@ -95,8 +98,7 @@ Each component maintains internal buffers to handle burst traffic:
 ### Connection Patterns
 **Chaining Design**:
- TCP Client Sink → TCP Source: Direct TCP forwarding
+- Future plan
 - HTTP Client Sink → HTTP Source: HTTP-based forwarding
 **Monitoring Design**:
 - TCP Sink: Debugging interface
@@ -106,8 +108,6 @@ Each component maintains internal buffers to handle burst traffic:
 - HTTP/1.1 and HTTP/2 for HTTP connections
 - Raw TCP connections
 - TLS 1.2/1.3 for HTTPS connections (HTTP only)
 - Server-Sent Events for real-time streaming
 ## Resource Management
@@ -125,6 +125,10 @@ Each component maintains internal buffers to handle burst traffic:
 ### Connection Management
 - HTTP sink silenty drops IPv6 connections (deliberate IPv4-only enforcement)
 *Note:* Placeholder; below features are removed in restructuring and will be added in the future release.
 - Per-IP connection limits
 - Global connection caps
 - Automatic reconnection with exponential backoff
@@ -136,7 +140,6 @@ Each component maintains internal buffers to handle burst traffic:
 - Panic recovery in pipeline processing
 - Independent pipeline operation
 - Automatic source restart on failure
 - Sink failure isolation
 ### Data Integrity
@@ -15,30 +15,8 @@ logwisp [options]
 | Command | Description |
 |---------|-------------|
-| `tls` | Generate TLS certificates |
+| `--version` | Display version information |
-| `version` | Display version information |
+| `--help` | Show help information |
 | `help` | Show help information |
 ### tls Command
 Generate TLS certificates.
 ```bash
 logwisp tls [options]
 ```
 **Options:**
 | Flag | Description | Default |
 |------|-------------|---------|
 | `-ca` | Generate CA certificate | - |
 | `-server` | Generate server certificate | - |
 | `-client` | Generate client certificate | - |
 | `-host` | Comma-separated hosts/IPs | localhost |
 | `-o` | Output file prefix | Required |
 | `-ca-cert` | CA certificate file | Required for server/client |
 | `-ca-key` | CA key file | Required for server/client |
 | `-days` | Certificate validity days | 365 |
 ### version Command
@@ -63,10 +41,9 @@ Output includes:
 | Flag | Description | Default |
 |------|-------------|---------|
 | `-c, --config` | Configuration file path | `./logwisp.toml` |
 | `-b, --background` | Run as daemon | false |
 | `-q, --quiet` | Suppress console output | false |
-| `--disable-status-reporter` | Disable status logging | false |
+| `--status-reporter` | Status logging | true |
-| `--config-auto-reload` | Enable config hot reload | false |
+| `--auto-reload` | Enable config hot reload | false |
 ### Logging Options
@@ -91,9 +68,9 @@ Configure pipelines via CLI (N = array index, 0-based).
 | Flag | Description |
 |------|-------------|
 | `--pipelines.N.name` | Pipeline name |
-| `--pipelines.N.sources.N.type` | Source type |
+| `--pipelines.N.plugin_sources.N.type` | Source type |
-| `--pipelines.N.filters.N.type` | Filter type |
+| `--pipelines.N.flow.filters.N.type` | Filter type |
-| `--pipelines.N.sinks.N.type` | Sink type |
+| `--pipelines.N.plugin_sinks.N.type` | Sink type |
 ## Flag Formats
@@ -102,7 +79,7 @@ Configure pipelines via CLI (N = array index, 0-based).
 ```bash
 logwisp --quiet
 logwisp --quiet=true
-logwisp --quiet=false
+logwisp --pipelines.0.plugin_sources.0.type=console
 ```
 ### String Flags
@@ -117,13 +94,13 @@ logwisp -c config.toml
 ```bash
 logwisp --logging.level=debug
 logwisp --pipelines.0.name=myapp
-logwisp --pipelines.0.sources.0.type=stdin
+logwisp --pipelines.0.sources.0.type=console
 ```
 ### Array Values (JSON)
 ```bash
-logwisp --pipelines.0.filters.0.patterns='["ERROR","WARN"]'
+logwisp --pipelines.0.flow.filters.0.patterns='["ERROR","WARN"]'
 ```
 ## Environment Variables
@@ -171,7 +148,7 @@ export LOGWISP_PIPELINES_0_NAME=myapp
 logwisp --logging.output=stderr --logging.level=debug
 # Quick test with stdin
-logwisp --pipelines.0.sources.0.type=stdin --pipelines.0.sinks.0.type=console
+logwisp --pipelines.0.plugin_sources.0.type=console --pipelines.0.plugin_sinks.0.type=console
 ```
 ### Production Deployment
@@ -194,19 +171,6 @@ logwisp --config test.toml --logging.level=debug --disable-status-reporter
 logwisp --config test.toml --quiet
 ```
 ### Quick Commands
 ```bash
 # Generate admin password
 logwisp auth -u admin -b
 # Create self-signed certs
 logwisp tls -server -host localhost -o server
 # Check version
 logwisp version
 ```
 ## Help System
 ### General Help
@@ -217,14 +181,6 @@ logwisp -h
 logwisp help
 ```
 ### Command Help
 ```bash
 logwisp auth --help
 logwisp tls --help
 logwisp help auth
 ```
 ## Special Flags
 ### Internal Flags
@@ -235,6 +191,6 @@ These flags are for internal use:
 ### Hidden Behaviors
- SIGHUP ignored by default (nohup behavior)
+- SIGHUP ignored ignored during startup (after startup triggers config reload)
 - Automatic panic recovery in pipelines
 - Resource cleanup on shutdown
@@ -24,10 +24,9 @@ Top-level configuration options:
 | Setting | Type | Default | Description |
 |---------|------|---------|-------------|
 | `background` | bool | false | Run as daemon process |
 | `quiet` | bool | false | Suppress console output |
-| `disable_status_reporter` | bool | false | Disable periodic status logging |
+| `status_reporter` | bool | true | Periodic status logging |
-| `config_auto_reload` | bool | false | Enable file watch for auto-reload |
+| `auto_reload` | bool | false | Enable file watch for auto-reload |
 ## Logging Configuration
@@ -47,7 +46,6 @@ retention_hours = 168.0
 [logging.console]
 target = "stdout"  # stdout|stderr|split
 format = "txt"     # txt|json
 ```
 ### Output Modes
@@ -68,30 +66,34 @@ Each `[[pipelines]]` section defines an independent processing pipeline:
 name = "pipeline-name"
 # Rate limiting (optional)
-[pipelines.rate_limit]
+[pipelines.flow.rate_limit]
 rate = 1000.0
 burst = 2000.0
 policy = "drop"  # pass|drop
 max_entry_size_bytes = 0  # 0=unlimited
 # Format configuration (optional)
-[pipelines.format]
+[pipelines.flow.format]
 type = "json"  # raw|json|txt
 sanitizer_policy = "json"
-# Sources (required, 1+)
+[[pipelines.plugin_sources]]
-[[pipelines.sources]]
+id = "my_source"
-type = "directory"
+type = "file"
 [pipelines.plugin_sources.config]
 # ... source-specific config
 # Filters (optional)
-[[pipelines.filters]]
+[[pipelines.flow.filters]]
 type = "include"
 logic = "or"
 patterns = ["ERROR", "WARN"]
 # Sinks (required, 1+)
-[[pipelines.sinks]]
+[[pipelines.plugin_sinks]]
 id = "my_sink"
 type = "http"
 [pipelines.plugin_sinks.config]
 # ... sink-specific config
 ```
@@ -113,7 +115,7 @@ All configuration options support environment variable overrides:
 | `quiet` | `LOGWISP_QUIET` |
 | `logging.level` | `LOGWISP_LOGGING_LEVEL` |
 | `pipelines[0].name` | `LOGWISP_PIPELINES_0_NAME` |
-| `pipelines[0].sources[0].type` | `LOGWISP_PIPELINES_0_SOURCES_0_TYPE` |
+| `pipelines[0].plugin_sources[0].type` | `LOGWISP_PIPELINES_0_PLUGIN_SOURCES_0_TYPE` |
 ## Command-Line Overrides
@@ -123,13 +125,15 @@ All configuration options can be overridden via CLI flags:
 logwisp --quiet \
  --logging.level=debug \
  --pipelines.0.name=myapp \
-  --pipelines.0.sources.0.type=stdin
+  --pipelines.0.plugin_sources.0.type=console
 ```
 ## Configuration Validation
 LogWisp validates configuration at startup:
- Required fields presence
+- Rpipelines non-empty, name non-empty, ≥1 source, ≥1 sink, logging enum values.equired fields presence
 Partial check in plugin constructor:
 - Type correctness
 - Port conflicts
 - Path accessibility
@@ -141,12 +145,12 @@ LogWisp validates configuration at startup:
 Enable configuration hot reload:
 ```toml
-config_auto_reload = true
+auto_reload = true
 ```
 Or via command line:
 ```bash
-logwisp --config-auto-reload
+logwisp --auto-reload
 ```
 Reload triggers:
@@ -161,7 +165,6 @@ Reloadable items:
 Non-reloadable (requires restart):
 - Logging configuration
 - Background mode
 - Global settings
 ## Default Configuration
@@ -172,15 +175,17 @@ Minimal working configuration:
 [[pipelines]]
 name = "default"
-[[pipelines.sources]]
+[[pipelines.plugin_sources]]
-type = "directory"
+id = "default_source"
-[pipelines.sources.directory]
+type = "file"
-path = "./"
+[pipelines.plugin_sources.config]
 directory = "./"
 pattern = "*.log"
-[[pipelines.sinks]]
+[[pipelines.plugin_sinks]]
 id = "default_sink"
 type = "console"
-[pipelines.sinks.console]
+[pipelines.plugin_sinks.config]
 target = "stdout"
 ```
@@ -9,7 +9,7 @@ LogWisp filters control which log entries pass through the pipeline using patter
 Only entries matching patterns pass through.
 ```toml
-[[pipelines.filters]]
+[[pipelines.flow.filters]]
 type = "include"
 logic = "or"  # or|and
 patterns = [
@@ -24,7 +24,7 @@ patterns = [
 Entries matching patterns are dropped.
 ```toml
-[[pipelines.filters]]
+[[pipelines.flow.filters]]
 type = "exclude"
 patterns = [
    "DEBUG",
@@ -89,12 +89,12 @@ Multiple filters execute sequentially:
 ```toml
 # First filter: Include errors and warnings
-[[pipelines.filters]]
+[[pipelines.flow.filters]]
 type = "include"
 patterns = ["ERROR", "WARN"]
 # Second filter: Exclude test environments
-[[pipelines.filters]]
+[[pipelines.flow.filters]]
 type = "exclude"
 patterns = ["test-env", "staging"]
 ```
@@ -137,14 +137,14 @@ patterns = ["ERROR", "WARN", "FATAL", "CRITICAL"]
 ### Application Filtering
 ```toml
-[[pipelines.filters]]
+[[pipelines.flow.filters]]
 type = "include"
 patterns = ["app1", "app2", "app3"]
 ```
 ### Noise Reduction
 ```toml
-[[pipelines.filters]]
+[[pipelines.flow.filters]]
 type = "exclude"
 patterns = [
    "health-check",
@@ -156,7 +156,7 @@ patterns = [
 ### Security Filtering
 ```toml
-[[pipelines.filters]]
+[[pipelines.flow.filters]]
 type = "exclude"
 patterns = [
    "password",
@@ -169,17 +169,17 @@ patterns = [
 ### Multi-stage Filtering
 ```toml
 # Include production logs
-[[pipelines.filters]]
+[[pipelines.flow.filters]]
 type = "include"
 patterns = ["prod-", "production"]
 # Include only errors
-[[pipelines.filters]]
+[[pipelines.flow.filters]]
 type = "include"
 patterns = ["ERROR", "EXCEPTION", "FATAL"]
 # Exclude known issues
-[[pipelines.filters]]
+[[pipelines.flow.filters]]
 type = "exclude"
 patterns = ["ECONNRESET", "broken pipe"]
 ```
@@ -9,11 +9,10 @@ LogWisp formatters transform log entries before output to sinks.
 Outputs the log message as-is with optional newline.
 ```toml
-[pipelines.format]
+[pipelines.flow.format]
 type = "raw"
-
+sanitizer_policy = "raw"
-[pipelines.format.raw]
+flags = 1
 add_new_line = true
 ```
 **Configuration Options:**
@@ -21,6 +20,9 @@ add_new_line = true
 | Option | Type | Default | Description |
 |--------|------|---------|-------------|
 | `add_new_line` | bool | true | Append newline to messages |
 | `type` | string | "raw" | raw, json, or txt |
 | `flags` | int64 | 0 | log/formatter flags override |
 | `sanitizer_policy` | string | | Sanitizer policy (e.g. "json", "raw", "txt", "shell") |
 ### JSON Formatter
@@ -30,24 +32,11 @@ Produces structured JSON output.
 [pipelines.format]
 type = "json"
-[pipelines.format.json]
+[pipelines.flow.format]
-pretty = false
+ type = "json"
-timestamp_field = "timestamp"
+sanitizer_policy = "json"
 level_field = "level"
 message_field = "message"
 source_field = "source"
 ```
 **Configuration Options:**
 | Option | Type | Default | Description |
 |--------|------|---------|-------------|
 | `pretty` | bool | false | Pretty print JSON |
 | `timestamp_field` | string | "timestamp" | Field name for timestamp |
 | `level_field` | string | "level" | Field name for log level |
 | `message_field` | string | "message" | Field name for message |
 | `source_field` | string | "source" | Field name for source |
 **Output Structure:**
 ```json
 {
@@ -63,11 +52,9 @@ source_field = "source"
 Template-based text formatting.
 ```toml
-[pipelines.format]
+[pipelines.flow.format]
 type = "txt"
-
+sanitizer_policy = "txt"
 [pipelines.format.txt]
 template = "[{{.Timestamp | FmtTime}}] [{{.Level | ToUpper}}] {{.Source}} - {{.Message}}"
 timestamp_format = "2006-01-02T15:04:05.000Z07:00"
 ```
@@ -75,8 +62,7 @@ timestamp_format = "2006-01-02T15:04:05.000Z07:00"
 | Option | Type | Default | Description |
 |--------|------|---------|-------------|
-| `template` | string | See below | Go template string |
+| `timestamp_format` | string | "" | Time format override |
 | `timestamp_format` | string | RFC3339 | Go time format string |
 **Default Template:**
 ```
@@ -134,12 +120,12 @@ Each pipeline can have its own formatter:
 ```toml
 [[pipelines]]
 name = "json-pipeline"
-[pipelines.format]
+[pipelines.flow.format]
 type = "json"
 [[pipelines]]
 name = "text-pipeline"
-[pipelines.format]
+[pipelines.flow.format]
 type = "txt"
 ```
@@ -182,34 +168,13 @@ Relative performance (fastest to slowest):
 ### Structured Logging
 ```toml
-[pipelines.format]
+[pipelines.flow.format]
 type = "json"
 [pipelines.format.json]
 pretty = false
 ```
 ### Human-Readable Logs
 ```toml
-[pipelines.format]
+[pipelines.flow.format]
 type = "txt"
 [pipelines.format.txt]
 template = "{{.Timestamp | FmtTime}} [{{.Level}}] {{.Message}}"
 timestamp_format = "15:04:05"
 ```
 ### Syslog Format
 ```toml
 [pipelines.format]
 type = "txt"
 [pipelines.format.txt]
 template = "{{.Timestamp | FmtTime}} {{.Source}} {{.Level}}: {{.Message}}"
 timestamp_format = "Jan 2 15:04:05"
 ```
 ### Minimal Output
 ```toml
 [pipelines.format]
 type = "txt"
 [pipelines.format.txt]
 template = "{{.Message}}"
 ```
@@ -1,173 +1,40 @@
 # Networking
-Network configuration for LogWisp connections, including TLS, rate limiting, and access control.
+*Note: Under redesign*
 ## TLS Configuration
-### TLS Support Matrix
+*Note: As of the latest architecture updates, network TLS, mTLS, and Rate Limiting features are undergoing a redesign and are currently acting as placeholders. The documentation below details the structure for future updates.*
 | Component | TLS Support | Notes |
 |-----------|-------------|-------|
 | HTTP Source | ✓ | Full TLS 1.2/1.3 |
 | HTTP Sink | ✓ | Full TLS 1.2/1.3 |
 | HTTP Client | ✓ | Client certificates |
 | TCP Source | ✗ | No encryption |
 | TCP Sink | ✗ | No encryption |
 | TCP Client | ✗ | No encryption |
 ### Server TLS Configuration
 ```toml
 [pipelines.sources.http.tls]
 enabled = true
 cert_file = "/path/to/server.pem"
 key_file = "/path/to/server.key"
 min_version = "TLS1.2"  # TLS1.2|TLS1.3
 client_auth = false
 client_ca_file = "/path/to/client-ca.pem"
 verify_client_cert = true
 ```
 ### Client TLS Configuration
 ```toml
 [pipelines.sinks.http_client.tls]
 enabled = true
 server_ca_file = "/path/to/ca.pem"  # For server verification
 server_name = "logs.example.com"
 insecure_skip_verify = false
 client_cert_file = "/path/to/client.pem"  # For mTLS
 client_key_file = "/path/to/client.key"   # For mTLS
 ```
 ### TLS Certificate Generation
 Using the `tls` command:
 ```bash
 # Generate CA certificate
 logwisp tls -ca -o myca
 # Generate server certificate
 logwisp tls -server -ca-cert myca.pem -ca-key myca.key -host localhost,server.example.com -o server
 # Generate client certificate
 logwisp tls -client -ca-cert myca.pem -ca-key myca.key -o client
 ```
 Command options:
 | Flag | Description |
 |------|-------------|
 | `-ca` | Generate CA certificate |
 | `-server` | Generate server certificate |
 | `-client` | Generate client certificate |
 | `-host` | Comma-separated hostnames/IPs |
 | `-o` | Output file prefix |
 | `-days` | Certificate validity (default: 365) |
 ## Network Rate Limiting
 ### Configuration Options
 ```toml
 [pipelines.sources.http.net_limit]
 enabled = true
 max_connections_per_ip = 10
 max_connections_total = 100
 requests_per_second = 100.0
 burst_size = 200
 response_code = 429
 response_message = "Rate limit exceeded"
 ip_whitelist = ["192.168.1.0/24"]
 ip_blacklist = ["10.0.0.0/8"]
 ```
 ### Rate Limiting Parameters
 | Parameter | Type | Description |
 |-----------|------|-------------|
 | `enabled` | bool | Enable rate limiting |
 | `max_connections_per_ip` | int | Per-IP connection limit |
 | `max_connections_total` | int | Global connection limit |
 | `requests_per_second` | float | Request rate limit |
 | `burst_size` | int | Token bucket burst capacity |
 | `response_code` | int | HTTP response code when limited |
 | `response_message` | string | Response message when limited |
 ### IP Access Control
 **Whitelist**: Only specified IPs/networks allowed
 ```toml
 ip_whitelist = [
    "192.168.1.0/24",  # Local network
    "10.0.0.0/8",      # Private network
    "203.0.113.5"      # Specific IP
 ]
 ```
 **Blacklist**: Specified IPs/networks denied
 ```toml
 ip_blacklist = [
    "192.168.1.100",   # Blocked host
    "10.0.0.0/16"      # Blocked subnet
 ]
 ```
 Processing order:
 1. Blacklist (immediate deny if matched)
 2. Whitelist (must match if configured)
 3. Rate limiting
 4. Authentication
 ## Connection Management
 ### TCP Keep-Alive
 ```toml
-[pipelines.sources.tcp]
+[[pipelines.plugin_sinks]]
 id = "tcp_out"
 type = "tcp"
 [pipelines.plugin_sinks.config]
 keep_alive = true
 keep_alive_period_ms = 30000  # 30 seconds
 ```
 Benefits:
 - Detect dead connections
 - Prevent connection timeout
 - Maintain NAT mappings
 ### Connection Timeouts
 ```toml
-[pipelines.sources.http]
+[[pipelines.plugin_sinks]]
-read_timeout_ms = 10000   # 10 seconds
+id = "http_out"
 type = "http"
 [pipelines.plugin_sinks.config]
 write_timeout_ms = 10000  # 10 seconds
 [pipelines.sinks.tcp_client]
 dial_timeout = 10         # Connection timeout
 write_timeout = 30        # Write timeout
 read_timeout = 10         # Read timeout
 ```
 ### Connection Limits
 Global limits:
 ```toml
 max_connections = 100     # Total concurrent connections
 ```
 Per-IP limits:
 ```toml
 max_connections_per_ip = 10
 ```
 ## Heartbeat Configuration
-Keep connections alive with periodic heartbeats:
+Keep connections alive with periodic heartbeats. Note that Heartbeat is a flow-level feature in the new architecture.
 ### HTTP Sink Heartbeat
 ```toml
-[pipelines.sinks.http.heartbeat]
+[pipelines.flow.heartbeat]
 enabled = true
 interval_ms = 30000
 include_timestamp = true
@@ -175,37 +42,18 @@ include_stats = false
 format = "comment"  # comment|event|json
 ```
 Formats:
 - **comment**: SSE comment (`: heartbeat`)
 - **event**: SSE event with data
 - **json**: JSON-formatted heartbeat
 ### TCP Sink Heartbeat
 ```toml
 [pipelines.sinks.tcp.heartbeat]
 enabled = true
 interval_ms = 30000
 include_timestamp = true
 include_stats = false
 format = "json"  # json|txt
 ```
 ## Network Protocols
 ### HTTP/HTTPS
- HTTP/1.1 and HTTP/2 support
+- HTTP/1.1 support
 - Persistent connections
 - Chunked transfer encoding
 - Server-Sent Events (SSE)
 ### TCP
 - Raw TCP sockets
 - Newline-delimited protocol
 - Binary-safe transmission
 - No encryption available
 ## Port Configuration
@@ -213,9 +61,7 @@ format = "json"  # json|txt
 | Service | Default Port | Protocol |
 |---------|--------------|----------|
-| HTTP Source | 8081 | HTTP/HTTPS |
+| HTTP Sink | 8080 | HTTP |
 | HTTP Sink | 8080 | HTTP/HTTPS |
 | TCP Source | 9091 | TCP |
 | TCP Sink | 9090 | TCP |
 ### Port Conflict Prevention
@@ -223,46 +69,6 @@ format = "json"  # json|txt
 LogWisp validates port usage at startup:
 - Detects port conflicts across pipelines
 - Prevents duplicate bindings
 - Suggests alternative ports
 ## Network Security
 ### Best Practices
 1. **Use TLS for HTTP** connections when possible
 2. **Implement rate limiting** to prevent DoS
 3. **Configure IP whitelists** for restricted access
 4. **Enable authentication** for all network endpoints
 5. **Use non-standard ports** to reduce scanning exposure
 6. **Monitor connection metrics** for anomalies
 7. **Set appropriate timeouts** to prevent resource exhaustion
 ### Security Warnings
 - TCP connections are **always unencrypted**
 - HTTP Basic/Token auth **requires TLS**
 - Avoid `skip_verify` in production
 - Never expose unauthenticated endpoints publicly
 ## Load Balancing
 ### Client-Side Load Balancing
 Configure multiple endpoints (future feature):
 ```toml
 [[pipelines.sinks.http_client]]
 urls = [
    "https://log1.example.com/ingest",
    "https://log2.example.com/ingest"
 ]
 strategy = "round-robin"  # round-robin|random|least-conn
 ```
 ### Server-Side Considerations
 - Use reverse proxy for load distribution
 - Configure session affinity if needed
 - Monitor individual instance health
 ## Troubleshooting
@@ -2,6 +2,8 @@
 Running, monitoring, and maintaining LogWisp in production.
 *Note: TLS, acccess control under redesign*
 ## Starting LogWisp
 ### Manual Start
@@ -60,7 +62,7 @@ kill -USR1 $(pidof logwisp)
 Test configuration without starting:
 ```bash
-logwisp --config test.toml --quiet --disable-status-reporter
+logwisp --config test.toml --quiet --status-reporter=false
 ```
 Check for errors:
@@ -168,18 +170,10 @@ Production recommendation: `info` or `warn`
 Adjust buffers based on load:
 ```toml
-# High-volume source
+[[pipelines.plugin_sources]]
-[[pipelines.sources]]
+id = "file_in"
-type = "http"
+type = "file"
-[pipelines.sources.http]
+[pipelines.plugin_sources.config]
 buffer_size = 5000  # Increase for burst traffic
 # Slow consumer sink
 [[pipelines.sinks]]
 type = "http_client"
 [pipelines.sinks.http_client]
 buffer_size = 10000  # Larger buffer for slow endpoints
 batch_size = 500     # Larger batches
 ```
 ### Rate Limiting
@@ -187,22 +181,12 @@ batch_size = 500     # Larger batches
 Protect against overload:
 ```toml
-[pipelines.rate_limit]
+[pipelines.flow.rate_limit]
 rate = 1000.0        # Entries per second
 burst = 2000.0       # Burst capacity
 policy = "drop"      # Drop excess entries
 ```
 ### Connection Limits
 Prevent resource exhaustion:
 ```toml
 [pipelines.sources.http.net_limit]
 max_connections_total = 1000
 max_connections_per_ip = 50
 ```
 ## Troubleshooting
 ### Common Issues
@@ -1,58 +1,4 @@
 # Security
-## mTLS (Mutual TLS)
+*Note: Security features like mTLS and IP-Based Access Control are currently under redesign and act as placeholders in the new architecture. Future versions will reintroduce full security capabilities.*
 Certificate-based authentication for HTTPS.
 ### Server Configuration
 ```toml
 [pipelines.sources.http.tls]
 enabled = true
 cert_file = "/path/to/server.pem"
 key_file = "/path/to/server.key"
 client_auth = true
 client_ca_file = "/path/to/ca.pem"
 verify_client_cert = true
 ```
 ### Client Configuration
 ```toml
 [pipelines.sinks.http_client.tls]
 enabled = true
 cert_file = "/path/to/client.pem"
 key_file = "/path/to/client.key"
 ```
 ### Certificate Generation
 Use the `tls` command:
 ```bash
 # Generate CA
 logwisp tls -ca -o ca
 # Generate server certificate
 logwisp tls -server -ca-cert ca.pem -ca-key ca.key -host localhost -o server
 # Generate client certificate
 logwisp tls -client -ca-cert ca.pem -ca-key ca.key -o client
 ```
 ## Access Control
 ogWisp provides IP-based access control for network connections.
 +## IP-Based Access Control
 Configure IP-based access control for sources:
 ```toml
 [pipelines.sources.http.net_limit]
 enabled = true
 ip_whitelist = ["192.168.1.0/24", "10.0.0.0/8"]
 ip_blacklist = ["192.168.1.100"]
 ```
 Priority order:
 1. Blacklist (checked first, immediate deny)
 2. Whitelist (if configured, must match)
@@ -9,13 +9,12 @@ LogWisp sinks deliver processed log entries to various destinations.
 Output to stdout/stderr.
 ```toml
-[[pipelines.sinks]]
+[[pipelines.plugin_sinks]]
 id = "console_out"
 type = "console"
-
+[pipelines.plugin_sinks.config]
 [pipelines.sinks.console]
 target = "stdout"  # stdout|stderr|split
-colorize = false
+buffer_size = 1000
 buffer_size = 100
 ```
 **Configuration Options:**
@@ -23,8 +22,7 @@ buffer_size = 100
 | Option | Type | Default | Description |
 |--------|------|---------|-------------|
 | `target` | string | "stdout" | Output target (stdout/stderr/split) |
-| `colorize` | bool | false | Enable colored output |
+| `buffer_size` | int | 1000 | Internal buffer size |
 | `buffer_size` | int | 100 | Internal buffer size |
 **Target Modes:**
 - **stdout**: All output to standard output
@@ -36,10 +34,10 @@ buffer_size = 100
 Write logs to rotating files.
 ```toml
-[[pipelines.sinks]]
+[[pipelines.plugin_sinks]]
 id = "file_out"
 type = "file"
-
+[pipelines.plugin_sinks.config]
 [pipelines.sinks.file]
 directory = "./logs"
 name = "output"
 max_size_mb = 100
@@ -74,17 +72,15 @@ flush_interval_ms = 1000
 SSE (Server-Sent Events) streaming server.
 ```toml
-[[pipelines.sinks]]
+[[pipelines.plugin_sinks]]
 id = "http_out"
 type = "http"
-
+[pipelines.plugin_sinks.config]
 [pipelines.sinks.http]
 host = "0.0.0.0"
 port = 8080
 stream_path = "/stream"
 status_path = "/status"
 buffer_size = 1000
 max_connections = 100
 read_timeout_ms = 10000
 write_timeout_ms = 10000
 ```
@@ -97,34 +93,20 @@ write_timeout_ms = 10000
 | `stream_path` | string | "/stream" | SSE stream endpoint |
 | `status_path` | string | "/status" | Status endpoint |
 | `buffer_size` | int | 1000 | Internal buffer size |
 | `max_connections` | int | 100 | Maximum concurrent clients |
 | `read_timeout_ms` | int | 10000 | Read timeout |
 | `write_timeout_ms` | int | 10000 | Write timeout |
 **Heartbeat Configuration:**
 ```toml
 [pipelines.sinks.http.heartbeat]
 enabled = true
 interval_ms = 30000
 include_timestamp = true
 include_stats = false
 format = "comment"  # comment|event|json
 ```
 ### TCP Sink
 TCP streaming server for debugging.
 ```toml
-[[pipelines.sinks]]
+[[pipelines.plugin_sinks]]
 id = "tcp_out"
 type = "tcp"
-
+[pipelines.plugin_sinks.config]
 [pipelines.sinks.tcp]
 host = "0.0.0.0"
 port = 9090
 buffer_size = 1000
 max_connections = 100
 keep_alive = true
 keep_alive_period_ms = 30000
 ```
@@ -136,132 +118,21 @@ keep_alive_period_ms = 30000
 | `host` | string | "0.0.0.0" | Bind address |
 | `port` | int | Required | Listen port |
 | `buffer_size` | int | 1000 | Internal buffer size |
 | `max_connections` | int | 100 | Maximum concurrent clients |
 | `keep_alive` | bool | true | Enable TCP keep-alive |
 | `keep_alive_period_ms` | int | 30000 | Keep-alive interval |
 | `write_timeout_ms` | int | 10000 | Write timeout |
-**Note:** TCP Sink has no authentication support (debugging only).
+### Null Sink
 ### HTTP Client Sink
 Forward logs to remote HTTP endpoints.
 ```toml
-[[pipelines.sinks]]
+[[pipelines.plugin_sinks]]
-type = "http_client"
+id = "null_out"
-
+type = "null"
 [pipelines.sinks.http_client]
 url = "https://logs.example.com/ingest"
 buffer_size = 1000
 batch_size = 100
 batch_delay_ms = 1000
 timeout_seconds = 30
 max_retries = 3
 retry_delay_ms = 1000
 retry_backoff = 2.0
 insecure_skip_verify = false
 ```
-**Configuration Options:**
+## Buffer Management
-| Option | Type | Default | Description |
+- Full input buffer: entry dropped for that sink only (counted per pipeline as `total_dropped_by_sink`)"
 |--------|------|---------|-------------|
 | `url` | string | Required | Target URL |
 | `buffer_size` | int | 1000 | Internal buffer size |
 | `batch_size` | int | 100 | Logs per request |
 | `batch_delay_ms` | int | 1000 | Max wait before sending |
 | `timeout_seconds` | int | 30 | Request timeout |
 | `max_retries` | int | 3 | Retry attempts |
 | `retry_delay_ms` | int | 1000 | Initial retry delay |
 | `retry_backoff` | float | 2.0 | Exponential backoff multiplier |
 | `insecure_skip_verify` | bool | false | Skip TLS verification |
 ### TCP Client Sink
 Forward logs to remote TCP servers.
 ```toml
 [[pipelines.sinks]]
 type = "tcp_client"
 [pipelines.sinks.tcp_client]
 host = "logs.example.com"
 port = 9090
 buffer_size = 1000
 dial_timeout = 10
 write_timeout = 30
 read_timeout = 10
 keep_alive = 30
 reconnect_delay_ms = 1000
 max_reconnect_delay_ms = 30000
 reconnect_backoff = 1.5
 ```
 **Configuration Options:**
 | Option | Type | Default | Description |
 |--------|------|---------|-------------|
 | `host` | string | Required | Target host |
 | `port` | int | Required | Target port |
 | `buffer_size` | int | 1000 | Internal buffer size |
 | `dial_timeout` | int | 10 | Connection timeout (seconds) |
 | `write_timeout` | int | 30 | Write timeout (seconds) |
 | `read_timeout` | int | 10 | Read timeout (seconds) |
 | `keep_alive` | int | 30 | TCP keep-alive (seconds) |
 | `reconnect_delay_ms` | int | 1000 | Initial reconnect delay |
 | `max_reconnect_delay_ms` | int | 30000 | Maximum reconnect delay |
 | `reconnect_backoff` | float | 1.5 | Backoff multiplier |
 ## Network Sink Features
 ### Network Rate Limiting
 Available for HTTP and TCP sinks:
 ```toml
 [pipelines.sinks.http.net_limit]
 enabled = true
 max_connections_per_ip = 10
 max_connections_total = 100
 ip_whitelist = ["192.168.1.0/24"]
 ip_blacklist = ["10.0.0.0/8"]
 ```
 ### TLS Configuration (HTTP Only)
 ```toml
 [pipelines.sinks.http.tls]
 enabled = true
 cert_file = "/path/to/cert.pem"
 key_file = "/path/to/key.pem"
 ca_file = "/path/to/ca.pem"
 min_version = "TLS1.2"
 client_auth = false
 ```
 HTTP Client TLS:
 ```toml
 [pipelines.sinks.http_client.tls]
 enabled = true
 server_ca_file = "/path/to/ca.pem"  # For server verification
 server_name = "logs.example.com"
 insecure_skip_verify = false
 client_cert_file = "/path/to/client.pem"  # For mTLS
 client_key_file = "/path/to/client.key"   # For mTLS
 ```
 ## Sink Chaining
 Designed connection patterns:
 ### Log Aggregation
 - **HTTP Client Sink → HTTP Source**: HTTP/HTTPS (optional mTLS for HTTPS)
 - **TCP Client Sink → TCP Source**: Raw TCP
 ### Live Monitoring
 - **HTTP Sink**: Browser-based SSE streaming
 - **TCP Sink**: Debug interface (telnet/netcat)
 ## Sink Statistics
@@ -6,31 +6,29 @@ LogWisp sources monitor various inputs and generate log entries for pipeline pro
 ### Directory Source
-Monitors a directory for log files matching a pattern.
+Monitors a directory for log files matching a pattern. (type: `file`)
 ```toml
-[[pipelines.sources]]
+[[pipelines.plugin_sources]]
-type = "directory"
+id = "file_in"
-
+type = "file"
-[pipelines.sources.directory]
+[pipelines.plugin_sources.config]
-path = "/var/log/myapp"
+directory = "/var/log/myapp"
 pattern = "*.log"          # Glob pattern
 check_interval_ms = 100    # Poll interval
 recursive = false           # Scan subdirectories
 ```
 **Configuration Options:**
 | Option | Type | Default | Description |
 |--------|------|---------|-------------|
-| `path` | string | Required | Directory to monitor |
+| `directory` | string | Required | Directory to monitor |
 | `pattern` | string | "*" | File pattern (glob) |
 | `check_interval_ms` | int | 100 | File check interval in milliseconds |
 | `recursive` | bool | false | Include subdirectories |
 **Features:**
- Automatic file rotation detection
+- Automatic rotation detection (inode + size tracking)
- Position tracking (resume after restart)
+- In-memory position tracking; on restart, monitoring resumes from the current end of each file (offsets are not persisted)
 - Concurrent file monitoring
 - Pattern-based file selection
@@ -39,10 +37,10 @@ recursive = false           # Scan subdirectories
 Reads log entries from standard input.
 ```toml
-[[pipelines.sources]]
+[[pipelines.plugin_sources]]
 id = "console_in"
 type = "console"
-
+[pipelines.plugin_sources.config]
 [pipelines.sources.stdin]
 buffer_size = 1000
 ```
@@ -57,106 +55,29 @@ buffer_size = 1000
 - Automatic level detection
 - Non-blocking reads
-### HTTP Source
+### Random Source
 REST endpoint for log ingestion.
 ```toml
-[[pipelines.sources]]
+[[pipelines.plugin_sources]]
-type = "http"
+id = "random_in"
-
+type = "random"
-[pipelines.sources.http]
+[pipelines.plugin_sources.config]
-host = "0.0.0.0"
+interval_ms = 500
-port = 8081
+jitter_ms = 0
-ingest_path = "/ingest"
+format = "txt"
-buffer_size = 1000
+length = 20
-max_body_size = 1048576  # 1MB
+special = false
 read_timeout_ms = 10000
 write_timeout_ms = 10000
 ```
-**Configuration Options:**
+ **Configuration Options:**
 | Option | Type | Default | Description |
 |--------|------|---------|-------------|
-| `host` | string | "0.0.0.0" | Bind address |
+| `interval_ms` | int | 500 | Generation interval |
-| `port` | int | Required | Listen port |
+| `jitter_ms` | int | 0 | Random jitter interval |
-| `ingest_path` | string | "/ingest" | Ingestion endpoint path |
+| `format` | string | "txt" | "txt", "json", "raw" |
-| `buffer_size` | int | 1000 | Internal buffer size |
+| `length` | int | 20 | Log length |
-| `max_body_size` | int | 1048576 | Maximum request body size |
+| `special` | bool | false | Include special characters |
 | `read_timeout_ms` | int | 10000 | Read timeout |
 | `write_timeout_ms` | int | 10000 | Write timeout |
 **Input Formats:**
 - Single JSON object
 - JSON array
 - Newline-delimited JSON (NDJSON)
 - Plain text (one entry per line)
 ### TCP Source
 Raw TCP socket listener for log ingestion.
 ```toml
 [[pipelines.sources]]
 type = "tcp"
 [pipelines.sources.tcp]
 host = "0.0.0.0"
 port = 9091
 buffer_size = 1000
 read_timeout_ms = 10000
 keep_alive = true
 keep_alive_period_ms = 30000
 ```
 **Configuration Options:**
 | Option | Type | Default | Description |
 |--------|------|---------|-------------|
 | `host` | string | "0.0.0.0" | Bind address |
 | `port` | int | Required | Listen port |
 | `buffer_size` | int | 1000 | Internal buffer size |
 | `read_timeout_ms` | int | 10000 | Read timeout |
 | `keep_alive` | bool | true | Enable TCP keep-alive |
 | `keep_alive_period_ms` | int | 30000 | Keep-alive interval |
 **Protocol:**
 - Newline-delimited JSON
 - One log entry per line
 - UTF-8 encoding
 ## Network Source Features
 ### Network Rate Limiting
 Available for HTTP and TCP sources:
 ```toml
 [pipelines.sources.http.net_limit]
 enabled = true
 max_connections_per_ip = 10
 max_connections_total = 100
 requests_per_second = 100.0
 burst_size = 200
 response_code = 429
 response_message = "Rate limit exceeded"
 ip_whitelist = ["192.168.1.0/24"]
 ip_blacklist = ["10.0.0.0/8"]
 ```
 ### TLS Configuration (HTTP Only)
 ```toml
 [pipelines.sources.http.tls]
 enabled = true
 cert_file = "/path/to/cert.pem"
 key_file = "/path/to/key.pem"
 min_version = "TLS1.2"
 client_auth = true
 client_ca_file = "/path/to/client-ca.pem"
 verify_client_cert = true
 ```
 ## Source Statistics
@@ -168,6 +89,15 @@ All sources track:
 - Active connections (network sources)
 - Source-specific metrics
 ### Null Source
 ```toml
 [[pipelines.plugin_sources]]
 id = "null_in"
 type = "null"
 [pipelines.plugin_sources.config]
 ```
 ## Buffer Management
 Each source maintains internal buffers:
@@ -1,26 +1,26 @@
 module logwisp
-go 1.25.4
+go 1.26.0
 require (
-	github.com/lixenwraith/config v0.1.1-0.20251114180219-f7875023a51b
+	github.com/lixenwraith/config v0.1.1-0.20260712172228-ccd280ba6a98
-	github.com/lixenwraith/log v0.1.1-0.20251115213227-55d2c92d483f
+	github.com/lixenwraith/log v0.1.1-0.20251117213308-9ae1b6669bf3
-	github.com/panjf2000/gnet/v2 v2.9.7
+	github.com/panjf2000/gnet/v2 v2.10.0
-	github.com/valyala/fasthttp v1.68.0
+	github.com/valyala/fasthttp v1.72.0
 )
 require (
 	github.com/BurntSushi/toml v1.6.0 // indirect
-	github.com/andybalholm/brotli v1.2.0 // indirect
+	github.com/andybalholm/brotli v1.2.2 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
-	github.com/go-viper/mapstructure/v2 v2.4.0 // indirect
+	github.com/go-viper/mapstructure/v2 v2.5.0 // indirect
-	github.com/klauspost/compress v1.18.2 // indirect
+	github.com/klauspost/compress v1.19.0 // indirect
-	github.com/panjf2000/ants/v2 v2.11.4 // indirect
+	github.com/panjf2000/ants/v2 v2.12.1 // indirect
 	github.com/valyala/bytebufferpool v1.0.0 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
-	go.uber.org/zap v1.27.1 // indirect
+	go.uber.org/zap v1.28.0 // indirect
-	golang.org/x/sync v0.19.0 // indirect
+	golang.org/x/sync v0.22.0 // indirect
-	golang.org/x/sys v0.39.0 // indirect
+	golang.org/x/sys v0.47.0 // indirect
 	gopkg.in/natefinch/lumberjack.v2 v2.2.1 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )
@@ -2,24 +2,50 @@ github.com/BurntSushi/toml v1.6.0 h1:dRaEfpa2VI55EwlIW72hMRHdWouJeRF7TPYhI+AUQjk
 github.com/BurntSushi/toml v1.6.0/go.mod h1:ukJfTF/6rtPPRCnwkur4qwRxa8vTRFBF0uk2lLoLwho=
 github.com/andybalholm/brotli v1.2.0 h1:ukwgCxwYrmACq68yiUqwIWnGY0cTPox/M94sVwToPjQ=
 github.com/andybalholm/brotli v1.2.0/go.mod h1:rzTDkvFWvIrjDXZHkuS16NPggd91W3kUSvPlQ1pLaKY=
 github.com/andybalholm/brotli v1.2.1 h1:R+f5xP285VArJDRgowrfb9DqL18yVK0gKAW/F+eTWro=
 github.com/andybalholm/brotli v1.2.1/go.mod h1:rzTDkvFWvIrjDXZHkuS16NPggd91W3kUSvPlQ1pLaKY=
 github.com/andybalholm/brotli v1.2.2 h1:HzTuoo2ErYQqf5qvcJInB8uvqSVxRttzkFexPWtnceM=
 github.com/andybalholm/brotli v1.2.2/go.mod h1:rzTDkvFWvIrjDXZHkuS16NPggd91W3kUSvPlQ1pLaKY=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/go-viper/mapstructure/v2 v2.4.0 h1:EBsztssimR/CONLSZZ04E8qAkxNYq4Qp9LvH92wZUgs=
 github.com/go-viper/mapstructure/v2 v2.4.0/go.mod h1:oJDH3BJKyqBA2TXFhDsKDGDTlndYOZ6rGS0BRZIxGhM=
 github.com/go-viper/mapstructure/v2 v2.5.0 h1:vM5IJoUAy3d7zRSVtIwQgBj7BiWtMPfmPEgAXnvj1Ro=
 github.com/go-viper/mapstructure/v2 v2.5.0/go.mod h1:oJDH3BJKyqBA2TXFhDsKDGDTlndYOZ6rGS0BRZIxGhM=
 github.com/klauspost/compress v1.18.1 h1:bcSGx7UbpBqMChDtsF28Lw6v/G94LPrrbMbdC3JH2co=
 github.com/klauspost/compress v1.18.1/go.mod h1:ZQFFVG+MdnR0P+l6wpXgIL4NTtwiKIdBnrBd8Nrxr+0=
 github.com/klauspost/compress v1.18.2 h1:iiPHWW0YrcFgpBYhsA6D1+fqHssJscY/Tm/y2Uqnapk=
 github.com/klauspost/compress v1.18.2/go.mod h1:R0h/fSBs8DE4ENlcrlib3PsXS61voFxhIs2DeRhCvJ4=
 github.com/klauspost/compress v1.18.3 h1:9PJRvfbmTabkOX8moIpXPbMMbYN60bWImDDU7L+/6zw=
 github.com/klauspost/compress v1.18.3/go.mod h1:R0h/fSBs8DE4ENlcrlib3PsXS61voFxhIs2DeRhCvJ4=
 github.com/klauspost/compress v1.18.4 h1:RPhnKRAQ4Fh8zU2FY/6ZFDwTVTxgJ/EMydqSTzE9a2c=
 github.com/klauspost/compress v1.18.4/go.mod h1:R0h/fSBs8DE4ENlcrlib3PsXS61voFxhIs2DeRhCvJ4=
 github.com/klauspost/compress v1.18.6 h1:2jupLlAwFm95+YDR+NwD2MEfFO9d4z4Prjl1XXDjuao=
 github.com/klauspost/compress v1.18.6/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ=
 github.com/klauspost/compress v1.19.0 h1:sXLILfc9jV2QYWkzFOPWStmcUVH2RHEB1JCdY2oVvCQ=
 github.com/klauspost/compress v1.19.0/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ=
 github.com/lixenwraith/config v0.1.1-0.20251114180219-f7875023a51b h1:TzTV0ArJ+nzVGPN8aiEJ2MknUqJdmHRP/0/RSfov2Qw=
 github.com/lixenwraith/config v0.1.1-0.20251114180219-f7875023a51b/go.mod h1:roNPTSCT5HSV9dru/zi/Catwc3FZVCFf7vob2pSlNW0=
 github.com/lixenwraith/config v0.1.1-0.20260712172228-ccd280ba6a98 h1:VEFo2WsYgM5YYfzAfvXRKigCWr3zEz6h4M59N5TXMpk=
 github.com/lixenwraith/config v0.1.1-0.20260712172228-ccd280ba6a98/go.mod h1:J9ydxY7he4Dz+S59xKs/kDQ2YAv+zOLF05b3Rr2/ogE=
 github.com/lixenwraith/log v0.1.1-0.20251115213227-55d2c92d483f h1:X2LX5FQEuWYGBS3qp5z7XxBB1sWAlqumf/oW7n/f9c0=
 github.com/lixenwraith/log v0.1.1-0.20251115213227-55d2c92d483f/go.mod h1:XcRPRuijAs+43Djk8VmioUJhcK8irRzUjCZaZqkd3gg=
 github.com/lixenwraith/log v0.1.1-0.20251117213308-9ae1b6669bf3 h1:brSUhER7EZ28aMRFTSovZskiIoobUiAWlx+DYXYvWXQ=
 github.com/lixenwraith/log v0.1.1-0.20251117213308-9ae1b6669bf3/go.mod h1:MY59N65ltw/9uTqJKwCRKjxO6w3CApXsLXCVuaekbu0=
 github.com/panjf2000/ants/v2 v2.11.3 h1:AfI0ngBoXJmYOpDh9m516vjqoUu2sLrIVgppI9TZVpg=
 github.com/panjf2000/ants/v2 v2.11.3/go.mod h1:8u92CYMUc6gyvTIw8Ru7Mt7+/ESnJahz5EVtqfrilek=
 github.com/panjf2000/ants/v2 v2.11.4 h1:UJQbtN1jIcI5CYNocTj0fuAUYvsLjPoYi0YuhqV/Y48=
 github.com/panjf2000/ants/v2 v2.11.4/go.mod h1:8u92CYMUc6gyvTIw8Ru7Mt7+/ESnJahz5EVtqfrilek=
 github.com/panjf2000/ants/v2 v2.11.5 h1:a7LMnMEeux/ebqTux140tRiaqcFTV0q2bEHF03nl6Rg=
 github.com/panjf2000/ants/v2 v2.11.5/go.mod h1:8u92CYMUc6gyvTIw8Ru7Mt7+/ESnJahz5EVtqfrilek=
 github.com/panjf2000/ants/v2 v2.12.1 h1:BWvU2wHpyXWxhhNXsGB6JXLCNbshyLd1QxvoAmZnu10=
 github.com/panjf2000/ants/v2 v2.12.1/go.mod h1:tSQuaNQ6r6NRhPt+IZVUevvDyFMTs+eS4ztZc52uJTY=
 github.com/panjf2000/gnet/v2 v2.9.7 h1:6zW7Jl3oAfXwSuh1PxHLndoL2MQRWx0AJR6aaQjxUgA=
 github.com/panjf2000/gnet/v2 v2.9.7/go.mod h1:WQTxDWYuQ/hz3eccH0FN32IVuvZ19HewEWx0l62fx7E=
 github.com/panjf2000/gnet/v2 v2.9.8 h1:OzVnIKm4UMoGphWgypcVZMq8roVjXzXOZvLdyZHxvkE=
 github.com/panjf2000/gnet/v2 v2.9.8/go.mod h1:f9wdbOFsdbZqlSvXctWbPRW5bB/W++q8Zqz+D7tQIVQ=
 github.com/panjf2000/gnet/v2 v2.10.0 h1:rC4jNF+jtXj/FH+8JOIQ3XxjD+yBunYBLKg9TE3dc4g=
 github.com/panjf2000/gnet/v2 v2.10.0/go.mod h1:f9wdbOFsdbZqlSvXctWbPRW5bB/W++q8Zqz+D7tQIVQ=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA=
@@ -28,6 +54,12 @@ github.com/valyala/bytebufferpool v1.0.0 h1:GqA5TC/0021Y/b9FG4Oi9Mr3q7XYx6Kllzaw
 github.com/valyala/bytebufferpool v1.0.0/go.mod h1:6bBcMArwyJ5K/AmCkWv1jt77kVWyCJ6HpOuEn7z0Csc=
 github.com/valyala/fasthttp v1.68.0 h1:v12Nx16iepr8r9ySOwqI+5RBJ/DqTxhOy1HrHoDFnok=
 github.com/valyala/fasthttp v1.68.0/go.mod h1:5EXiRfYQAoiO/khu4oU9VISC/eVY6JqmSpPJoHCKsz4=
 github.com/valyala/fasthttp v1.69.0 h1:fNLLESD2SooWeh2cidsuFtOcrEi4uB4m1mPrkJMZyVI=
 github.com/valyala/fasthttp v1.69.0/go.mod h1:4wA4PfAraPlAsJ5jMSqCE2ug5tqUPwKXxVj8oNECGcw=
 github.com/valyala/fasthttp v1.71.0 h1:tepR7H+Guh9VUqxxcPggYi8R3lGUu2Rsdh+z7/FCY3k=
 github.com/valyala/fasthttp v1.71.0/go.mod h1:z1sDUvOShhXq/C9mwH/fSm1Vb71tUJwmQdgkBrBNwnA=
 github.com/valyala/fasthttp v1.72.0 h1:R7kYdoWhn1ye1fVpP+cDHDJwYm3NkwLliwgzJ/Abg7M=
 github.com/valyala/fasthttp v1.72.0/go.mod h1:zsbLTYqcpIktdQytlVBwIjY9La5d6bs990nBxWg8efk=
 github.com/xyproto/randomstring v1.0.5 h1:YtlWPoRdgMu3NZtP45drfy1GKoojuR7hmRcnhZqKjWU=
 github.com/xyproto/randomstring v1.0.5/go.mod h1:rgmS5DeNXLivK7YprL0pY+lTuhNQW3iGxZ18UQApw/E=
 go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
@@ -38,14 +70,30 @@ go.uber.org/zap v1.27.0 h1:aJMhYGrd5QSmlpLMr2MftRKl7t8J8PTZPA732ud/XR8=
 go.uber.org/zap v1.27.0/go.mod h1:GB2qFLM7cTU87MWRP2mPIjqfIDnGu+VIO4V/SdhGo2E=
 go.uber.org/zap v1.27.1 h1:08RqriUEv8+ArZRYSTXy1LeBScaMpVSTBhCeaZYfMYc=
 go.uber.org/zap v1.27.1/go.mod h1:GB2qFLM7cTU87MWRP2mPIjqfIDnGu+VIO4V/SdhGo2E=
 go.uber.org/zap v1.28.0 h1:IZzaP1Fv73/T/pBMLk4VutPl36uNC+OSUh3JLG3FIjo=
 go.uber.org/zap v1.28.0/go.mod h1:rDLpOi171uODNm/mxFcuYWxDsqWSAVkFdX4XojSKg/Q=
 golang.org/x/sync v0.11.0 h1:GGz8+XQP4FvTTrjZPzNKTMFtSXH80RAzG+5ghFPgK9w=
 golang.org/x/sync v0.11.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
 golang.org/x/sync v0.19.0 h1:vV+1eWNmZ5geRlYjzm2adRgW2/mcpevXNg50YZtPCE4=
 golang.org/x/sync v0.19.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI=
 golang.org/x/sync v0.20.0 h1:e0PTpb7pjO8GAtTs2dQ6jYa5BWYlMuX047Dco/pItO4=
 golang.org/x/sync v0.20.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0=
 golang.org/x/sync v0.21.0 h1:HLII4xRRTtCRkxYp4HNFF0Js/Og6q2i++KXbg0gHCwM=
 golang.org/x/sync v0.21.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0=
 golang.org/x/sync v0.22.0 h1:SZjpbeLmrCk4xhRSZFNZW5gFUeCeFgjekvI/+gfScek=
 golang.org/x/sync v0.22.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0=
 golang.org/x/sys v0.37.0 h1:fdNQudmxPjkdUTPnLn5mdQv7Zwvbvpaxqs831goi9kQ=
 golang.org/x/sys v0.37.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks=
 golang.org/x/sys v0.39.0 h1:CvCKL8MeisomCi6qNZ+wbb0DN9E5AATixKsvNtMoMFk=
 golang.org/x/sys v0.39.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks=
 golang.org/x/sys v0.41.0 h1:Ivj+2Cp/ylzLiEU89QhWblYnOE9zerudt9Ftecq2C6k=
 golang.org/x/sys v0.41.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks=
 golang.org/x/sys v0.45.0 h1:dO4czNzziLiiXplLQgBCEpCvXQ3dnkn0SdaZSYdQ+FY=
 golang.org/x/sys v0.45.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
 golang.org/x/sys v0.46.0 h1:noSf2Fq6F8DBgS+LysIkx7rIExoNHJsxOAtPp4rthXw=
 golang.org/x/sys v0.46.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
 golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs=
 golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/natefinch/lumberjack.v2 v2.2.1 h1:bBRl1b0OH9s/DuPhuXpNl+VtCaJXFZ5/uEFST95x9zc=
@@ -6,20 +6,25 @@ import (
 	_ "logwisp/src/internal/source/console"
 	_ "logwisp/src/internal/source/file"
 	_ "logwisp/src/internal/source/httpchain"
 	_ "logwisp/src/internal/source/null"
 	_ "logwisp/src/internal/source/random"
 	_ "logwisp/src/internal/source/tcpchain"
 	_ "logwisp/src/internal/sink/console"
 	_ "logwisp/src/internal/sink/file"
 	_ "logwisp/src/internal/sink/http"
 	_ "logwisp/src/internal/sink/httpchain"
 	_ "logwisp/src/internal/sink/null"
 	_ "logwisp/src/internal/sink/tcp"
 	_ "logwisp/src/internal/sink/tcpchain"
 	"logwisp/src/internal/config"
 	"logwisp/src/internal/service"
 	"logwisp/src/internal/version"
 	"github.com/lixenwraith/log"
 	"github.com/lixenwraith/log/sanitizer"
 )
 // bootstrapInitial handles initial service startup with status reporter
@@ -132,6 +137,14 @@ func initializeLogger(cfg *config.Config) error {
 	}
 	logCfg.Level = levelValue
 	// Configure log format
 	if cfg.Logging.Format != "" {
 		logCfg.Format = cfg.Logging.Format
 	}
 	if cfg.Logging.Sanitization != "" {
 		logCfg.Sanitization = sanitizer.PolicyPreset(cfg.Logging.Sanitization)
 	}
 	// Configure based on output mode
 	switch cfg.Logging.Output {
 	case "none":
@@ -0,0 +1,73 @@
 package main
 import (
 	"fmt"
 	"logwisp/src/internal/version"
 	"os"
 )
 // helpText is the CLI usage reference. Flags map 1:1 to TOML config paths.
 const helpText = `LogWisp %s - log collection, processing, and distribution
 Usage:
  logwisp [options]
  logwisp help | -h | --help
  logwisp --version
 Any configuration key is settable as a flag using its TOML path:
  --<path>=<value>              e.g. --logging.level=debug
 Common options:
  -c, --config <path>           Configuration file (default: ./logwisp.toml)
      --quiet                   Suppress console output
      --status_reporter=<bool>  Periodic status logging (default: true)
      --auto_reload=<bool>      Config hot reload on file change (default: false)
 Logging:
      --logging.output=<mode>   file|stdout|stderr|split|all|none
      --logging.level=<level>   debug|info|warn|error
      --logging.file.directory=<path>
      --logging.console.target=<target>   stdout|stderr|split
 Pipelines (N = 0-based index):
      --pipelines.N.name=<name>
      --pipelines.N.plugin_sources.N.type=<type>   file|console|random|null
      --pipelines.N.plugin_sinks.N.type=<type>     console|file|http|tcp|null
      --pipelines.N.flow.filters.N.patterns='["ERROR","WARN"]'
 Environment:
  LOGWISP_<PATH>                Config path, '.' -> '_', uppercase
                                e.g. LOGWISP_LOGGING_LEVEL=debug
  LOGWISP_CONFIG_FILE           Configuration file path
  LOGWISP_CONFIG_DIR            Configuration directory
 Signals:
  SIGINT, SIGTERM               Graceful shutdown
  SIGHUP, SIGUSR1               Reload configuration
 Exit codes:
  0  success
  1  general error
  2  configuration file not found
 `
 // handleHelp prints usage and exits if a help request is present in args
 func handleHelp(args []string) {
 	if len(args) > 0 && args[0] == "help" {
 		printHelp()
 	}
 	for _, arg := range args {
 		if arg == "--" {
 			break // end of flags
 		}
 		if arg == "-h" || arg == "--help" {
 			printHelp()
 		}
 	}
 }
 // printHelp writes usage to stdout and exits with success
 func printHelp() {
 	fmt.Printf(helpText, version.Short())
 	os.Exit(0)
 }
@@ -7,7 +7,6 @@ import (
 	"os/signal"
 	"strings"
 	"syscall"
 	"time"
 	"logwisp/src/internal/config"
 	"logwisp/src/internal/core"
@@ -25,6 +24,10 @@ func main() {
 	// Emulates nohup
 	signal.Ignore(syscall.SIGHUP)
 	// Help handled before config parsing; loader has no help flag.
 	// Also the future dispatch point for subcommands (tls, etc.)
 	handleHelp(os.Args[1:])
 	// Load configuration with automatic CLI parsing
 	cfg, err := config.Load(os.Args[1:])
 	if err != nil {
@@ -64,8 +67,6 @@ func main() {
 		"status_reporter", cfg.StatusReporter,
 		"auto_reload", cfg.ConfigAutoReload)
 	time.Sleep(time.Second)
 	// Create context for shutdown
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
@@ -0,0 +1,98 @@
 package chain
 import (
 	"encoding/json"
 	"fmt"
 	"logwisp/src/internal/core"
 	"math/rand/v2"
 	"time"
 )
 // ProtocolVersion is declared in the hello preamble
 const ProtocolVersion = 1
 // Hello is the first NDJSON line sent by the dialing side after connect.
 // Reserved for future revisions: auth credential, feature flags (ack, compression).
 type Hello struct {
 	LogWisp int    `json:"logwisp"`
 	Node    string `json:"node,omitempty"`
 	// Auth     string   `json:"auth,omitempty"`
 	// Features []string `json:"features,omitempty"`
 }
 // HTTP transport mapping of the chain protocol.
 // Hello preamble equivalent: protocol + node carried as request headers.
 // Reserved extension point: Authorization header for auth, TLS at transport.
 const (
 	HeaderProtocol    = "X-Logwisp-Protocol"
 	HeaderNode        = "X-Logwisp-Node"
 	HeaderAccepted    = "X-Logwisp-Accepted"
 	ContentTypeNDJSON = "application/x-ndjson"
 )
 // EncodeHello serializes a newline-terminated hello preamble
 func EncodeHello(node string) ([]byte, error) {
 	b, err := json.Marshal(Hello{LogWisp: ProtocolVersion, Node: node})
 	if err != nil {
 		return nil, err
 	}
 	return append(b, '\n'), nil
 }
 // DecodeHello parses and validates a hello preamble line
 func DecodeHello(line []byte) (Hello, error) {
 	var h Hello
 	if err := json.Unmarshal(line, &h); err != nil {
 		return h, fmt.Errorf("malformed hello: %w", err)
 	}
 	if h.LogWisp != ProtocolVersion {
 		return h, fmt.Errorf("unsupported protocol version: %d", h.LogWisp)
 	}
 	return h, nil
 }
 // DecodeEntry parses a canonical LogEntry line and applies the node trust policy
 func DecodeEntry(line []byte, connNode string, trustNode bool) (core.LogEntry, error) {
 	var entry core.LogEntry
 	if err := json.Unmarshal(line, &entry); err != nil {
 		return core.LogEntry{}, err
 	}
 	if entry.Time.IsZero() {
 		entry.Time = time.Now()
 	}
 	if entry.Node == "" || !trustNode {
 		entry.Node = connNode
 	}
 	entry.RawSize = int64(len(line))
 	return entry, nil
 }
 // EntryFromEvent extracts the structured entry, stamping node identity at
 // first hop. Second return is true when synthesized from a formatted payload.
 func EntryFromEvent(event core.TransportEvent, node, fallbackSource string) (core.LogEntry, bool) {
 	entry := event.Entry
 	synthesized := false
 	if entry.Time.IsZero() {
 		synthesized = true
 		entry = core.LogEntry{
 			Time:    event.Time,
 			Source:  fallbackSource,
 			Message: string(event.Payload),
 		}
 	}
 	if entry.Node == "" {
 		entry.Node = node
 	}
 	return entry, synthesized
 }
 // BackoffDelay computes exponential backoff with ±20% jitter
 func BackoffDelay(minD, maxD time.Duration, failures int) time.Duration {
 	d := maxD
 	if failures < 63 {
 		if v := minD << uint(failures-1); v > 0 && v < maxD {
 			d = v
 		}
 	}
 	return d - d/5 + time.Duration(rand.Int64N(int64(2*d/5)+1))
 }
@@ -205,6 +205,32 @@ type ConsoleSourceOptions struct {
 	BufferSize int64 `toml:"buffer_size"`
 }
 // TCPChainSourceOptions defines settings for a stdlib TCP listener ingesting
 // NDJSON entries from upstream logwisp tcp_chain sinks
 type TCPChainSourceOptions struct {
 	Host           string `toml:"host"`
 	Port           int64  `toml:"port"`
 	BufferSize     int64  `toml:"buffer_size"`
 	MaxConnections int64  `toml:"max_connections"`  // 0 = unlimited
 	ReadTimeoutMS  int64  `toml:"read_timeout_ms"`  // per-connection idle deadline, 0 = none
 	HelloTimeoutMS int64  `toml:"hello_timeout_ms"` // preamble deadline
 	TrustNode      bool   `toml:"trust_node"`       // false: force node label from remote address
 	// Future: TLS/auth options
 }
 // HTTPChainSourceOptions defines settings for a stdlib HTTP listener ingesting
 // NDJSON batches from upstream logwisp http_chain sinks
 type HTTPChainSourceOptions struct {
 	Host          string `toml:"host"`
 	Port          int64  `toml:"port"`
 	IngestPath    string `toml:"ingest_path"`
 	BufferSize    int64  `toml:"buffer_size"`
 	MaxBodyBytes  int64  `toml:"max_body_bytes"`  // per-request cap
 	ReadTimeoutMS int64  `toml:"read_timeout_ms"` // full request read deadline
 	TrustNode     bool   `toml:"trust_node"`      // false: force node label from remote address
 	// Future: TLS/auth options
 }
 // --- Sink Options ---
 // PluginSinkConfig represents a sink plugin instance configuration
@@ -251,16 +277,49 @@ type TCPSinkOptions struct {
 	Port            int64  `toml:"port"`
 	BufferSize      int64  `toml:"buffer_size"`
 	WriteTimeout    int64  `toml:"write_timeout_ms"`
 	KeepAlive       bool   `toml:"keep_alive"`
 	KeepAlivePeriod int64  `toml:"keep_alive_period_ms"`
 	KeepAlive       bool   `toml:"keep_alive"`
 }
 // HTTPSinkOptions defines settings for an HTTP SSE server sink
 type HTTPSinkOptions struct {
 	Host         string `toml:"host"`
 	Port         int64  `toml:"port"`
 	StreamPath   string `toml:"stream_path"`
 	StatusPath   string `toml:"status_path"`
 	Host         string `toml:"host"`
 	Port         int64  `toml:"port"`
 	BufferSize   int64  `toml:"buffer_size"`
 	WriteTimeout int64  `toml:"write_timeout_ms"`
 }
 // TCPChainSinkOptions defines settings for a stdlib TCP client forwarding
 // entries to a downstream logwisp tcp_chain source
 type TCPChainSinkOptions struct {
 	Node              string `toml:"node"` // origin label, default: os.Hostname()
 	Host              string `toml:"host"`
 	Port              int64  `toml:"port"`
 	BufferSize        int64  `toml:"buffer_size"`
 	DialTimeoutMS     int64  `toml:"dial_timeout_ms"`
 	WriteTimeoutMS    int64  `toml:"write_timeout_ms"`
 	BackoffMinMS      int64  `toml:"backoff_min_ms"`
 	BackoffMaxMS      int64  `toml:"backoff_max_ms"`
 	KeepAlivePeriodMS int64  `toml:"keep_alive_period_ms"`
 	KeepAlive         bool   `toml:"keep_alive"`
 	// Future: TLS/auth options
 }
 // HTTPChainSinkOptions defines settings for a stdlib HTTP client posting
 // NDJSON batches to a downstream logwisp http_chain source
 type HTTPChainSinkOptions struct {
 	Node             string `toml:"node"` // origin label, default: os.Hostname()
 	Host             string `toml:"host"`
 	Port             int64  `toml:"port"`
 	IngestPath       string `toml:"ingest_path"`
 	BufferSize       int64  `toml:"buffer_size"`
 	MaxBatchCount    int64  `toml:"max_batch_count"`
 	MaxBatchBytes    int64  `toml:"max_batch_bytes"`
 	FlushIntervalMS  int64  `toml:"flush_interval_ms"`
 	RequestTimeoutMS int64  `toml:"request_timeout_ms"` // covers dial + write + response
 	BackoffMinMS     int64  `toml:"backoff_min_ms"`
 	BackoffMaxMS     int64  `toml:"backoff_max_ms"`
 	// Future: TLS/auth options
 }
@@ -65,6 +65,12 @@ func Load(args []string) (*Config, error) {
 	// Store the manager for hot reload
 	configManager = cfg
 	// Surface typo'd flags (e.g. --status-reporter vs --status_reporter);
 	// pre-logger phase, stderr only, suppressed in quiet mode
 	if unknown := cfg.UnknownCLIKeys(); len(unknown) > 0 && !finalConfig.Quiet {
 		fmt.Fprintf(os.Stderr, "Warning: unrecognized flags ignored: %v\n", unknown)
 	}
 	// Start watcher if auto-reload is enabled
 	if finalConfig.ConfigAutoReload {
 		watchOpts := lconfig.WatchOptions{
@@ -99,6 +105,7 @@ func defaults() *Config {
 		Logging: &LogConfig{
 			Output: "stdout",
 			Level:  "info",
 			Format: "txt",
 			File: &LogFileConfig{
 				Directory:      "./log",
 				Name:           "logwisp",
@@ -17,6 +17,15 @@ func ValidateConfig(cfg *Config) error {
 		return fmt.Errorf("no pipelines configured")
 	}
 	// Reject duplicate pipeline names (service map is keyed by name)
 	names := make(map[string]struct{}, len(cfg.Pipelines))
 	for i, p := range cfg.Pipelines {
 		if _, dup := names[p.Name]; dup {
 			return fmt.Errorf("pipeline[%d]: duplicate name %q", i, p.Name)
 		}
 		names[p.Name] = struct{}{}
 	}
 	if err := validateLogConfig(cfg.Logging); err != nil {
 		return fmt.Errorf("logging: %w", err)
 	}
@@ -52,6 +61,17 @@ func validateLogConfig(cfg *LogConfig) error {
 		return fmt.Errorf("level: %w", err)
 	}
 	if cfg.Format != "" {
 		if err := lconfig.OneOf("raw", "txt", "json")(cfg.Format); err != nil {
 			return fmt.Errorf("format: %w", err)
 		}
 	}
 	if cfg.Sanitization != "" {
 		if err := lconfig.OneOf("raw", "json", "txt", "shell")(cfg.Sanitization); err != nil {
 			return fmt.Errorf("sanitization: %w", err)
 		}
 	}
 	if cfg.Console != nil {
 		validateTarget := lconfig.OneOf("stdout", "stderr", "split")
 		if err := validateTarget(cfg.Console.Target); err != nil {
@@ -5,9 +5,10 @@ import (
 	"time"
 )
-// Represents a single log record flowing through the pipeline
+// LogEntry represents a single log record flowing through the pipeline
 type LogEntry struct {
 	Time    time.Time       `json:"time"`
 	Node    string          `json:"node,omitempty"` // origin node identity for chained topologies; first hop stamps, relays preserve
 	Source  string          `json:"source"`
 	Level   string          `json:"level,omitempty"`
 	Message string          `json:"message"`
@@ -20,4 +21,7 @@ type TransportEvent struct {
 	Time time.Time
 	// Formatted, serialized log payload
 	Payload []byte
 	// Structured entry for re-serializing sinks (chain links). Zero Time => absent
 	Entry LogEntry
 }
@@ -119,6 +119,7 @@ func (f *Flow) Process(entry core.LogEntry) (core.TransportEvent, bool) {
 	event := core.TransportEvent{
 		Time:    entry.Time,
 		Payload: formatted,
 		Entry:   entry, // Carry structured entry so chain sinks are format-independent
 	}
 	return event, true
@@ -160,3 +161,4 @@ func (f *Flow) GetStats() map[string]any {
 	return stats
 }
@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
 	"strconv"
 	"sync/atomic"
 	"time"
@@ -127,7 +128,8 @@ func (hg *HeartbeatGenerator) generateHeartbeat(t time.Time) core.TransportEvent
 		// SSE comment format - bypass formatter for this special case
 		if hg.config.IncludeStats {
 			beatNum := hg.beatCount.Load()
-			payload = []byte(": heartbeat " + t.Format(time.RFC3339) + " [#" + string(beatNum) + "]\n")
+			payload = []byte(": heartbeat " + t.Format(time.RFC3339) +
 				" [#" + strconv.FormatUint(beatNum, 10) + "]\n")
 		} else {
 			payload = []byte(": heartbeat " + t.Format(time.RFC3339) + "\n")
 		}
@@ -159,6 +161,7 @@ func (hg *HeartbeatGenerator) generateHeartbeat(t time.Time) core.TransportEvent
 	return core.TransportEvent{
 		Time:    t,
 		Payload: payload,
 		Entry:   entry, // heartbeats traverse chain links as structured entries
 	}
 }
@@ -89,6 +89,8 @@ func NewFormatterAdapter(cfg *config.FormatConfig) (*FormatterAdapter, error) {
 func (a *FormatterAdapter) Format(entry core.LogEntry) ([]byte, error) {
 	// Map logwisp LogEntry to formatter args
 	level := mapLevel(entry.Level)
 	// syslog-style origin prefix for chained entries
 	src := sourceLabel(entry)
 	// Build args based on whether we have structured fields
 	var args []any
@@ -101,18 +103,19 @@ func (a *FormatterAdapter) Format(entry core.LogEntry) ([]byte, error) {
 			args = []any{entry.Message, fields}
 			// Add structured flag to properly format fields as JSON object
 			effectiveFlags := a.flags | formatter.FlagStructuredJSON
-			return a.formatter.Format(effectiveFlags, entry.Time, level, entry.Source, args), nil
+			return a.formatter.Format(effectiveFlags, entry.Time, level, src, args), nil
 		}
 	}
 	// Simple message without fields
 	args = []any{entry.Message}
-	return a.formatter.Format(a.flags, entry.Time, level, entry.Source, args), nil
+	return a.formatter.Format(a.flags, entry.Time, level, src, args), nil
 }
 // FormatWithFlags allows custom flags for specific formatting needs
 func (a *FormatterAdapter) FormatWithFlags(entry core.LogEntry, customFlags int64) ([]byte, error) {
 	level := mapLevel(entry.Level)
 	src := sourceLabel(entry)
 	var args []any
 	if len(entry.Fields) > 0 {
@@ -127,7 +130,7 @@ func (a *FormatterAdapter) FormatWithFlags(entry core.LogEntry, customFlags int6
 		args = []any{entry.Message}
 	}
-	return a.formatter.Format(customFlags, entry.Time, level, entry.Source, args), nil
+	return a.formatter.Format(customFlags, entry.Time, level, src, args), nil
 }
 // Name returns formatter type
@@ -150,3 +153,12 @@ func mapLevel(level string) int64 {
 		return 0
 	}
 }
 // sourceLabel prefixes origin node onto source (syslog HOSTNAME + TAG convention)
 func sourceLabel(entry core.LogEntry) string {
 	if entry.Node == "" {
 		return entry.Source
 	}
 	return entry.Node + "/" + entry.Source
 }
@@ -41,13 +41,11 @@ type Pipeline struct {
 // PipelineStats contains runtime statistics for a pipeline
 type PipelineStats struct {
-	StartTime                      time.Time
+	StartTime                 time.Time
-	TotalEntriesProcessed          atomic.Uint64
+	TotalEntriesDroppedBySink atomic.Uint64
-	TotalEntriesDroppedByRateLimit atomic.Uint64
+	SourceStats               []source.SourceStats
-	TotalEntriesFiltered           atomic.Uint64
+	SinkStats                 []sink.SinkStats
-	SourceStats                    []source.SourceStats
+	FlowStats                 map[string]any
 	SinkStats                      []sink.SinkStats
 	FlowStats                      map[string]any
 }
 // NewPipeline creates a new pipeline with registry support
@@ -74,7 +72,6 @@ func NewPipeline(
 		cancel:   pipelineCancel,
 	}
 	// Create flow processor
 	// Create flow processor
 	flowProcessor, err := flow.NewFlow(cfg.Flow, logger)
 	if err != nil {
@@ -177,7 +174,7 @@ func (p *Pipeline) initSourceCapabilities(s source.Source, cfg config.PluginSour
 // initSinkCapabilities checks and injects optional capabilities
 func (p *Pipeline) initSinkCapabilities(s sink.Sink, cfg config.PluginSinkConfig) error {
-	// Initiate and activate source capabilities
+	// Initiate and activate sink capabilities
 	for _, c := range s.Capabilities() {
 		switch c {
 		// Network capabilities
@@ -203,48 +200,36 @@ func (p *Pipeline) run() {
 	defer p.logger.Info("msg", "Pipeline processing loop stopped", "pipeline", p.Config.Name)
 	var componentWg sync.WaitGroup
 	// Start a goroutine for each source to fan-in data
 	for _, src := range p.Sources {
 		componentWg.Add(1)
 		go func(s source.Source) {
 			defer componentWg.Done()
 			ch := s.Subscribe()
-			for {
+			// Range allows in-flight data to drain cleanly once Source.Stop() closes the channel
-				select {
+			for entry := range ch {
-				case entry, ok := <-ch:
+				if event, passed := p.Flow.Process(entry); passed {
-					if !ok {
+					// Use non-blocking dispatcher
-						return
+					p.dispatch(event)
 					}
 					// Process and distribute the log entry
 					if event, passed := p.Flow.Process(entry); passed {
 						// Fan-out to all sinks
 						for _, snk := range p.Sinks {
 							snk.Input() <- event
 						}
 					}
 				case <-p.ctx.Done():
 					return
 				}
 			}
 		}(src)
 	}
 	var hbWg sync.WaitGroup
 	// Start heartbeat generator if enabled
 	if heartbeatCh := p.Flow.StartHeartbeat(p.ctx); heartbeatCh != nil {
-		componentWg.Add(1)
+		hbWg.Add(1)
 		go func() {
-			defer componentWg.Done()
+			defer hbWg.Done()
 			for {
 				select {
 				case event, ok := <-heartbeatCh:
 					if !ok {
 						return
 					}
-					// Fan-out heartbeat to all sinks
+					// Use non-blocking dispatcher
-					for _, snk := range p.Sinks {
+					p.dispatch(event)
 						snk.Input() <- event
 					}
 				case <-p.ctx.Done():
 					return
 				}
@@ -253,6 +238,23 @@ func (p *Pipeline) run() {
 	}
 	componentWg.Wait()
 	// Terminate internal contexts (heartbeat) once flow is complete
 	p.cancel()
 	hbWg.Wait()
 }
 // dispatch performs a non-blocking send to all sinks.
 // A full/stalled sink must never block the run loop or starve sibling sinks.
 func (p *Pipeline) dispatch(event core.TransportEvent) {
 	for _, snk := range p.Sinks {
 		select {
 		case snk.Input() <- event:
 		default:
 			// Buffer full - drop to avoid deadlocking the pipeline
 			p.Stats.TotalEntriesDroppedBySink.Add(1)
 		}
 	}
 }
 // Start starts the pipeline operation and all its components including flow, sources, and sinks
@@ -294,10 +296,7 @@ func (p *Pipeline) Stop() error {
 	p.logger.Info("msg", "Stopping pipeline", "pipeline", p.Config.Name)
-	// Signal all components and the run loop to stop
+	// 1. Stop all sources concurrently to halt new data ingress and close their channels
 	p.cancel()
 	// Stop all sources concurrently to halt new data ingress
 	var sourceWg sync.WaitGroup
 	for _, src := range p.Sources {
 		sourceWg.Add(1)
@@ -308,10 +307,11 @@ func (p *Pipeline) Stop() error {
 	}
 	sourceWg.Wait()
-	// Wait for the run loop to finish processing and sending all in-flight data
+	// 2. Wait for the run loop to finish processing and sending all in-flight data
 	// run() inherently calls p.cancel() when the source channels are empty
 	p.wg.Wait()
-	// Stop all sinks concurrently now that no new data will be sent
+	// 3. Stop all sinks concurrently now that no new data will be sent
 	var sinkWg sync.WaitGroup
 	for _, s := range p.Sinks {
 		sinkWg.Add(1)
@@ -361,92 +361,82 @@ func (p *Pipeline) GetStats() map[string]any {
 		}
 	}()
-	// Collect source stats
+	// 1. Live collect source stats
-	sourceStats := make([]map[string]any, 0, len(p.Sources))
+	sources := make([]map[string]any, 0, len(p.Sources))
 	for _, src := range p.Sources {
 		if src == nil {
-			continue // Skip nil sources
+			continue
 		}
-
+		s := src.GetStats()
-		stats := src.GetStats()
+		sources = append(sources, map[string]any{
-		sourceStats = append(sourceStats, map[string]any{
+			"id":              s.ID,
-			"id":              stats.ID,
+			"type":            s.Type,
-			"type":            stats.Type,
+			"total_entries":   s.TotalEntries,
-			"total_entries":   stats.TotalEntries,
+			"dropped_entries": s.DroppedEntries,
-			"dropped_entries": stats.DroppedEntries,
+			"start_time":      s.StartTime,
-			"start_time":      stats.StartTime,
+			"last_entry_time": s.LastEntryTime,
-			"last_entry_time": stats.LastEntryTime,
+			"details":         s.Details,
 			"details":         stats.Details,
 		})
 	}
-	// Collect sink stats
+	// 2. Live collect sink stats
-	sinkStats := make([]map[string]any, 0, len(p.Sinks))
+	sinks := make([]map[string]any, 0, len(p.Sinks))
-	for _, s := range p.Sinks {
+	for _, snk := range p.Sinks {
-		if s == nil {
+		if snk == nil {
-			continue // Skip nil sinks
+			continue
 		}
-
+		s := snk.GetStats()
-		stats := s.GetStats()
+		sinks = append(sinks, map[string]any{
-		sinkStats = append(sinkStats, map[string]any{
+			"id":                 s.ID,
-			"id":                 stats.ID,
+			"type":               s.Type,
-			"type":               stats.Type,
+			"total_processed":    s.TotalProcessed,
-			"total_processed":    stats.TotalProcessed,
+			"active_connections": s.ActiveConnections,
-			"active_connections": stats.ActiveConnections,
+			"start_time":         s.StartTime,
-			"start_time":         stats.StartTime,
+			"last_processed":     s.LastProcessed,
-			"last_processed":     stats.LastProcessed,
+			"details":            s.Details,
 			"details":            stats.Details,
 		})
 	}
-	// Get flow stats
+	// 3. Collect flow stats and calculate filtered total
 	var flowStats map[string]any
 	var totalFiltered uint64
 	var totalProcessed uint64
 	if p.Flow != nil {
 		flowStats = p.Flow.GetStats()
-		// Extract total_filtered from flow for top-level visibility
+
 		// Map the top-level processed counter directly from Flow's source of truth
 		if tp, ok := flowStats["total_processed"].(uint64); ok {
 			totalProcessed = tp
 		}
 		// Calculate total dropped specifically by the filter chain
 		if filters, ok := flowStats["filters"].(map[string]any); ok {
 			if totalPassed, ok := filters["total_passed"].(uint64); ok {
-				if totalProcessed, ok := filters["total_processed"].(uint64); ok {
+				if tProc, ok := filters["total_processed"].(uint64); ok {
-					totalFiltered = totalProcessed - totalPassed
+					totalFiltered = tProc - totalPassed
 				}
 			}
 		}
 	}
 	// 4. Calculate Uptime
 	var uptime int
 	if p.running.Load() && !p.Stats.StartTime.IsZero() {
 		uptime = int(time.Since(p.Stats.StartTime).Seconds())
 	}
 	return map[string]any{
-		"name":            p.Config.Name,
+		"name":                  p.Config.Name,
-		"running":         p.running.Load(),
+		"running":               p.running.Load(),
-		"uptime_seconds":  uptime,
+		"uptime_seconds":        uptime,
-		"total_processed": p.Stats.TotalEntriesProcessed.Load(),
+		"total_processed":       totalProcessed,
-		"total_filtered":  totalFiltered,
+		"total_filtered":        totalFiltered,
-		"source_count":    len(p.Sources),
+		"total_dropped_by_sink": p.Stats.TotalEntriesDroppedBySink.Load(),
-		"sources":         sourceStats,
+		"source_count":          len(p.Sources),
-		"sink_count":      len(p.Sinks),
+		"sources":               sources,
-		"sinks":           sinkStats,
+		"sink_count":            len(p.Sinks),
-		"flow":            flowStats,
+		"sinks":                 sinks,
 		"flow":                  flowStats,
 	}
 }
 // TODO: incomplete implementation
 // startStatsUpdater runs a periodic stats updater
 func (p *Pipeline) startStatsUpdater(ctx context.Context) {
 	go func() {
 		ticker := time.NewTicker(core.ServiceStatsUpdateInterval)
 		defer ticker.Stop()
 		for {
 			select {
 			case <-ctx.Done():
 				return
 			case <-ticker.C:
 				// Periodic stats updates if needed
 			}
 		}
 	}()
 }
@@ -34,16 +34,6 @@ type PluginMetadata struct {
 	MaxInstances int // 0 = unlimited, 1 = single instance only
 }
 // // global variables holding available source and sink plugins
 // var (
 // 	sourceFactories map[string]SourceFactory
 // 	sinkFactories   map[string]SinkFactory
 // 	sourceMetadata  map[string]*PluginMetadata
 // 	sinkMetadata    map[string]*PluginMetadata
 // 	mu              sync.RWMutex
 // 	// once            sync.Once
 // )
 // registry encapsulates all plugin factories with lazy initialization
 type registry struct {
 	sourceFactories map[string]SourceFactory
@@ -71,11 +61,6 @@ func getRegistry() *registry {
 	return globalRegistry
 }
 // func init() {
 // 	sourceFactories = make(map[string]SourceFactory)
 // 	sinkFactories = make(map[string]SinkFactory)
 // }
 // RegisterSource registers a source factory function
 func RegisterSource(name string, constructor SourceFactory) error {
 	r := getRegistry()
@@ -170,16 +170,16 @@ func (h *HTTPSink) Start(ctx context.Context) error {
 	addr := fmt.Sprintf("%s:%d", h.config.Host, h.config.Port)
-	errChan := make(chan error, 1)
+	ln, err := net.Listen("tcp4", addr)
 	if err != nil {
 		return fmt.Errorf("http sink bind %s: %w", addr, err)
 	}
 	go func() {
-		h.logger.Info("msg", "HTTP server starting",
+		if err := h.server.Serve(ln); err != nil {
-			"component", "http_sink",
+			h.logger.Error("msg", "HTTP server terminated",
-			"instance_id", h.id,
+				"component", "http_sink",
-			"address", addr)
+				"instance_id", h.id,
-
+				"error", err)
 		err := h.server.ListenAndServe(addr)
 		if err != nil {
 			errChan <- err
 		}
 	}()
@@ -193,18 +193,12 @@ func (h *HTTPSink) Start(ctx context.Context) error {
 		}
 	}()
-	// Check if server started
+	h.logger.Info("msg", "HTTP server started",
-	select {
+		"component", "http_sink",
-	case err := <-errChan:
+		"instance_id", h.id,
-		return err
+		"host", h.config.Host,
-	case <-time.After(HttpServerStartTimeout):
+		"port", h.config.Port)
-		h.logger.Info("msg", "HTTP server started",
+	return nil
 			"component", "http_sink",
 			"instance_id", h.id,
 			"host", h.config.Host,
 			"port", h.config.Port)
 		return nil
 	}
 }
 // Stop gracefully shuts down the HTTP server and all client connections
@@ -356,6 +350,8 @@ func (h *HTTPSink) requestHandler(ctx *fasthttp.RequestCtx) {
 	remoteAddr := ctx.RemoteAddr()
 	if tcpAddr, ok := remoteAddr.(*net.TCPAddr); ok {
 		if tcpAddr.IP.To4() == nil {
 			h.logger.Debug("msg", "IPv6 connection rejected",
 				"component", "http_sink", "remote_addr", remoteAddr.String())
 			ctx.SetConnectionClose()
 			return
 		}
@@ -414,8 +410,6 @@ func (h *HTTPSink) handleStream(ctx *fasthttp.RequestCtx) {
 			"client_id", clientID,
 			"active_clients", connectCount)
 		h.wg.Add(1)
 		defer func() {
 			disconnectCount := h.activeClients.Add(-1)
 			h.logger.Debug("msg", "HTTP client disconnected",
@@ -431,7 +425,6 @@ func (h *HTTPSink) handleStream(ctx *fasthttp.RequestCtx) {
 			}
 			h.proxy.RemoveSession(sess.ID)
 			h.wg.Done()
 		}()
 		// Send connected event with metadata
@@ -0,0 +1,422 @@
 package httpchain
 import (
 	"bytes"
 	"context"
 	"encoding/json"
 	"fmt"
 	"io"
 	"net"
 	"net/http"
 	"os"
 	"strconv"
 	"strings"
 	"sync"
 	"sync/atomic"
 	"time"
 	"logwisp/src/internal/chain"
 	"logwisp/src/internal/config"
 	"logwisp/src/internal/core"
 	"logwisp/src/internal/plugin"
 	"logwisp/src/internal/session"
 	"logwisp/src/internal/sink"
 	lconfig "github.com/lixenwraith/config"
 	"github.com/lixenwraith/log"
 )
 func init() {
 	if err := plugin.RegisterSink("http_chain", NewHTTPChainSinkPlugin); err != nil {
 		panic(fmt.Sprintf("failed to register http_chain sink: %v", err))
 	}
 }
 const (
 	DefaultHTTPChainSinkBufferSize       = 1000
 	DefaultHTTPChainSinkIngestPath       = "/ingest"
 	DefaultHTTPChainSinkMaxBatchCount    = 100
 	DefaultHTTPChainSinkMaxBatchBytes    = 1024 * 1024
 	DefaultHTTPChainSinkFlushIntervalMS  = 1000
 	DefaultHTTPChainSinkRequestTimeoutMS = 10000
 	DefaultHTTPChainSinkBackoffMinMS     = 500
 	DefaultHTTPChainSinkBackoffMaxMS     = 30000
 )
 // HTTPChainSink batches structured entries and posts NDJSON to a downstream
 // http_chain source. Delivery is at-least-once per batch.
 type HTTPChainSink struct {
 	id      string
 	proxy   *session.Proxy
 	session *session.Session
 	config  *config.HTTPChainSinkOptions
 	node string
 	url  string
 	client *http.Client
 	input  chan core.TransportEvent
 	logger *log.Logger
 	// Batch state owned exclusively by run loop goroutine
 	batch      bytes.Buffer
 	batchCount int64
 	reqTimeout time.Duration
 	done       chan struct{}
 	wg         sync.WaitGroup
 	startTime  time.Time
 	totalProcessed atomic.Uint64
 	batchesSent    atomic.Uint64
 	requestErrors  atomic.Uint64
 	droppedBatches atomic.Uint64
 	synthesized    atomic.Uint64
 	lastProcessed  atomic.Value // time.Time
 }
 // NewHTTPChainSinkPlugin creates an http_chain sink through plugin factory
 func NewHTTPChainSinkPlugin(
 	id string,
 	configMap map[string]any,
 	logger *log.Logger,
 	proxy *session.Proxy,
 ) (sink.Sink, error) {
 	opts := &config.HTTPChainSinkOptions{}
 	if err := lconfig.ScanMap(configMap, opts); err != nil {
 		return nil, fmt.Errorf("failed to parse config: %w", err)
 	}
 	if err := lconfig.NonEmpty(opts.Host); err != nil {
 		return nil, fmt.Errorf("host: %w", err)
 	}
 	if err := lconfig.Port(opts.Port); err != nil {
 		return nil, fmt.Errorf("port: %w", err)
 	}
 	if opts.IngestPath == "" {
 		opts.IngestPath = DefaultHTTPChainSinkIngestPath
 	} else if !strings.HasPrefix(opts.IngestPath, "/") {
 		return nil, fmt.Errorf("ingest_path: must start with '/'")
 	}
 	if opts.BufferSize <= 0 {
 		opts.BufferSize = DefaultHTTPChainSinkBufferSize
 	}
 	if opts.MaxBatchCount <= 0 {
 		opts.MaxBatchCount = DefaultHTTPChainSinkMaxBatchCount
 	}
 	if opts.MaxBatchBytes <= 0 {
 		opts.MaxBatchBytes = DefaultHTTPChainSinkMaxBatchBytes
 	}
 	if opts.FlushIntervalMS <= 0 {
 		opts.FlushIntervalMS = DefaultHTTPChainSinkFlushIntervalMS
 	}
 	if opts.RequestTimeoutMS <= 0 {
 		opts.RequestTimeoutMS = DefaultHTTPChainSinkRequestTimeoutMS
 	}
 	if opts.BackoffMinMS <= 0 {
 		opts.BackoffMinMS = DefaultHTTPChainSinkBackoffMinMS
 	}
 	if opts.BackoffMaxMS < opts.BackoffMinMS {
 		opts.BackoffMaxMS = DefaultHTTPChainSinkBackoffMaxMS
 	}
 	node := opts.Node
 	if node == "" {
 		if hn, err := os.Hostname(); err == nil {
 			node = hn
 		} else {
 			node = "unknown"
 		}
 	}
 	addr := net.JoinHostPort(opts.Host, strconv.FormatInt(opts.Port, 10))
 	transport := &http.Transport{
 		DialContext: func(ctx context.Context, network, address string) (net.Conn, error) {
 			// IPv4-only, aligns with tcp/http sinks
 			d := net.Dialer{}
 			return d.DialContext(ctx, "tcp4", address)
 		},
 		MaxIdleConnsPerHost: 2,
 		IdleConnTimeout:     90 * time.Second,
 		DisableCompression:  true,
 		// Future: TLSClientConfig; HTTP/2 via ALPN once TLS lands
 	}
 	t := &HTTPChainSink{
 		id:     id,
 		proxy:  proxy,
 		config: opts,
 		node:   node,
 		// Future: "https" scheme with TLS
 		url:        "http://" + addr + opts.IngestPath,
 		client:     &http.Client{Transport: transport},
 		input:      make(chan core.TransportEvent, opts.BufferSize),
 		done:       make(chan struct{}),
 		logger:     logger,
 		reqTimeout: time.Duration(opts.RequestTimeoutMS) * time.Millisecond,
 	}
 	t.lastProcessed.Store(time.Time{})
 	t.session = proxy.CreateSession(
 		"http_chain://"+addr,
 		map[string]any{
 			"instance_id": id,
 			"type":        "http_chain",
 			"target":      t.url,
 			"node":        node,
 		},
 	)
 	logger.Info("msg", "HTTP chain sink initialized",
 		"component", "http_chain_sink",
 		"instance_id", id,
 		"target", t.url,
 		"node", node)
 	return t, nil
 }
 // Capabilities returns supported capabilities
 func (t *HTTPChainSink) Capabilities() []core.Capability {
 	// CapTLS/CapAuth added when transport security lands
 	return []core.Capability{
 		core.CapSessionAware,
 	}
 }
 // Input returns the channel for sending transport events
 func (t *HTTPChainSink) Input() chan<- core.TransportEvent {
 	return t.input
 }
 // Start launches the batching loop; downstream availability is not required
 func (t *HTTPChainSink) Start(ctx context.Context) error {
 	t.startTime = time.Now()
 	t.wg.Add(1)
 	go t.runLoop(ctx)
 	t.logger.Info("msg", "HTTP chain sink started",
 		"component", "http_chain_sink",
 		"instance_id", t.id,
 		"target", t.url)
 	return nil
 }
 // Stop terminates the loop. Worst case: one in-flight request timeout plus
 // one final-flush request timeout.
 func (t *HTTPChainSink) Stop() {
 	t.logger.Info("msg", "Stopping HTTP chain sink",
 		"component", "http_chain_sink",
 		"instance_id", t.id)
 	close(t.done)
 	t.wg.Wait()
 	t.client.CloseIdleConnections()
 	if t.session != nil {
 		t.proxy.RemoveSession(t.session.ID)
 	}
 	t.logger.Info("msg", "HTTP chain sink stopped",
 		"component", "http_chain_sink",
 		"instance_id", t.id,
 		"total_processed", t.totalProcessed.Load())
 }
 // GetStats returns sink statistics
 func (t *HTTPChainSink) GetStats() sink.SinkStats {
 	lastProc, _ := t.lastProcessed.Load().(time.Time)
 	return sink.SinkStats{
 		ID:             t.id,
 		Type:           "http_chain",
 		TotalProcessed: t.totalProcessed.Load(),
 		StartTime:      t.startTime,
 		LastProcessed:  lastProc,
 		Details: map[string]any{
 			"target":          t.url,
 			"node":            t.node,
 			"batches_sent":    t.batchesSent.Load(),
 			"request_errors":  t.requestErrors.Load(),
 			"dropped_batches": t.droppedBatches.Load(),
 			"synthesized":     t.synthesized.Load(),
 		},
 	}
 }
 // runLoop batches events and flushes on size or interval
 func (t *HTTPChainSink) runLoop(ctx context.Context) {
 	defer t.wg.Done()
 	// Fold done channel into a context for request/backoff interruption
 	runCtx, cancel := context.WithCancel(ctx)
 	defer cancel()
 	go func() {
 		select {
 		case <-t.done:
 			cancel()
 		case <-runCtx.Done():
 		}
 	}()
 	ticker := time.NewTicker(time.Duration(t.config.FlushIntervalMS) * time.Millisecond)
 	defer ticker.Stop()
 	for {
 		select {
 		case <-runCtx.Done():
 			t.finalFlush()
 			return
 		case <-ticker.C:
 			if t.batchCount > 0 && !t.flush(runCtx) {
 				t.finalFlush()
 				return
 			}
 		case event, ok := <-t.input:
 			if !ok {
 				t.finalFlush()
 				return
 			}
 			t.append(event)
 			if t.batchCount >= t.config.MaxBatchCount ||
 				int64(t.batch.Len()) >= t.config.MaxBatchBytes {
 				if !t.flush(runCtx) {
 					t.finalFlush()
 					return
 				}
 			}
 		}
 	}
 }
 // append serializes one event into the pending batch
 func (t *HTTPChainSink) append(event core.TransportEvent) {
 	entry, synthesized := chain.EntryFromEvent(event, t.node, t.id)
 	if synthesized {
 		t.synthesized.Add(1)
 	}
 	line, err := json.Marshal(entry)
 	if err != nil {
 		// Non-transient: drop entry
 		t.logger.Error("msg", "Failed to marshal chain entry",
 			"component", "http_chain_sink",
 			"error", err)
 		return
 	}
 	t.batch.Write(line)
 	t.batch.WriteByte('\n')
 	t.batchCount++
 }
 // flush delivers the pending batch, retrying transient failures with backoff.
 // Returns false when shutdown interrupts delivery; undelivered batch is dropped
 // by finalFlush semantics (batch already consumed here).
 func (t *HTTPChainSink) flush(ctx context.Context) bool {
 	body := bytes.Clone(t.batch.Bytes())
 	count := t.batchCount
 	t.batch.Reset()
 	t.batchCount = 0
 	failures := 0
 	for {
 		if failures > 0 && !t.waitBackoff(ctx, failures) {
 			t.droppedBatches.Add(1)
 			return false
 		}
 		transient, err := t.post(ctx, body)
 		if err == nil {
 			t.batchesSent.Add(1)
 			t.totalProcessed.Add(uint64(count))
 			t.lastProcessed.Store(time.Now())
 			t.proxy.UpdateActivity(t.session.ID)
 			return true
 		}
 		t.requestErrors.Add(1)
 		if !transient {
 			t.droppedBatches.Add(1)
 			t.logger.Error("msg", "Chain batch rejected, dropping",
 				"component", "http_chain_sink",
 				"target", t.url,
 				"entries", count,
 				"error", err)
 			return true
 		}
 		if ctx.Err() != nil {
 			t.droppedBatches.Add(1)
 			return false
 		}
 		failures++
 		t.logger.Warn("msg", "Chain batch delivery failed",
 			"component", "http_chain_sink",
 			"target", t.url,
 			"attempt", failures,
 			"error", err)
 	}
 }
 // finalFlush best-effort delivers the pending batch during shutdown (single attempt)
 func (t *HTTPChainSink) finalFlush() {
 	if t.batchCount == 0 {
 		return
 	}
 	fctx, cancel := context.WithTimeout(context.Background(), t.reqTimeout)
 	defer cancel()
 	count := t.batchCount
 	if _, err := t.post(fctx, t.batch.Bytes()); err != nil {
 		t.droppedBatches.Add(1)
 		t.logger.Warn("msg", "Final chain batch dropped on shutdown",
 			"component", "http_chain_sink",
 			"entries", count,
 			"error", err)
 		return
 	}
 	t.batchesSent.Add(1)
 	t.totalProcessed.Add(uint64(count))
 }
 // post sends one NDJSON batch; transient=true marks retryable failures
 func (t *HTTPChainSink) post(ctx context.Context, body []byte) (transient bool, err error) {
 	reqCtx, cancel := context.WithTimeout(ctx, t.reqTimeout)
 	defer cancel()
 	req, err := http.NewRequestWithContext(reqCtx, http.MethodPost, t.url, bytes.NewReader(body))
 	if err != nil {
 		return false, err
 	}
 	req.Header.Set("Content-Type", chain.ContentTypeNDJSON)
 	req.Header.Set(chain.HeaderProtocol, strconv.Itoa(chain.ProtocolVersion))
 	req.Header.Set(chain.HeaderNode, t.node)
 	// Future: Authorization header for auth
 	resp, err := t.client.Do(req)
 	if err != nil {
 		return true, err
 	}
 	defer resp.Body.Close()
 	// Drain for connection reuse
 	io.Copy(io.Discard, resp.Body)
 	switch {
 	case resp.StatusCode >= 200 && resp.StatusCode < 300:
 		return false, nil
 	case resp.StatusCode == http.StatusRequestTimeout,
 		resp.StatusCode == http.StatusTooManyRequests,
 		resp.StatusCode >= 500:
 		return true, fmt.Errorf("status %s", resp.Status)
 	default:
 		return false, fmt.Errorf("status %s", resp.Status)
 	}
 }
 // waitBackoff sleeps for the computed delay, interruptible by shutdown
 func (t *HTTPChainSink) waitBackoff(ctx context.Context, failures int) bool {
 	minD := time.Duration(t.config.BackoffMinMS) * time.Millisecond
 	maxD := time.Duration(t.config.BackoffMaxMS) * time.Millisecond
 	timer := time.NewTimer(chain.BackoffDelay(minD, maxD, failures))
 	defer timer.Stop()
 	select {
 	case <-timer.C:
 		return true
 	case <-ctx.Done():
 		return false
 	}
 }
@@ -40,6 +40,7 @@ type TCPSink struct {
 	server   *tcpServer
 	engine   *gnet.Engine
 	engineMu sync.Mutex
 	booted   chan struct{}
 	// Application
 	input  chan core.TransportEvent
@@ -63,7 +64,7 @@ type TCPSink struct {
 const (
 	// Server lifecycle
-	TCPServerStartTimeout    = 100 * time.Millisecond
+	TCPServerStartTimeout    = 2 * time.Second
 	TCPServerShutdownTimeout = 2 * time.Second
 	// Connection management
@@ -151,6 +152,8 @@ func (t *TCPSink) Start(ctx context.Context) error {
 		sink:    t,
 		clients: make(map[gnet.Conn]*tcpClient),
 	}
 	// Fresh channel per Start
 	t.booted = make(chan struct{})
 	t.startTime = time.Now()
@@ -213,12 +216,25 @@ func (t *TCPSink) Start(ctx context.Context) error {
 		close(t.done)
 		t.wg.Wait()
 		return err
-	case <-time.After(TCPServerStartTimeout):
+	// Bind confirmation via OnBoot
 	case <-t.booted:
 		t.logger.Info("msg", "TCP server started",
 			"component", "tcp_sink",
 			"instance_id", t.id,
 			"port", t.config.Port)
 		return nil
 	// Timeout failure
 	case <-time.After(TCPServerStartTimeout):
 		t.engineMu.Lock()
 		if t.engine != nil {
 			stopCtx, cancel := context.WithTimeout(context.Background(), TCPServerShutdownTimeout)
 			(*t.engine).Stop(stopCtx)
 			cancel()
 		}
 		t.engineMu.Unlock()
 		close(t.done)
 		t.wg.Wait()
 		return fmt.Errorf("tcp sink start timeout on %s", addr)
 	}
 }
@@ -309,6 +325,9 @@ func (s *tcpServer) OnBoot(eng gnet.Engine) gnet.Action {
 	s.sink.engine = &eng
 	s.sink.engineMu.Unlock()
 	// Listener is bound at this point; unblock Start
 	close(s.sink.booted)
 	s.sink.logger.Debug("msg", "TCP server booted",
 		"component", "tcp_sink",
 		"instance_id", s.sink.id)
@@ -409,8 +428,10 @@ func (s *tcpServer) OnTraffic(c gnet.Conn) gnet.Action {
 		s.sink.proxy.UpdateActivity(client.sessionID)
 	}
-	// TCP sink doesn't expect data from clients, discard
+	// TCP sink doesn't expect data from clients, discard safely
-	c.Discard(-1)
+	if bufLen := c.InboundBuffered(); bufLen > 0 {
 		c.Next(bufLen)
 	}
 	return gnet.None
 }
@@ -0,0 +1,398 @@
 package tcpchain
 import (
 	"context"
 	"encoding/json"
 	"fmt"
 	"math/rand/v2"
 	"net"
 	"os"
 	"strconv"
 	"sync"
 	"sync/atomic"
 	"time"
 	"logwisp/src/internal/chain"
 	"logwisp/src/internal/config"
 	"logwisp/src/internal/core"
 	"logwisp/src/internal/plugin"
 	"logwisp/src/internal/session"
 	"logwisp/src/internal/sink"
 	lconfig "github.com/lixenwraith/config"
 	"github.com/lixenwraith/log"
 )
 func init() {
 	if err := plugin.RegisterSink("tcp_chain", NewTCPChainSinkPlugin); err != nil {
 		panic(fmt.Sprintf("failed to register tcp_chain sink: %v", err))
 	}
 }
 const (
 	DefaultChainSinkBufferSize        = 1000
 	DefaultChainSinkDialTimeoutMS     = 5000
 	DefaultChainSinkWriteTimeoutMS    = 5000
 	DefaultChainSinkBackoffMinMS      = 500
 	DefaultChainSinkBackoffMaxMS      = 30000
 	DefaultChainSinkKeepAlivePeriodMS = 30000
 )
 // TCPChainSink forwards structured entries to a downstream tcp_chain source
 type TCPChainSink struct {
 	id      string
 	proxy   *session.Proxy
 	session *session.Session
 	config  *config.TCPChainSinkOptions
 	node      string
 	addr      string
 	helloLine []byte
 	input  chan core.TransportEvent
 	logger *log.Logger
 	// conn owned exclusively by run loop goroutine
 	conn          net.Conn
 	everConnected bool
 	dialTimeout   time.Duration
 	writeTimeout  time.Duration
 	done      chan struct{}
 	wg        sync.WaitGroup
 	startTime time.Time
 	totalProcessed atomic.Uint64
 	writeErrors    atomic.Uint64
 	reconnects     atomic.Uint64
 	synthesized    atomic.Uint64
 	connected      atomic.Bool
 	lastProcessed  atomic.Value // time.Time
 }
 // NewTCPChainSinkPlugin creates a tcp_chain sink through plugin factory
 func NewTCPChainSinkPlugin(
 	id string,
 	configMap map[string]any,
 	logger *log.Logger,
 	proxy *session.Proxy,
 ) (sink.Sink, error) {
 	opts := &config.TCPChainSinkOptions{
 		KeepAlive: true,
 	}
 	if err := lconfig.ScanMap(configMap, opts); err != nil {
 		return nil, fmt.Errorf("failed to parse config: %w", err)
 	}
 	if err := lconfig.NonEmpty(opts.Host); err != nil {
 		return nil, fmt.Errorf("host: %w", err)
 	}
 	if err := lconfig.Port(opts.Port); err != nil {
 		return nil, fmt.Errorf("port: %w", err)
 	}
 	if opts.BufferSize <= 0 {
 		opts.BufferSize = DefaultChainSinkBufferSize
 	}
 	if opts.DialTimeoutMS <= 0 {
 		opts.DialTimeoutMS = DefaultChainSinkDialTimeoutMS
 	}
 	if opts.WriteTimeoutMS <= 0 {
 		opts.WriteTimeoutMS = DefaultChainSinkWriteTimeoutMS
 	}
 	if opts.BackoffMinMS <= 0 {
 		opts.BackoffMinMS = DefaultChainSinkBackoffMinMS
 	}
 	if opts.BackoffMaxMS < opts.BackoffMinMS {
 		opts.BackoffMaxMS = DefaultChainSinkBackoffMaxMS
 	}
 	if opts.KeepAlivePeriodMS <= 0 {
 		opts.KeepAlivePeriodMS = DefaultChainSinkKeepAlivePeriodMS
 	}
 	node := opts.Node
 	if node == "" {
 		if hn, err := os.Hostname(); err == nil {
 			node = hn
 		} else {
 			node = "unknown"
 		}
 	}
 	helloLine, err := chain.EncodeHello(node)
 	if err != nil {
 		return nil, fmt.Errorf("hello: %w", err)
 	}
 	t := &TCPChainSink{
 		id:           id,
 		proxy:        proxy,
 		config:       opts,
 		node:         node,
 		addr:         net.JoinHostPort(opts.Host, strconv.FormatInt(opts.Port, 10)),
 		helloLine:    helloLine,
 		input:        make(chan core.TransportEvent, opts.BufferSize),
 		done:         make(chan struct{}),
 		logger:       logger,
 		dialTimeout:  time.Duration(opts.DialTimeoutMS) * time.Millisecond,
 		writeTimeout: time.Duration(opts.WriteTimeoutMS) * time.Millisecond,
 	}
 	t.lastProcessed.Store(time.Time{})
 	t.session = proxy.CreateSession(
 		"tcp_chain://"+t.addr,
 		map[string]any{
 			"instance_id": id,
 			"type":        "tcp_chain",
 			"target":      t.addr,
 			"node":        node,
 		},
 	)
 	logger.Info("msg", "TCP chain sink initialized",
 		"component", "tcp_chain_sink",
 		"instance_id", id,
 		"target", t.addr,
 		"node", node)
 	return t, nil
 }
 // Capabilities returns supported capabilities
 func (t *TCPChainSink) Capabilities() []core.Capability {
 	// CapTLS/CapAuth added when transport security lands
 	return []core.Capability{
 		core.CapSessionAware,
 	}
 }
 // Input returns the channel for sending transport events
 func (t *TCPChainSink) Input() chan<- core.TransportEvent {
 	return t.input
 }
 // Start launches the forwarding loop; connection is established lazily so
 // pipeline start does not depend on downstream availability
 func (t *TCPChainSink) Start(ctx context.Context) error {
 	t.startTime = time.Now()
 	t.wg.Add(1)
 	go t.runLoop(ctx)
 	t.logger.Info("msg", "TCP chain sink started",
 		"component", "tcp_chain_sink",
 		"instance_id", t.id,
 		"target", t.addr)
 	return nil
 }
 // Stop terminates the forwarding loop. Worst-case latency: one write timeout
 // plus one backoff wait (both interruptible or bounded).
 func (t *TCPChainSink) Stop() {
 	t.logger.Info("msg", "Stopping TCP chain sink",
 		"component", "tcp_chain_sink",
 		"instance_id", t.id)
 	close(t.done)
 	t.wg.Wait()
 	if t.session != nil {
 		t.proxy.RemoveSession(t.session.ID)
 	}
 	t.logger.Info("msg", "TCP chain sink stopped",
 		"component", "tcp_chain_sink",
 		"instance_id", t.id,
 		"total_processed", t.totalProcessed.Load())
 }
 // GetStats returns sink statistics
 func (t *TCPChainSink) GetStats() sink.SinkStats {
 	lastProc, _ := t.lastProcessed.Load().(time.Time)
 	var active int64
 	if t.connected.Load() {
 		active = 1
 	}
 	return sink.SinkStats{
 		ID:                t.id,
 		Type:              "tcp_chain",
 		TotalProcessed:    t.totalProcessed.Load(),
 		ActiveConnections: active,
 		StartTime:         t.startTime,
 		LastProcessed:     lastProc,
 		Details: map[string]any{
 			"target":       t.addr,
 			"node":         t.node,
 			"connected":    t.connected.Load(),
 			"reconnects":   t.reconnects.Load(),
 			"write_errors": t.writeErrors.Load(),
 			"synthesized":  t.synthesized.Load(),
 		},
 	}
 }
 // runLoop consumes transport events and forwards them downstream
 func (t *TCPChainSink) runLoop(ctx context.Context) {
 	defer t.wg.Done()
 	defer t.closeConn()
 	for {
 		select {
 		case <-ctx.Done():
 			return
 		case <-t.done:
 			return
 		case event, ok := <-t.input:
 			if !ok {
 				return
 			}
 			entry, synthesized := chain.EntryFromEvent(event, t.node, t.id)
 			if synthesized {
 				t.synthesized.Add(1)
 			}
 			line, err := json.Marshal(entry)
 			if err != nil {
 				// Non-transient: drop
 				t.logger.Error("msg", "Failed to marshal chain entry",
 					"component", "tcp_chain_sink",
 					"error", err)
 				continue
 			}
 			if !t.deliver(ctx, append(line, '\n')) {
 				return // shutdown during retry
 			}
 			t.totalProcessed.Add(1)
 			t.lastProcessed.Store(time.Now())
 			t.proxy.UpdateActivity(t.session.ID)
 		}
 	}
 }
 // toEntry extracts the structured entry, stamping node identity at first hop
 func (t *TCPChainSink) toEntry(event core.TransportEvent) core.LogEntry {
 	entry := event.Entry
 	if entry.Time.IsZero() {
 		// Defensive: event without structured entry, wrap formatted payload
 		t.synthesized.Add(1)
 		entry = core.LogEntry{
 			Time:    event.Time,
 			Source:  t.id,
 			Message: string(event.Payload),
 		}
 	}
 	if entry.Node == "" {
 		entry.Node = t.node
 	}
 	return entry
 }
 // deliver writes one line, holding it across reconnects until sent or shutdown.
 // Backpressure during outage propagates to the pipeline dispatch drop counter.
 func (t *TCPChainSink) deliver(ctx context.Context, line []byte) bool {
 	failures := 0
 	for {
 		if t.conn == nil {
 			if failures > 0 && !t.waitBackoff(ctx, failures) {
 				return false
 			}
 			if err := t.connect(ctx); err != nil {
 				if ctx.Err() != nil {
 					return false
 				}
 				failures++
 				t.logger.Debug("msg", "Chain connect failed",
 					"component", "tcp_chain_sink",
 					"target", t.addr,
 					"attempt", failures,
 					"error", err)
 				continue
 			}
 		}
 		t.conn.SetWriteDeadline(time.Now().Add(t.writeTimeout))
 		if _, err := t.conn.Write(line); err != nil {
 			t.writeErrors.Add(1)
 			failures++
 			t.logger.Warn("msg", "Chain write failed",
 				"component", "tcp_chain_sink",
 				"target", t.addr,
 				"error", err)
 			t.closeConn()
 			continue
 		}
 		return true
 	}
 }
 // connect performs a single dial + hello attempt
 func (t *TCPChainSink) connect(ctx context.Context) error {
 	d := net.Dialer{Timeout: t.dialTimeout}
 	if t.config.KeepAlive {
 		d.KeepAliveConfig = net.KeepAliveConfig{
 			Enable: true,
 			Idle:   time.Duration(t.config.KeepAlivePeriodMS) * time.Millisecond,
 		}
 	}
 	// IPv4-only
 	conn, err := d.DialContext(ctx, "tcp4", t.addr)
 	if err != nil {
 		return err
 	}
 	conn.SetWriteDeadline(time.Now().Add(t.writeTimeout))
 	if _, err := conn.Write(t.helloLine); err != nil {
 		conn.Close()
 		return fmt.Errorf("hello: %w", err)
 	}
 	t.conn = conn
 	t.connected.Store(true)
 	if t.everConnected {
 		t.reconnects.Add(1)
 	}
 	t.everConnected = true
 	t.logger.Info("msg", "Chain link established",
 		"component", "tcp_chain_sink",
 		"target", t.addr,
 		"node", t.node)
 	return nil
 }
 // closeConn tears down the current connection (run loop goroutine only)
 func (t *TCPChainSink) closeConn() {
 	if t.conn != nil {
 		t.conn.Close()
 		t.conn = nil
 	}
 	t.connected.Store(false)
 }
 // waitBackoff sleeps for the computed delay, interruptible by shutdown
 func (t *TCPChainSink) waitBackoff(ctx context.Context, failures int) bool {
 	minD := time.Duration(t.config.BackoffMinMS) * time.Millisecond
 	maxD := time.Duration(t.config.BackoffMaxMS) * time.Millisecond
 	timer := time.NewTimer(chain.BackoffDelay(minD, maxD, failures))
 	defer timer.Stop()
 	select {
 	case <-timer.C:
 		return true
 	case <-ctx.Done():
 		return false
 	case <-t.done:
 		return false
 	}
 }
 // backoffDelay computes exponential backoff with ±20% jitter
 func (t *TCPChainSink) backoffDelay(failures int) time.Duration {
 	minD := time.Duration(t.config.BackoffMinMS) * time.Millisecond
 	maxD := time.Duration(t.config.BackoffMaxMS) * time.Millisecond
 	d := maxD
 	if failures < 63 {
 		if v := minD << uint(failures-1); v > 0 && v < maxD {
 			d = v
 		}
 	}
 	return d - d/5 + time.Duration(rand.Int64N(int64(2*d/5)+1))
 }
@@ -161,7 +161,7 @@ func (fs *FileSource) Stop() {
 	}
 	fs.wg.Wait()
-	fs.proxy.RemoveSession(fs.id)
+	fs.proxy.RemoveSession(fs.session.ID)
 	fs.mu.Lock()
 	for _, w := range fs.watchers {
@@ -361,3 +361,4 @@ func globToRegex(glob string) string {
 	regex = strings.ReplaceAll(regex, `\?`, `.`)
 	return "^" + regex + "$"
 }
@@ -0,0 +1,324 @@
 package httpchain
 import (
 	"bufio"
 	"context"
 	"errors"
 	"fmt"
 	"net"
 	"net/http"
 	"strconv"
 	"strings"
 	"sync"
 	"sync/atomic"
 	"time"
 	"logwisp/src/internal/chain"
 	"logwisp/src/internal/config"
 	"logwisp/src/internal/core"
 	"logwisp/src/internal/plugin"
 	"logwisp/src/internal/session"
 	"logwisp/src/internal/source"
 	lconfig "github.com/lixenwraith/config"
 	"github.com/lixenwraith/log"
 )
 func init() {
 	if err := plugin.RegisterSource("http_chain", NewHTTPChainSourcePlugin); err != nil {
 		panic(fmt.Sprintf("failed to register http_chain source: %v", err))
 	}
 }
 const (
 	DefaultHTTPChainSourceBufferSize    = 1000
 	DefaultHTTPChainSourceIngestPath    = "/ingest"
 	DefaultHTTPChainSourceMaxBodyBytes  = 8 * 1024 * 1024
 	DefaultHTTPChainSourceReadTimeoutMS = 30000
 	HTTPChainReadHeaderTimeout          = 10 * time.Second
 	HTTPChainServerShutdownTimeout      = 2 * time.Second
 )
 // HTTPChainSource accepts NDJSON batches from upstream http_chain sinks
 type HTTPChainSource struct {
 	id     string
 	proxy  *session.Proxy
 	config *config.HTTPChainSourceOptions
 	subscribers []chan core.LogEntry
 	server      *http.Server
 	logger      *log.Logger
 	// Session cache: one session per remote host + declared node
 	sessions   map[string]string // key -> sessionID
 	sessionsMu sync.Mutex
 	mu sync.RWMutex
 	startTime        time.Time
 	totalEntries     atomic.Uint64
 	droppedEntries   atomic.Uint64
 	parseErrors      atomic.Uint64
 	totalRequests    atomic.Uint64
 	rejectedRequests atomic.Uint64
 	lastEntryTime    atomic.Value // time.Time
 }
 // NewHTTPChainSourcePlugin creates an http_chain source through plugin factory
 func NewHTTPChainSourcePlugin(
 	id string,
 	configMap map[string]any,
 	logger *log.Logger,
 	proxy *session.Proxy,
 ) (source.Source, error) {
 	opts := &config.HTTPChainSourceOptions{
 		Host:      "0.0.0.0",
 		TrustNode: true,
 	}
 	if err := lconfig.ScanMap(configMap, opts); err != nil {
 		return nil, fmt.Errorf("failed to parse config: %w", err)
 	}
 	if err := lconfig.Port(opts.Port); err != nil {
 		return nil, fmt.Errorf("port: %w", err)
 	}
 	if opts.IngestPath == "" {
 		opts.IngestPath = DefaultHTTPChainSourceIngestPath
 	} else if !strings.HasPrefix(opts.IngestPath, "/") {
 		return nil, fmt.Errorf("ingest_path: must start with '/'")
 	}
 	if opts.BufferSize <= 0 {
 		opts.BufferSize = DefaultHTTPChainSourceBufferSize
 	}
 	if opts.MaxBodyBytes <= 0 {
 		opts.MaxBodyBytes = DefaultHTTPChainSourceMaxBodyBytes
 	}
 	if opts.ReadTimeoutMS <= 0 {
 		opts.ReadTimeoutMS = DefaultHTTPChainSourceReadTimeoutMS
 	}
 	s := &HTTPChainSource{
 		id:          id,
 		proxy:       proxy,
 		config:      opts,
 		subscribers: make([]chan core.LogEntry, 0),
 		sessions:    make(map[string]string),
 		logger:      logger,
 	}
 	s.lastEntryTime.Store(time.Time{})
 	logger.Info("msg", "HTTP chain source initialized",
 		"component", "http_chain_source",
 		"instance_id", id,
 		"host", opts.Host,
 		"port", opts.Port,
 		"ingest_path", opts.IngestPath)
 	return s, nil
 }
 // Capabilities returns supported capabilities
 func (s *HTTPChainSource) Capabilities() []core.Capability {
 	// CapTLS/CapAuth added when transport security lands
 	return []core.Capability{
 		core.CapSessionAware,
 		core.CapMultiSession,
 	}
 }
 // Subscribe returns a channel for receiving log entries
 func (s *HTTPChainSource) Subscribe() <-chan core.LogEntry {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 	ch := make(chan core.LogEntry, s.config.BufferSize)
 	s.subscribers = append(s.subscribers, ch)
 	return ch
 }
 // Start binds the listener and serves the ingest endpoint
 func (s *HTTPChainSource) Start() error {
 	addr := net.JoinHostPort(s.config.Host, strconv.FormatInt(s.config.Port, 10))
 	// IPv4-only, aligns with tcp/http sinks
 	ln, err := net.Listen("tcp4", addr)
 	if err != nil {
 		return fmt.Errorf("listen %s: %w", addr, err)
 	}
 	mux := http.NewServeMux()
 	// Method-scoped pattern: mux answers 405 with Allow header on non-POST
 	mux.HandleFunc(http.MethodPost+" "+s.config.IngestPath, s.handleIngest)
 	s.server = &http.Server{
 		Handler:           mux,
 		ReadTimeout:       time.Duration(s.config.ReadTimeoutMS) * time.Millisecond,
 		ReadHeaderTimeout: HTTPChainReadHeaderTimeout,
 		// Future: TLSConfig for transport security
 	}
 	s.startTime = time.Now()
 	go func() {
 		if err := s.server.Serve(ln); err != nil && !errors.Is(err, http.ErrServerClosed) {
 			s.logger.Error("msg", "HTTP chain server terminated",
 				"component", "http_chain_source",
 				"instance_id", s.id,
 				"error", err)
 		}
 	}()
 	s.logger.Info("msg", "HTTP chain source started",
 		"component", "http_chain_source",
 		"instance_id", s.id,
 		"addr", addr)
 	return nil
 }
 // Stop shuts down the server, sessions, and subscriber channels
 func (s *HTTPChainSource) Stop() {
 	if s.server != nil {
 		ctx, cancel := context.WithTimeout(context.Background(), HTTPChainServerShutdownTimeout)
 		defer cancel()
 		s.server.Shutdown(ctx)
 	}
 	s.sessionsMu.Lock()
 	for _, id := range s.sessions {
 		s.proxy.RemoveSession(id)
 	}
 	s.sessions = make(map[string]string)
 	s.sessionsMu.Unlock()
 	s.mu.Lock()
 	for _, ch := range s.subscribers {
 		close(ch)
 	}
 	s.mu.Unlock()
 	s.logger.Info("msg", "HTTP chain source stopped",
 		"component", "http_chain_source",
 		"instance_id", s.id)
 }
 // GetStats returns the source's statistics
 func (s *HTTPChainSource) GetStats() source.SourceStats {
 	lastEntry, _ := s.lastEntryTime.Load().(time.Time)
 	s.sessionsMu.Lock()
 	cachedSessions := len(s.sessions)
 	s.sessionsMu.Unlock()
 	return source.SourceStats{
 		ID:             s.id,
 		Type:           "http_chain",
 		TotalEntries:   s.totalEntries.Load(),
 		DroppedEntries: s.droppedEntries.Load(),
 		StartTime:      s.startTime,
 		LastEntryTime:  lastEntry,
 		Details: map[string]any{
 			"host":              s.config.Host,
 			"port":              s.config.Port,
 			"ingest_path":       s.config.IngestPath,
 			"total_requests":    s.totalRequests.Load(),
 			"rejected_requests": s.rejectedRequests.Load(),
 			"parse_errors":      s.parseErrors.Load(),
 			"cached_sessions":   cachedSessions,
 			"trust_node":        s.config.TrustNode,
 		},
 	}
 }
 // handleIngest validates protocol headers and ingests one NDJSON batch.
 // Batch acceptance is atomic: entries publish only after a clean full read.
 func (s *HTTPChainSource) handleIngest(w http.ResponseWriter, r *http.Request) {
 	s.totalRequests.Add(1)
 	if r.Header.Get(chain.HeaderProtocol) != strconv.Itoa(chain.ProtocolVersion) {
 		s.rejectedRequests.Add(1)
 		http.Error(w, "unsupported protocol version", http.StatusBadRequest)
 		return
 	}
 	remoteHost := r.RemoteAddr
 	if host, _, err := net.SplitHostPort(r.RemoteAddr); err == nil {
 		remoteHost = host
 	}
 	connNode := r.Header.Get(chain.HeaderNode)
 	if connNode == "" || !s.config.TrustNode {
 		connNode = remoteHost
 	}
 	body := http.MaxBytesReader(w, r.Body, s.config.MaxBodyBytes)
 	scanner := bufio.NewScanner(body)
 	scanner.Buffer(make([]byte, 0, 64*1024), core.MaxLogEntryBytes)
 	entries := make([]core.LogEntry, 0, 128)
 	for scanner.Scan() {
 		line := scanner.Bytes()
 		if len(line) == 0 {
 			continue
 		}
 		entry, err := chain.DecodeEntry(line, connNode, s.config.TrustNode)
 		if err != nil {
 			// Content error within a clean transfer: skip line, keep batch
 			s.parseErrors.Add(1)
 			continue
 		}
 		entries = append(entries, entry)
 	}
 	if err := scanner.Err(); err != nil {
 		// Transfer error: reject batch without partial ingestion, sender retries
 		s.rejectedRequests.Add(1)
 		var maxErr *http.MaxBytesError
 		if errors.As(err, &maxErr) {
 			http.Error(w, "body too large", http.StatusRequestEntityTooLarge)
 			return
 		}
 		s.logger.Debug("msg", "Chain batch read failed",
 			"component", "http_chain_source",
 			"remote_addr", r.RemoteAddr,
 			"error", err)
 		http.Error(w, "malformed body", http.StatusBadRequest)
 		return
 	}
 	for _, entry := range entries {
 		s.publish(entry)
 	}
 	s.proxy.UpdateActivity(s.sessionFor(remoteHost, connNode))
 	w.Header().Set(chain.HeaderAccepted, strconv.Itoa(len(entries)))
 	w.WriteHeader(http.StatusNoContent)
 }
 // sessionFor returns the cached session for a remote+node, recreating after idle expiry
 func (s *HTTPChainSource) sessionFor(remoteHost, node string) string {
 	key := remoteHost + "|" + node
 	s.sessionsMu.Lock()
 	defer s.sessionsMu.Unlock()
 	if id, ok := s.sessions[key]; ok {
 		if _, exists := s.proxy.GetSession(id); exists {
 			return id
 		}
 	}
 	sess := s.proxy.CreateSession(remoteHost, map[string]any{
 		"type": "http_chain",
 		"node": node,
 	})
 	s.sessions[key] = sess.ID
 	return sess.ID
 }
 // publish sends a log entry to all subscribers
 func (s *HTTPChainSource) publish(entry core.LogEntry) {
 	s.mu.RLock()
 	defer s.mu.RUnlock()
 	s.totalEntries.Add(1)
 	s.lastEntryTime.Store(entry.Time)
 	for _, ch := range s.subscribers {
 		select {
 		case ch <- entry:
 		default:
 			s.droppedEntries.Add(1)
 		}
 	}
 }
@@ -0,0 +1,353 @@
 package tcpchain
 import (
 	"bufio"
 	"context"
 	"encoding/json"
 	"errors"
 	"fmt"
 	"net"
 	"strconv"
 	"sync"
 	"sync/atomic"
 	"time"
 	"logwisp/src/internal/chain"
 	"logwisp/src/internal/config"
 	"logwisp/src/internal/core"
 	"logwisp/src/internal/plugin"
 	"logwisp/src/internal/session"
 	"logwisp/src/internal/source"
 	lconfig "github.com/lixenwraith/config"
 	"github.com/lixenwraith/log"
 )
 func init() {
 	if err := plugin.RegisterSource("tcp_chain", NewTCPChainSourcePlugin); err != nil {
 		panic(fmt.Sprintf("failed to register tcp_chain source: %v", err))
 	}
 }
 const (
 	DefaultChainSourceBufferSize     = 1000
 	DefaultChainSourceHelloTimeoutMS = 10000
 )
 // TCPChainSource accepts connections from upstream tcp_chain sinks and ingests NDJSON entries
 type TCPChainSource struct {
 	id     string
 	proxy  *session.Proxy
 	config *config.TCPChainSourceOptions
 	subscribers []chan core.LogEntry
 	listener    net.Listener
 	conns       map[net.Conn]struct{}
 	logger      *log.Logger
 	mu     sync.RWMutex
 	ctx    context.Context
 	cancel context.CancelFunc
 	wg     sync.WaitGroup
 	startTime      time.Time
 	totalEntries   atomic.Uint64
 	droppedEntries atomic.Uint64
 	parseErrors    atomic.Uint64
 	rejectedConns  atomic.Uint64
 	activeConns    atomic.Int64
 	lastEntryTime  atomic.Value // time.Time
 }
 // NewTCPChainSourcePlugin creates a tcp_chain source through plugin factory
 func NewTCPChainSourcePlugin(
 	id string,
 	configMap map[string]any,
 	logger *log.Logger,
 	proxy *session.Proxy,
 ) (source.Source, error) {
 	opts := &config.TCPChainSourceOptions{
 		Host:      "0.0.0.0",
 		TrustNode: true,
 	}
 	if err := lconfig.ScanMap(configMap, opts); err != nil {
 		return nil, fmt.Errorf("failed to parse config: %w", err)
 	}
 	if err := lconfig.Port(opts.Port); err != nil {
 		return nil, fmt.Errorf("port: %w", err)
 	}
 	if opts.BufferSize <= 0 {
 		opts.BufferSize = DefaultChainSourceBufferSize
 	}
 	if opts.HelloTimeoutMS <= 0 {
 		opts.HelloTimeoutMS = DefaultChainSourceHelloTimeoutMS
 	}
 	s := &TCPChainSource{
 		id:          id,
 		proxy:       proxy,
 		config:      opts,
 		subscribers: make([]chan core.LogEntry, 0),
 		conns:       make(map[net.Conn]struct{}),
 		logger:      logger,
 	}
 	s.lastEntryTime.Store(time.Time{})
 	logger.Info("msg", "TCP chain source initialized",
 		"component", "tcp_chain_source",
 		"instance_id", id,
 		"host", opts.Host,
 		"port", opts.Port)
 	return s, nil
 }
 // Capabilities returns supported capabilities
 func (s *TCPChainSource) Capabilities() []core.Capability {
 	// CapTLS/CapAuth added when transport security lands
 	return []core.Capability{
 		core.CapSessionAware,
 		core.CapMultiSession,
 	}
 }
 // Subscribe returns a channel for receiving log entries
 func (s *TCPChainSource) Subscribe() <-chan core.LogEntry {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 	ch := make(chan core.LogEntry, s.config.BufferSize)
 	s.subscribers = append(s.subscribers, ch)
 	return ch
 }
 // Start binds the listener and begins accepting connections
 func (s *TCPChainSource) Start() error {
 	addr := net.JoinHostPort(s.config.Host, strconv.FormatInt(s.config.Port, 10))
 	// IPv4-only
 	ln, err := net.Listen("tcp4", addr)
 	if err != nil {
 		return fmt.Errorf("listen %s: %w", addr, err)
 	}
 	s.listener = ln
 	s.ctx, s.cancel = context.WithCancel(context.Background())
 	s.startTime = time.Now()
 	s.wg.Add(1)
 	go s.acceptLoop()
 	s.logger.Info("msg", "TCP chain source started",
 		"component", "tcp_chain_source",
 		"instance_id", s.id,
 		"addr", addr)
 	return nil
 }
 // Stop closes the listener, all connections, and subscriber channels
 func (s *TCPChainSource) Stop() {
 	if s.cancel != nil {
 		s.cancel()
 	}
 	if s.listener != nil {
 		s.listener.Close()
 	}
 	s.mu.Lock()
 	for conn := range s.conns {
 		conn.Close() // unblocks per-connection reads
 	}
 	s.mu.Unlock()
 	s.wg.Wait()
 	s.mu.Lock()
 	for _, ch := range s.subscribers {
 		close(ch)
 	}
 	s.mu.Unlock()
 	s.logger.Info("msg", "TCP chain source stopped",
 		"component", "tcp_chain_source",
 		"instance_id", s.id)
 }
 // GetStats returns the source's statistics
 func (s *TCPChainSource) GetStats() source.SourceStats {
 	lastEntry, _ := s.lastEntryTime.Load().(time.Time)
 	return source.SourceStats{
 		ID:             s.id,
 		Type:           "tcp_chain",
 		TotalEntries:   s.totalEntries.Load(),
 		DroppedEntries: s.droppedEntries.Load(),
 		StartTime:      s.startTime,
 		LastEntryTime:  lastEntry,
 		Details: map[string]any{
 			"host":               s.config.Host,
 			"port":               s.config.Port,
 			"active_connections": s.activeConns.Load(),
 			"rejected_conns":     s.rejectedConns.Load(),
 			"parse_errors":       s.parseErrors.Load(),
 			"trust_node":         s.config.TrustNode,
 		},
 	}
 }
 // acceptLoop accepts upstream connections until listener close
 func (s *TCPChainSource) acceptLoop() {
 	defer s.wg.Done()
 	for {
 		conn, err := s.listener.Accept()
 		if err != nil {
 			if errors.Is(err, net.ErrClosed) || s.ctx.Err() != nil {
 				return
 			}
 			s.logger.Warn("msg", "Accept error",
 				"component", "tcp_chain_source",
 				"error", err)
 			continue
 		}
 		if s.config.MaxConnections > 0 && s.activeConns.Load() >= s.config.MaxConnections {
 			s.rejectedConns.Add(1)
 			conn.Close()
 			continue
 		}
 		s.mu.Lock()
 		s.conns[conn] = struct{}{}
 		s.mu.Unlock()
 		s.wg.Add(1)
 		go s.handleConn(conn)
 	}
 }
 // handleConn validates the hello preamble, then streams entries until EOF/error
 func (s *TCPChainSource) handleConn(conn net.Conn) {
 	defer s.wg.Done()
 	remote := conn.RemoteAddr().String()
 	s.activeConns.Add(1)
 	var sessID string
 	defer func() {
 		conn.Close()
 		s.mu.Lock()
 		delete(s.conns, conn)
 		s.mu.Unlock()
 		if sessID != "" {
 			s.proxy.RemoveSession(sessID)
 		}
 		s.activeConns.Add(-1)
 	}()
 	scanner := bufio.NewScanner(conn)
 	// Oversized line (> MaxLogEntryBytes) is a protocol violation; scanner is
 	// unrecoverable after ErrTooLong, connection terminates
 	scanner.Buffer(make([]byte, 0, 64*1024), core.MaxLogEntryBytes)
 	// Hello preamble
 	conn.SetReadDeadline(time.Now().Add(time.Duration(s.config.HelloTimeoutMS) * time.Millisecond))
 	if !scanner.Scan() {
 		s.logger.Warn("msg", "Connection closed before hello",
 			"component", "tcp_chain_source",
 			"remote_addr", remote,
 			"error", scanner.Err())
 		return
 	}
 	hello, err := chain.DecodeHello(scanner.Bytes())
 	if err != nil {
 		s.logger.Warn("msg", "Rejected chain connection",
 			"component", "tcp_chain_source",
 			"remote_addr", remote,
 			"error", err)
 		return
 	}
 	connNode := hello.Node
 	if connNode == "" || !s.config.TrustNode {
 		if host, _, splitErr := net.SplitHostPort(remote); splitErr == nil {
 			connNode = host
 		} else {
 			connNode = remote
 		}
 	}
 	sess := s.proxy.CreateSession(remote, map[string]any{
 		"type": "tcp_chain",
 		"node": connNode,
 	})
 	sessID = sess.ID
 	s.logger.Info("msg", "Chain connection established",
 		"component", "tcp_chain_source",
 		"remote_addr", remote,
 		"node", connNode)
 	idle := time.Duration(s.config.ReadTimeoutMS) * time.Millisecond
 	for {
 		if idle > 0 {
 			conn.SetReadDeadline(time.Now().Add(idle))
 		} else {
 			conn.SetReadDeadline(time.Time{})
 		}
 		if !scanner.Scan() {
 			if err := scanner.Err(); err != nil && !errors.Is(err, net.ErrClosed) {
 				s.logger.Debug("msg", "Chain read terminated",
 					"component", "tcp_chain_source",
 					"remote_addr", remote,
 					"error", err)
 			}
 			return
 		}
 		line := scanner.Bytes()
 		if len(line) == 0 {
 			continue
 		}
 		s.proxy.UpdateActivity(sessID)
 		entry, err := chain.DecodeEntry(line, connNode, s.config.TrustNode)
 		if err != nil {
 			s.parseErrors.Add(1)
 			s.logger.Debug("msg", "Dropped malformed chain entry",
 				"component", "tcp_chain_source",
 				"error", err)
 			continue
 		}
 		s.publish(entry)
 	}
 }
 // parseEntry decodes a canonical LogEntry line and applies the node policy
 func (s *TCPChainSource) parseEntry(line []byte, connNode string) (core.LogEntry, bool) {
 	var entry core.LogEntry
 	if err := json.Unmarshal(line, &entry); err != nil {
 		s.parseErrors.Add(1)
 		s.logger.Debug("msg", "Dropped malformed chain entry",
 			"component", "tcp_chain_source",
 			"error", err)
 		return core.LogEntry{}, false
 	}
 	if entry.Time.IsZero() {
 		entry.Time = time.Now()
 	}
 	if entry.Node == "" || !s.config.TrustNode {
 		entry.Node = connNode
 	}
 	entry.RawSize = int64(len(line))
 	return entry, true
 }
 // publish sends a log entry to all subscribers
 func (s *TCPChainSource) publish(entry core.LogEntry) {
 	s.mu.RLock()
 	defer s.mu.RUnlock()
 	s.totalEntries.Add(1)
 	s.lastEntryTime.Store(entry.Time)
 	for _, ch := range s.subscribers {
 		select {
 		case ch <- entry:
 		default:
 			s.droppedEntries.Add(1)
 		}
 	}
 }