v0.18.2 resume rotated files instead of replay, keep quiet SSE alive, refuse HEAD on strem path

Dockerfile

@@ -0,0 +1,40 @@
+# Builder pin and go.mod directive are the same patch release deliberately;
+# an older builder reports the mismatch only after downloading the module graph.
+ARG GO_VERSION=1.27.1
+
+FROM docker.io/library/golang:${GO_VERSION}-alpine AS build
+
+# Git supplies Go's VCS build information; only /out/logwisp crosses stages.
+RUN apk add --no-cache git
+WORKDIR /src
+COPY go.mod go.sum ./
+RUN go mod download
+COPY . .
+
+ARG TARGETOS=linux
+ARG TARGETARCH=amd64
+ARG VERSION=dev
+ARG REVISION=unknown
+RUN CGO_ENABLED=0 GOOS=${TARGETOS} GOARCH=${TARGETARCH} \
+    go build -trimpath \
+      -ldflags="-s -w -X logwisp/internal/version.Version=${VERSION} -X logwisp/internal/version.GitCommit=${REVISION}" \
+      -o /out/logwisp ./cmd/logwisp
+
+FROM scratch
+
+ARG VERSION=dev
+ARG REVISION=unknown
+
+LABEL org.opencontainers.image.title="logwisp" \
+      org.opencontainers.image.description="Log transport: sources, flow, sinks" \
+      org.opencontainers.image.source="https://github.com/lixenwraith/logwisp" \
+      org.opencontainers.image.revision="${REVISION}" \
+      org.opencontainers.image.version="${VERSION}" \
+      org.opencontainers.image.licenses="BSD-3-Clause"
+
+COPY --from=build /out/logwisp /logwisp
+
+# Numeric identity is required in scratch and satisfies a restricted pod spec.
+USER 65532:65532
+
+ENTRYPOINT ["/logwisp"]

Makefile

@@ -33,7 +33,7 @@ install: build
 
 # Uninstall the binary
 uninstall:
-	rm -f $(BINDIR)/$(BINARY_PATH)
+	rm -f $(BINDIR)/$(BINARY_NAME)
 
 # Clean build artifacts
 clean:

README.md

@@ -6,7 +6,7 @@
     <td>
       <h1>LogWisp</h1>
       <p>
-        <a href="https://golang.org"><img src="https://img.shields.io/badge/Go-1.26-00ADD8?style=flat&logo=go" alt="Go"></a>
+        <a href="https://golang.org"><img src="https://img.shields.io/badge/Go-1.27.1-00ADD8?style=flat&logo=go" alt="Go"></a>
         <a href="https://opensource.org/licenses/BSD-3-Clause"><img src="https://img.shields.io/badge/License-BSD_3--Clause-blue.svg" alt="License"></a>
         <a href="doc/"><img src="https://img.shields.io/badge/Docs-Available-green.svg" alt="Documentation"></a>
       </p>
@@ -137,7 +137,7 @@ synthetic generator writing JSON to stdout.
 
 - **Operating systems**: Linux (kernel 6.10+), FreeBSD (14.0+)
 - **Architecture**: amd64
-- **Go**: 1.26+ to build from source
+- **Go**: 1.27.1+ to build from source
 
 Network sources and sinks bind and dial over IPv4 only.
 

config/logwisp.toml

@@ -152,6 +152,9 @@ directory = "./"                        # Directory to monitor (required, not re
 pattern = "*.log"                       # Glob pattern (* and ? only)
 ## Tailing an already-open file polls at a fixed 100ms, regardless of this value.
 check_interval_ms = 100                 # Directory rescan interval (min 10)
+## raw = true never parses a line; with format type "raw" the file is relayed byte for byte.
+raw = false                             # Keep the whole line as the message
+from = "end"                            # "end" or "start" of a newly discovered file
 
 ## Console source (stdin, single instance per pipeline)
 # [[pipelines.plugin_sources]]

doc/formatters.md

@@ -30,8 +30,10 @@ Omitting `[pipelines.flow.format]` entirely selects `raw`.
 
 ### raw
 
-Passthrough. `FlagRaw` bypasses both formatting and sanitization, so the
+Passthrough. `FlagRaw` bypasses formatting and sanitization: the message reaches
-message reaches the sink exactly as the source produced it.
+the sink exactly as the source produced it, with no timestamp, level or source
+prefix added. An entry that also carries `fields` gets the fields JSON appended
+verbatim after a single space — `raw` never drops data and never re-encodes it.
 
 ```toml
 [pipelines.flow.format]
@@ -42,6 +44,12 @@ Fastest option, and the right one when you are relaying text that is already in
 its final form. Note that it also bypasses sanitization, so control characters
 in the source data reach your sinks intact.
 
+Byte-exact transport needs a source that does not split the line: the `console`
+source, or the `file` source with `raw = true`. Both put the whole line —
+newline included — in the message and leave `fields` empty. The `file` source's
+JSON branch splits a line into message and fields, so `raw` reassembles it as
+`<msg> <fields>` rather than reproducing the original object.
+
 ### txt
 
 Human-readable line output with a timestamp and level.
@@ -89,7 +97,7 @@ you need to override the defaults.
 
 With `flags = 0` the formatter selects `1` for `type = "raw"` and `6`
 (timestamp + level) for every other type. `8` is added automatically whenever an
-entry carries parseable `fields`.
+entry carries parseable `fields` and `1` is not set; `1` always wins.
 
 Examples: `flags = 4` for level only, no timestamp; `flags = 2` for timestamp
 only, no level.
@@ -133,11 +141,11 @@ when writing downstream parsers or grep patterns.
 
 ## Structured Fields
 
-When an entry carries `Fields` (raw JSON), the formatter parses it and switches
+When an entry carries `Fields` (raw JSON) and `FlagRaw` is not set, the
-to structured rendering by adding the `StructuredJSON` flag automatically.
+formatter parses it and switches to structured rendering by adding the
-Fields reach a pipeline in two ways: from the `file` source when a tailed line
+`StructuredJSON` flag automatically. Fields reach a pipeline in two ways: from
-parses as JSON with a `fields` key, and from the heartbeat generator when
+the `file` source when a tailed line parses as JSON with a `fields` key, and
-`include_stats = true`.
+from the heartbeat generator when `include_stats = true`.
 
 ## Choosing a Configuration
 

doc/installation.md

@@ -4,7 +4,7 @@
 
 - **Operating systems**: Linux (kernel 6.10+), FreeBSD (14.0+)
 - **Architecture**: amd64
-- **Go**: 1.26 or newer, to build from source
+- **Go**: 1.27.1 or newer, to build from source
 
 ## Building from Source
 
@@ -22,7 +22,7 @@ The Makefile works with both GNU make and BSD make. Targets:
 | `make` / `make build` | Build `bin/logwisp` with version metadata |
 | `make dev` | Build with the race detector enabled |
 | `make install` | Install the binary to `$(PREFIX)/bin` (default `/usr/local`) |
-| `make uninstall` | Intended to remove the installed binary — currently broken: it expands to `$(BINDIR)/bin/logwisp` instead of `$(BINDIR)/logwisp`, so it removes nothing. Delete the binary by hand |
+| `make uninstall` | Remove `$(BINDIR)/logwisp` |
 | `make clean` | Remove the built binary |
 | `make version` | Print the version, commit, and build time that would be embedded |
 
@@ -37,6 +37,24 @@ go build -o bin/logwisp ./cmd/logwisp
 `go install github.com/lixenwraith/logwisp/cmd/logwisp@latest` also works, with
 the same loss of version metadata.
 
+## Container Image
+
+The root `Dockerfile` builds the same package into `scratch` under UID 65532,
+static and stripped. There is no shell and no config in the image: mount one and
+name it, as the binary has no daemon mode and no built-in defaults worth running.
+
+```bash
+REV=$(git rev-parse HEAD)
+docker build -t "logwisp:$(git rev-parse --short HEAD)" \
+  --build-arg VERSION="$(git describe --tags --always)" \
+  --build-arg REVISION="$REV" .
+docker run --rm -v /etc/logwisp:/etc/logwisp:ro logwisp:... -c /etc/logwisp/logwisp.toml
+```
+
+Sinks that listen (`http`, `tcp`) need their ports published; the read-only
+root filesystem and dropped capabilities a restricted runtime imposes are all
+compatible with it, provided a `file` sink's directory is writable by 65532.
+
 ## Configuration
 
 Copy the annotated reference configuration and edit it:
@@ -173,27 +191,20 @@ mode; see [Operations](operations.md#checking-a-configuration).
 
 ## Test Scripts
 
-Two end-to-end scripts under `test/` build multi-node chain topologies against a
+End-to-end scripts under `test/` run against a local build:
-local build:
 
 ```bash
 make
 ./test/chain-test.sh --auto             # two independent relay pipelines
 ./test/chain-aggregate-test.sh --auto   # fan-in: both edges into one pipeline
+./test/mtls-chain-test.sh --auto        # the same fan-in under mTLS
+./test/passthrough-test.sh              # file source relays a wide envelope intact
 ```
 
-Without `--auto` they run the relay in the foreground for interactive
+Without `--auto` the chain scripts run the relay in the foreground for
-inspection. They need bash 5+, coreutils, and curl, and they bind ports
+interactive inspection. They need bash 5+, coreutils, and curl, and they bind
-15801–15804. Generated configuration and logs land in `test/run/`.
+ports 15801–15804. The pass-through test binds nothing. Generated configuration
-
+and logs land in `test/run/`.
-> Two of the three `--auto` assertions currently report `FAIL` against a
-> working build. They grep the sink output for `"source":"edge-tcp/` and
-> `"node":"edge-http"`, but the JSON formatter emits the `node/source` label
-> under the key `trace`. The transport itself is healthy — the
-> `total_processed` assertion passes and the streamed entries carry
-> `"trace":"edge-tcp/random_rand"` as expected. Until the assertions are
-> updated, verify the streams by eye with `nc 127.0.0.1 15803` and
-> `curl -sN http://127.0.0.1:15804/stream`.
 
 ## Uninstall
 

doc/operations.md

@@ -124,6 +124,9 @@ curl -s http://127.0.0.1:8080/status | jq .
     "tls": false,
     "active_clients": 3,
     "buffer_size": 1000,
+    "client_buffer_size": 256,
+    "max_connections": 32,
+    "write_timeout_ms": 5000,
     "uptime_seconds": 8130
   },
   "endpoints": { "stream": "/stream", "status": "/status" },
@@ -145,7 +148,7 @@ This endpoint is scoped to one sink, not to the whole process, and it is
 | `dropped_entries` | source | Downstream cannot keep up with the source |
 | `total_dropped` | flow | Rate limit or filters are discarding entries (often intended) |
 | `total_dropped_by_sink` | pipeline | A sink's input queue is full |
-| `dropped_writes` | tcp/http sink | A specific client is too slow |
+| `dropped_writes` | tcp/http sink | A client's queue overflowed: either it is too slow, or one burst exceeded `client_buffer_size` |
 | `rejected_conns` / `rejected_clients` | tcp/http sink, tcp_chain source | `max_connections` is being hit |
 | `tls_handshake_errors` | tcp sink, tcp_chain source | Certificate or version mismatch, or scanning |
 | `parse_errors` | chain source | Protocol or version skew upstream |
@@ -182,9 +185,18 @@ the filter stage logs several lines per entry evaluated.
 ### Buffers
 
 Raise `buffer_size` when `total_dropped_by_sink` is climbing but the sink itself
-is healthy — that is a burst-absorption problem. Raise `client_buffer_size` when
+is healthy — that is a burst-absorption problem.
-`dropped_writes` is climbing for network sinks; that is a slow-consumer problem,
+
-and a bigger buffer only buys time.
+`dropped_writes` on a network sink has two causes that a counter alone does not
+separate. A consumer slower than the sustained rate cannot be bought off with
+buffer, and drops are the intended outcome. A burst the consumer would have
+drained, arriving faster than it reads, is configuration: the sink queues a
+whole burst while the client writes one frame at a time, so the part of a burst
+above `client_buffer_size` is lost even to a loopback reader.
+Where a `rate_limit` bounds the pipeline, its `burst` is that number — keep
+`client_buffer_size` at or above it and the second cause disappears. The HTTP
+status endpoint reports both queue bounds alongside the counters so an operator
+can tell which one is in play.
 
 ```toml
 [pipelines.plugin_sinks.config]
@@ -231,7 +243,8 @@ pipeline `total_dropped_by_sink` (sink backed up?), sink `total_processed`.
 
 - The watcher seeks to end-of-file on start; only content appended afterwards is
   read. Positions are in memory, so a restart re-seeks to end and anything
-  written during the downtime is lost.
+  written during the downtime is lost. `from = "start"` reads each file whole
+  instead, and replays it on every restart.
 - `pattern` is a filename glob with `*` and `?` only, and matching is not
   recursive.
 - `check_interval_ms` governs how quickly a *new file* is noticed; tailing an

doc/sinks.md

@@ -140,7 +140,9 @@ allow = ["viewer-01"]
 
 **Behaviour**
 
-- Only `GET` is routed to either path; anything else gets `405`.
+- Only `GET` is routed to either path; anything else gets `405`, `HEAD` on
+  `stream_path` included — a stream is a body, and a client registered to have
+  its body discarded never reads and never leaves.
 - With an `auth` block, one middleware gates **both** endpoints: an
   unauthorized client gets `403` with no body detail, and the rejection is
   logged at WARN and counted in `auth_rejected`. The authorized identity is
@@ -150,19 +152,29 @@ allow = ["viewer-01"]
 - Payloads are framed per the SSE spec, one `data:` line per newline in the
   payload, so multi-line entries stream correctly.
 - The server sets no `WriteTimeout` (that would kill long-lived streams);
-  per-write deadlines come from `write_timeout_ms` via `http.ResponseController`.
+  per-write deadlines come from `write_timeout_ms` via `http.ResponseController`
+  and cover the connected frame, every payload, and the idle comment.
+- A quiet stream emits an SSE comment every 15 s. It refreshes the client's
+  session and is how a peer that stopped reading is noticed.
 - A client whose send queue is full has that event dropped
-  (`dropped_writes`); it is not disconnected.
+  (`dropped_writes`); it is not disconnected. A `dropped_writes` that rises while
+  no client is behind is a burst larger than `client_buffer_size`, not
+  backpressure: size the queue at or above whatever burst the pipeline's
+  `rate_limit` releases at once.
+- A client is registered only once its connected frame has flushed, so the
+  broker never queues into a buffer whose reader has not started.
 - Clients whose session has been idle-expired by the session manager are
-  evicted by the broker.
+  evicted by the broker. With the idle comment above, that reaches only a peer
+  that has stopped accepting bytes on a sink configured `write_timeout_ms = 0`.
 - On shutdown, connected clients receive
   `event: disconnect / data: {"reason":"server_shutdown"}`.
 - HTTP/2 is negotiated via ALPN when TLS is enabled; plaintext is HTTP/1.1.
 
 **Status endpoint** returns service and version identity, host, port, TLS flag,
-the compiled auth policy, active client count, buffer size, uptime, endpoint
+the compiled auth policy, active client count, sink and per-client buffer sizes,
-paths, and the `total_processed` / `dropped_writes` / `rejected_clients` /
+connection limit, write timeout, uptime, endpoint paths, and the
-`auth_rejected` counters.
+`total_processed` / `dropped_writes` / `rejected_clients` / `auth_rejected`
+counters.
 
 > Without an `auth` block both endpoints are unauthenticated, and the stream
 > response carries `Access-Control-Allow-Origin: *`, so any web origin can read

doc/sources.md

@@ -32,6 +32,8 @@ type = "file"
 directory         = "/var/log/myapp"
 pattern           = "*.log"
 check_interval_ms = 100
+raw               = false
+from              = "end"
 ```
 
 | Option | Type | Default | Description |
@@ -39,6 +41,8 @@ check_interval_ms = 100
 | `directory` | string | **required** | Directory to scan; not recursive |
 | `pattern` | string | `*` | Glob over filenames; `*` and `?` only |
 | `check_interval_ms` | int | `100` | Directory rescan interval; minimum `10` |
+| `raw` | bool | `false` | Never parse a line: the whole line is the message |
+| `from` | string | `end` | Where a new watcher starts: `end` or `start` of the file |
 
 **Behaviour**
 
@@ -49,15 +53,27 @@ check_interval_ms = 100
   stopped and removed on the next scan.
 - A new watcher seeks to end-of-file. Positions live in memory only, so a
   restart resumes from the current end of each file and content written while
-  LogWisp was down is not read.
+  LogWisp was down is not read. `from = "start"` reads each file whole when its
+  watcher is created instead — what a process writing beside LogWisp needs, at
+  the cost of replaying a file already on disk at every restart.
 - Rotation is detected from size decrease, modification-time reset, a position
   beyond end-of-file, or an inode change. An inode change where the new file is
   already larger than the recorded position is treated as an atomic save, not a
   rotation, and the position is preserved.
-- Lines are parsed as JSON when they contain `time`, `level`, `msg`, and
+- A rotation that renames in place — what a size-capped writer does — puts the
-  `fields` keys; `time` is read as RFC3339Nano. Anything else is kept as plain
+  same inode back under a name `pattern` also matches. Its watcher resumes at
-  text with the level inferred from common markers (`[ERROR]`, `WARN:`, and so
+  the position the original reached, so `from = "start"` reads the tail an
-  on).
+  unfinished read left behind rather than the whole archive a second time.
+- A line is parsed as JSON only when it is an object whose top-level keys are
+  all drawn from `time`, `level`, `msg` and `fields` — the four an entry can
+  carry. `time` is read as RFC3339Nano. Any other key, and any non-object line,
+  is kept whole as text with the level inferred from common markers
+  (`[ERROR]`, `WARN:`, and so on), because parsing it would drop the rest.
+- `raw = true` skips the JSON branch entirely. The line, plus its newline,
+  becomes the message; `fields` stays empty, the time is the read time, and the
+  level is inferred from the text as for any unparsed line. Paired with
+  `format.type = "raw"` this is byte-exact transport for records LogWisp's
+  envelope cannot hold — see [Formatters](formatters.md#raw).
 - `Source` is set to the file's base name.
 
 **Statistics**: per-watcher size, position, entries read, rotation count, and

go.mod

@@ -1,6 +1,6 @@
 module logwisp
 
-go 1.26.5
+go 1.27.1
 
 require (
 	github.com/lixenwraith/config v0.1.1-0.20260712172228-ccd280ba6a98

internal/config/config.go

@@ -198,6 +198,8 @@ type FileSourceOptions struct {
 	Directory       string `toml:"directory"`
 	Pattern         string `toml:"pattern"` // glob pattern
 	CheckIntervalMS int64  `toml:"check_interval_ms"`
+	Raw             bool   `toml:"raw"`  // keep the whole line as the message, never parse it
+	From            string `toml:"from"` // "end" (default) or "start" of a newly discovered file
 }
 
 // ConsoleSourceOptions defines settings for a stdin-based source

internal/core/const.go

@@ -13,6 +13,10 @@ const (
 
 	SessionCleanupInterval = 5 * time.Minute
 
+	// Idle keepalive for a served stream. Well under SessionDefaultMaxIdleTime,
+	// so a quiet stream refreshes its session long before the sweep expires it.
+	StreamKeepaliveInterval = 15 * time.Second
+
 	ServiceStatsUpdateInterval = 1 * time.Second
 
 	ShutdownTimeout = 10 * time.Second

internal/format/adapter.go

@@ -90,57 +90,44 @@ func NewFormatterAdapter(cfg *config.FormatConfig) (*FormatterAdapter, error) {
 
 // Format implements Formatter interface
 func (a *FormatterAdapter) Format(entry core.LogEntry) ([]byte, error) {
-	// Map logwisp LogEntry to formatter args
+	return a.serialize(entry, a.flags), nil
-	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
-	effectiveFlags := a.flags
-
-	if len(entry.Fields) > 0 {
-		// Parse fields JSON
-		var fields map[string]any
-		if err := json.Unmarshal(entry.Fields, &fields); err == nil && len(fields) > 0 {
-			// Use structured JSON format for fields
-			args = []any{entry.Message, fields}
-			// Add structured flag to properly format fields as JSON object
-			effectiveFlags |= formatter.FlagStructuredJSON
-			return a.formatter.Format(effectiveFlags, entry.Time, level, src, args), nil
-		}
-	}
-	if args == nil {
-		args = []any{entry.Message}
-	}
-
-	a.mu.Lock()
-	out := bytes.Clone(a.formatter.Format(effectiveFlags, entry.Time, level, src, args))
-	a.mu.Unlock()
-	return out, nil
 }
 
 // FormatWithFlags allows custom flags for specific formatting needs
 func (a *FormatterAdapter) FormatWithFlags(entry core.LogEntry, customFlags int64) ([]byte, error) {
-	level := mapLevel(entry.Level)
+	return a.serialize(entry, customFlags), nil
-	src := sourceLabel(entry)
+}
 
-	var args []any
+// serialize renders an entry under the given flags. The returned slice is a
-	if len(entry.Fields) > 0 {
+// copy: the underlying formatter reuses one buffer and sinks retain payloads.
-		var fields map[string]any
+func (a *FormatterAdapter) serialize(entry core.LogEntry, flags int64) []byte {
-		if err := json.Unmarshal(entry.Fields, &fields); err == nil && len(fields) > 0 {
+	args, flags := formatArgs(entry, flags)
-			args = []any{entry.Message, fields}
-			customFlags |= formatter.FlagStructuredJSON
-		}
-	}
-	if args == nil {
-		args = []any{entry.Message}
-	}
 
 	a.mu.Lock()
-	out := bytes.Clone(a.formatter.Format(customFlags, entry.Time, level, src, args))
+	out := bytes.Clone(a.formatter.Format(flags, entry.Time, mapLevel(entry.Level), sourceLabel(entry), args))
 	a.mu.Unlock()
-	return out, nil
+	return out
+}
+
+// formatArgs pairs the entry with its flags. FlagRaw keeps the fields JSON
+// verbatim beside the message rather than silently overriding the caller's
+// choice of passthrough; every other mode renders it as a JSON object.
+func formatArgs(entry core.LogEntry, flags int64) ([]any, int64) {
+	if len(entry.Fields) == 0 {
+		return []any{entry.Message}, flags
+	}
+	if flags&formatter.FlagRaw != 0 {
+		if entry.Message == "" {
+			return []any{[]byte(entry.Fields)}, flags
+		}
+		return []any{entry.Message, []byte(entry.Fields)}, flags
+	}
+
+	var fields map[string]any
+	if err := json.Unmarshal(entry.Fields, &fields); err != nil || len(fields) == 0 {
+		return []any{entry.Message}, flags
+	}
+	return []any{entry.Message, fields}, flags | formatter.FlagStructuredJSON
 }
 
 // Name returns formatter type

internal/sink/http/http.go

@@ -68,6 +68,7 @@ type HTTPSink struct {
 	clientsMu    sync.Mutex
 	nextClientID atomic.Uint64
 	writeTimeout time.Duration
+	keepalive    time.Duration
 
 	// TLS
 	tlsConfig *tls.Config
@@ -150,6 +151,7 @@ func NewHTTPSinkPlugin(
 		logger:       logger,
 		clients:      make(map[uint64]*sseClient),
 		writeTimeout: time.Duration(opts.WriteTimeoutMS) * time.Millisecond,
+		keepalive:    core.StreamKeepaliveInterval,
 		tlsConfig:    tlsCfg,
 		auth:         authPolicy,
 	}
@@ -205,6 +207,10 @@ func (h *HTTPSink) Start(ctx context.Context) error {
 	// Method-scoped patterns: mux answers 405 with Allow header on non-GET
 	mux.HandleFunc(http.MethodGet+" "+h.config.StreamPath, h.handleStream)
 	mux.HandleFunc(http.MethodGet+" "+h.config.StatusPath, h.handleStatus)
+	// A GET pattern also serves HEAD, and a HEAD stream is a registered client
+	// whose body writes are discarded: it never reads, so nothing but the peer
+	// closing the connection ends it. The status path answers one either way.
+	mux.HandleFunc(http.MethodHead+" "+h.config.StreamPath, streamHeadNotAllowed)
 
 	// One wrapper covers stream and status, and keeps the handlers themselves
 	// unaware of authorization
@@ -287,9 +293,9 @@ func (h *HTTPSink) shutdown() {
 	})
 }
 
-// removeClient unregisters a client; the first caller closes the send
+// removeClient unregisters a client; the first caller closes the send channel.
-// channel and removes the session. Broker (stale-session eviction) and
+// Broker (stale-session eviction) and stream handler (disconnect) may race here
-// stream handler (disconnect) may race here safely.
+// safely. The session is the handler's, released when it returns.
 func (h *HTTPSink) removeClient(id uint64) {
 	h.clientsMu.Lock()
 	c, ok := h.clients[id]
@@ -299,7 +305,6 @@ func (h *HTTPSink) removeClient(id uint64) {
 	h.clientsMu.Unlock()
 	if ok {
 		close(c.send)
-		h.proxy.RemoveSession(c.sessionID)
 	}
 }
 
@@ -374,10 +379,6 @@ func (h *HTTPSink) handleStream(w http.ResponseWriter, r *http.Request) {
 	}
 	id := h.nextClientID.Add(1)
 
-	h.clientsMu.Lock()
-	h.clients[id] = c
-	h.clientsMu.Unlock()
-
 	count := h.activeClients.Add(1)
 	h.logger.Debug("msg", "HTTP client connected",
 		"component", "http_sink",
@@ -389,6 +390,7 @@ func (h *HTTPSink) handleStream(w http.ResponseWriter, r *http.Request) {
 
 	defer func() {
 		h.removeClient(id)
+		h.proxy.RemoveSession(sess.ID)
 		newCount := h.activeClients.Add(-1)
 		h.logger.Debug("msg", "HTTP client disconnected",
 			"component", "http_sink",
@@ -413,11 +415,24 @@ func (h *HTTPSink) handleStream(w http.ResponseWriter, r *http.Request) {
 		"status_path": h.config.StatusPath,
 		"buffer_size": h.config.ClientBufferSize,
 	})
+	h.armWrite(rc)
 	fmt.Fprintf(w, "event: connected\ndata: %s\n\n", info)
 	if err := rc.Flush(); err != nil {
 		return
 	}
 
+	// Registered only now: a client the broker can queue into before its reader
+	// reaches the loop below loses a burst to a buffer nobody is draining.
+	h.clientsMu.Lock()
+	h.clients[id] = c
+	h.clientsMu.Unlock()
+
+	// A stream with nothing to carry still has to prove the peer is there. The
+	// comment refreshes the session the broker evicts on, and fails on a peer
+	// that stopped reading.
+	idle := time.NewTicker(h.keepalive)
+	defer idle.Stop()
+
 	clientGone := r.Context().Done()
 	for {
 		select {
@@ -425,9 +440,7 @@ func (h *HTTPSink) handleStream(w http.ResponseWriter, r *http.Request) {
 			if !ok {
 				return // broker evicted (stale session)
 			}
-			if h.writeTimeout > 0 {
+			h.armWrite(rc)
-				_ = rc.SetWriteDeadline(time.Now().Add(h.writeTimeout))
-			}
 			if err := writeSSE(w, payload); err != nil {
 				return
 			}
@@ -435,6 +448,15 @@ func (h *HTTPSink) handleStream(w http.ResponseWriter, r *http.Request) {
 				return
 			}
 			h.proxy.UpdateActivity(sess.ID)
+		case <-idle.C:
+			h.armWrite(rc)
+			if _, err := fmt.Fprint(w, ":\n\n"); err != nil {
+				return
+			}
+			if err := rc.Flush(); err != nil {
+				return
+			}
+			h.proxy.UpdateActivity(sess.ID)
 		case <-clientGone:
 			return
 		case <-h.done:
@@ -445,6 +467,14 @@ func (h *HTTPSink) handleStream(w http.ResponseWriter, r *http.Request) {
 	}
 }
 
+// armWrite bounds the next response write. Without it an SSE write is unbounded
+// and a peer that stops reading wedges its handler for as long as it stays open.
+func (h *HTTPSink) armWrite(rc *http.ResponseController) {
+	if h.writeTimeout > 0 {
+		_ = rc.SetWriteDeadline(time.Now().Add(h.writeTimeout))
+	}
+}
+
 // handleStatus provides a JSON status report
 func (h *HTTPSink) handleStatus(w http.ResponseWriter, r *http.Request) {
 	status := map[string]any{
@@ -459,6 +489,9 @@ func (h *HTTPSink) handleStatus(w http.ResponseWriter, r *http.Request) {
 			"auth":               h.auth.Describe(),
 			"active_clients":     h.activeClients.Load(),
 			"buffer_size":        h.config.BufferSize,
+			"client_buffer_size": h.config.ClientBufferSize,
+			"max_connections":    h.config.MaxConnections,
+			"write_timeout_ms":   h.config.WriteTimeoutMS,
 			"uptime_seconds":     int(time.Since(h.startTime).Seconds()),
 		},
 		"endpoints": map[string]string{
@@ -484,6 +517,9 @@ func (h *HTTPSink) GetStats() sink.SinkStats {
 		"host":               h.config.Host,
 		"port":               h.config.Port,
 		"buffer_size":        h.config.BufferSize,
+		"client_buffer_size": h.config.ClientBufferSize,
+		"max_connections":    h.config.MaxConnections,
+		"write_timeout_ms":   h.config.WriteTimeoutMS,
 		"tls":                h.tlsConfig != nil,
 		"dropped_writes":     h.droppedWrites.Load(),
 		"rejected_clients":   h.rejectedClients.Load(),
@@ -530,6 +566,12 @@ func (h *HTTPSink) authMiddleware(next http.Handler) http.Handler {
 	})
 }
 
+// streamHeadNotAllowed refuses a body-less read of a stream that is only a body
+func streamHeadNotAllowed(w http.ResponseWriter, _ *http.Request) {
+	w.Header().Set("Allow", http.MethodGet)
+	http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
+}
+
 // writeSSE frames a payload per the W3C SSE spec (multi-line safe)
 func writeSSE(w http.ResponseWriter, payload []byte) error {
 	for _, line := range splitLines(payload) {

internal/sink/http/http_test.go

@@ -0,0 +1,167 @@
+package http
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+
+	"logwisp/internal/session"
+	"logwisp/internal/sink"
+
+	"github.com/lixenwraith/log"
+)
+
+func TestStatusReportsQueueAndConnectionBounds(t *testing.T) {
+	manager := session.NewManager(time.Hour)
+	defer manager.Stop()
+	created, err := NewHTTPSinkPlugin(
+		"stream",
+		map[string]any{
+			"host":               "127.0.0.1",
+			"port":               int64(8081),
+			"buffer_size":        int64(4096),
+			"client_buffer_size": int64(512),
+			"max_connections":    int64(32),
+			"write_timeout_ms":   int64(5000),
+		},
+		log.NewLogger(),
+		session.NewProxy(manager, "stream"),
+	)
+	if err != nil {
+		t.Fatal(err)
+	}
+	httpSink, ok := created.(*HTTPSink)
+	if !ok {
+		t.Fatalf("sink type = %T", created)
+	}
+
+	recorder := httptest.NewRecorder()
+	httpSink.handleStatus(recorder, httptest.NewRequest("GET", "/status", nil))
+	if recorder.Code != 200 {
+		t.Fatalf("status code = %d", recorder.Code)
+	}
+	var response struct {
+		Server map[string]any `json:"server"`
+	}
+	if err := json.NewDecoder(recorder.Body).Decode(&response); err != nil {
+		t.Fatal(err)
+	}
+	for key, want := range map[string]float64{
+		"buffer_size":        4096,
+		"client_buffer_size": 512,
+		"max_connections":    32,
+		"write_timeout_ms":   5000,
+	} {
+		if got := response.Server[key]; got != want {
+			t.Errorf("server.%s = %v, want %v", key, got, want)
+		}
+	}
+
+	stats := httpSink.GetStats()
+	details := stats.Details
+	for key, want := range map[string]int64{
+		"buffer_size":        4096,
+		"client_buffer_size": 512,
+		"max_connections":    32,
+		"write_timeout_ms":   5000,
+	} {
+		if got := details[key]; got != want {
+			t.Errorf("details[%q] = %v, want %v", key, got, want)
+		}
+	}
+
+	var _ sink.Sink = httpSink
+}
+
+// A stream carrying nothing still refreshes its session. Log traffic is what
+// bumps activity otherwise, so a quiet source would idle-expire a healthy client
+// and the broker would evict it on the next entry.
+func TestQuietStreamRefreshesItsSession(t *testing.T) {
+	manager := session.NewManager(time.Hour)
+	defer manager.Stop()
+	created, err := NewHTTPSinkPlugin(
+		"stream",
+		map[string]any{"host": "127.0.0.1", "port": int64(18191), "write_timeout_ms": int64(5000)},
+		log.NewLogger(),
+		session.NewProxy(manager, "stream"),
+	)
+	if err != nil {
+		t.Fatal(err)
+	}
+	httpSink := created.(*HTTPSink)
+	httpSink.keepalive = 100 * time.Millisecond
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	if err := httpSink.Start(ctx); err != nil {
+		t.Fatal(err)
+	}
+	defer httpSink.Stop()
+
+	resp, err := http.Get("http://127.0.0.1:18191/stream")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer resp.Body.Close()
+
+	activity := func() time.Time {
+		for _, s := range manager.GetActiveSessions() {
+			return s.LastActivity
+		}
+		t.Fatal("no session for the connected client")
+		return time.Time{}
+	}
+
+	deadline := time.Now().Add(2 * time.Second)
+	for activity().IsZero() && time.Now().Before(deadline) {
+		time.Sleep(10 * time.Millisecond)
+	}
+	before := activity()
+
+	// No events are sent for several keepalive periods.
+	time.Sleep(350 * time.Millisecond)
+	if after := activity(); !after.After(before) {
+		t.Fatalf("last activity %v did not advance on a silent stream", after)
+	}
+}
+
+// HEAD on the stream path is refused rather than served from the GET pattern:
+// its body writes are discarded, so the client it would register never reads.
+func TestHeadOnStreamPathIsRefused(t *testing.T) {
+	manager := session.NewManager(time.Hour)
+	defer manager.Stop()
+	created, err := NewHTTPSinkPlugin(
+		"stream",
+		map[string]any{"host": "127.0.0.1", "port": int64(18192)},
+		log.NewLogger(),
+		session.NewProxy(manager, "stream"),
+	)
+	if err != nil {
+		t.Fatal(err)
+	}
+	httpSink := created.(*HTTPSink)
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	if err := httpSink.Start(ctx); err != nil {
+		t.Fatal(err)
+	}
+	defer httpSink.Stop()
+
+	resp, err := http.Head("http://127.0.0.1:18192/stream")
+	if err != nil {
+		t.Fatal(err)
+	}
+	resp.Body.Close()
+	if resp.StatusCode != http.StatusMethodNotAllowed {
+		t.Fatalf("HEAD /stream = %d, want %d", resp.StatusCode, http.StatusMethodNotAllowed)
+	}
+	if got := resp.Header.Get("Allow"); got != http.MethodGet {
+		t.Errorf("Allow = %q, want %q", got, http.MethodGet)
+	}
+	if n := manager.GetSessionCount(); n != 0 {
+		t.Errorf("sessions after HEAD = %d, want 0", n)
+	}
+}

internal/source/file/file.go

@@ -10,6 +10,7 @@ import (
 	"strings"
 	"sync"
 	"sync/atomic"
+	"syscall"
 	"time"
 
 	"logwisp/internal/config"
@@ -61,6 +62,7 @@ const (
 	DefaultFileSourcePattern         = "*"
 	DefaultFileSourceCheckIntervalMS = 100
 	MinFileSourceCheckIntervalMS     = 10
+	DefaultFileSourceFrom            = "end"
 )
 
 // NewFileSourcePlugin creates a file source through plugin factory
@@ -90,6 +92,11 @@ func NewFileSourcePlugin(
 	} else if opts.CheckIntervalMS < MinFileSourceCheckIntervalMS {
 		return nil, fmt.Errorf("check_interval_ms: must be >= %d", MinFileSourceCheckIntervalMS)
 	}
+	if opts.From == "" {
+		opts.From = DefaultFileSourceFrom
+	} else if err := lconfig.OneOf("start", "end")(opts.From); err != nil {
+		return nil, fmt.Errorf("from: %w", err)
+	}
 
 	// Create and return plugin instance
 	fs := &FileSource{
@@ -116,7 +123,9 @@ func NewFileSourcePlugin(
 		"component", "file_source",
 		"instance_id", id,
 		"directory", opts.Directory,
-		"pattern", opts.Pattern)
+		"pattern", opts.Pattern,
+		"raw", opts.Raw,
+		"from", opts.From)
 
 	return fs, nil
 }
@@ -261,7 +270,13 @@ func (fs *FileSource) ensureWatcher(path string) {
 		return
 	}
 
-	w := newFileWatcher(path, fs.publish, fs.logger)
+	w := newFileWatcher(path, fs.config.Raw, fs.config.From == "start", fs.publish, fs.logger)
+	// A rotation renames the file out from under its watcher, so the same inode
+	// reappears here under the archive name. Resume where it was left: from the
+	// start would re-emit every record the file has already delivered.
+	if position, ok := fs.readPosition(path); ok {
+		w.position = position
+	}
 	fs.watchers[path] = w
 
 	fs.logger.Debug("msg", "Created file watcher",
@@ -284,12 +299,40 @@ func (fs *FileSource) ensureWatcher(path string) {
 			}
 		}
 
-		fs.mu.Lock()
+		fs.removeWatcher(path, w)
-		delete(fs.watchers, path)
-		fs.mu.Unlock()
 	}()
 }
 
+// readPosition reports how far a running watcher has read the file now at path.
+// Callers hold fs.mu.
+func (fs *FileSource) readPosition(path string) (int64, bool) {
+	info, err := os.Stat(path)
+	if err != nil {
+		return 0, false
+	}
+	stat, ok := info.Sys().(*syscall.Stat_t)
+	if !ok {
+		return 0, false
+	}
+	for _, w := range fs.watchers {
+		if position, ok := w.readTo(stat.Ino); ok {
+			return position, true
+		}
+	}
+	return 0, false
+}
+
+// removeWatcher removes only the watcher that finished. A deleted file can be
+// recreated before its old watcher observes stop; in that case ensureWatcher
+// has already installed a replacement under the same path, which must survive.
+func (fs *FileSource) removeWatcher(path string, watcher *fileWatcher) {
+	fs.mu.Lock()
+	if fs.watchers[path] == watcher {
+		delete(fs.watchers, path)
+	}
+	fs.mu.Unlock()
+}
+
 // cleanupWatchers stops and removes watchers for files that no longer exist.
 func (fs *FileSource) cleanupWatchers() {
 	fs.mu.Lock()
@@ -361,4 +404,3 @@ func globToRegex(glob string) string {
 	regex = strings.ReplaceAll(regex, `\?`, `.`)
 	return "^" + regex + "$"
 }
-

internal/source/file/file_test.go

@@ -0,0 +1,84 @@
+package file
+
+import (
+	"context"
+	"os"
+	"path/filepath"
+	"syscall"
+	"testing"
+	"time"
+
+	"logwisp/internal/core"
+)
+
+func TestStoppedWatcherReturnsNormally(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "session.jsonl")
+	if err := os.WriteFile(path, nil, 0o600); err != nil {
+		t.Fatal(err)
+	}
+	watcher := newFileWatcher(path, true, true, func(_ core.LogEntry) {}, nil)
+	watcher.stop()
+
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+	if err := watcher.watch(ctx); err != nil {
+		t.Fatalf("stopped watcher returned error: %v", err)
+	}
+}
+
+func TestRemoveWatcherPreservesReplacement(t *testing.T) {
+	oldWatcher := &fileWatcher{}
+	replacement := &fileWatcher{}
+	source := &FileSource{
+		watchers: map[string]*fileWatcher{
+			"session.jsonl": replacement,
+		},
+	}
+
+	source.removeWatcher("session.jsonl", oldWatcher)
+	if got := source.watchers["session.jsonl"]; got != replacement {
+		t.Fatalf("replacement watcher = %p, want %p", got, replacement)
+	}
+
+	source.removeWatcher("session.jsonl", replacement)
+	if _, exists := source.watchers["session.jsonl"]; exists {
+		t.Fatal("finished watcher was not removed")
+	}
+}
+
+// A rotated file reappears under its archive name with the same inode. Its
+// replacement watcher resumes where the original stopped, so a `from = "start"`
+// source does not re-emit every record the file already delivered.
+func TestRotatedFileResumesInsteadOfReplaying(t *testing.T) {
+	dir := t.TempDir()
+	active := filepath.Join(dir, "session.jsonl")
+	if err := os.WriteFile(active, []byte("one\ntwo\n"), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	info, err := os.Stat(active)
+	if err != nil {
+		t.Fatal(err)
+	}
+	inode := info.Sys().(*syscall.Stat_t).Ino
+
+	archive := filepath.Join(dir, "session_260916_120000.jsonl")
+	if err := os.Rename(active, archive); err != nil {
+		t.Fatal(err)
+	}
+
+	for name, watcher := range map[string]*fileWatcher{
+		"still tailing the renamed inode": {inode: inode, position: 8},
+		"already moved on from it":        {inode: 99, prevInode: inode, prevPosition: 8},
+	} {
+		source := &FileSource{watchers: map[string]*fileWatcher{active: watcher}}
+		position, ok := source.readPosition(archive)
+		if !ok || position != 8 {
+			t.Errorf("%s: position = %d, ok = %v, want 8, true", name, position, ok)
+		}
+	}
+
+	unrelated := &FileSource{watchers: map[string]*fileWatcher{active: {inode: 99}}}
+	if _, ok := unrelated.readPosition(archive); ok {
+		t.Error("a file no watcher has read was treated as rotated")
+	}
+}

internal/source/file/file_watcher.go

@@ -34,6 +34,7 @@ type WatcherInfo struct {
 type fileWatcher struct {
 	directory    string
 	callback     func(core.LogEntry)
+	raw          bool
 	position     int64
 	size         int64
 	inode        uint64
@@ -41,17 +42,25 @@ type fileWatcher struct {
 	mu           sync.Mutex
 	stopped      bool
 	rotationSeq  int64
+	prevInode    uint64
+	prevPosition int64
 	entriesRead  atomic.Uint64
 	lastReadTime atomic.Value // time.Time
 	logger       *log.Logger
 }
 
-// newFileWatcher creates a new watcher for a specific file path
+// newFileWatcher creates a new watcher for a specific file path.
-func newFileWatcher(directory string, callback func(core.LogEntry), logger *log.Logger) *fileWatcher {
+// A start position of 0 reads an existing file whole; -1 seeks to its end.
+func newFileWatcher(directory string, raw, fromStart bool, callback func(core.LogEntry), logger *log.Logger) *fileWatcher {
+	position := int64(-1)
+	if fromStart {
+		position = 0
+	}
 	w := &fileWatcher{
 		directory: directory,
 		callback:  callback,
-		position:  -1,
+		raw:       raw,
+		position:  position,
 		logger:    logger,
 	}
 	w.lastReadTime.Store(time.Time{})
@@ -60,8 +69,8 @@ func newFileWatcher(directory string, callback func(core.LogEntry), logger *log.
 
 // watch starts the main monitoring loop for the file
 func (w *fileWatcher) watch(ctx context.Context) error {
-	if err := w.seekToEnd(); err != nil {
+	if err := w.initPosition(); err != nil {
-		return fmt.Errorf("seekToEnd failed: %w", err)
+		return fmt.Errorf("initPosition failed: %w", err)
 	}
 
 	ticker := time.NewTicker(core.FileWatcherPollInterval)
@@ -73,7 +82,7 @@ func (w *fileWatcher) watch(ctx context.Context) error {
 			return ctx.Err()
 		case <-ticker.C:
 			if w.isStopped() {
-				return fmt.Errorf("watcher stopped")
+				return nil
 			}
 			if err := w.checkFile(); err != nil {
 				// Log error but continue watching
@@ -213,6 +222,9 @@ func (w *fileWatcher) checkFile() error {
 		w.mu.Lock()
 		w.rotationSeq++
 		seq := w.rotationSeq
+		// Retained for the source: the renamed file is about to be discovered
+		// under its archive name, and only this says how much of it was read.
+		w.prevInode, w.prevPosition = oldInode, oldPos
 		w.inode = currentInode
 		w.position = 0 // Reset position on rotation
 		w.mu.Unlock()
@@ -297,8 +309,9 @@ func (w *fileWatcher) checkFile() error {
 	return nil
 }
 
-// seekToEnd sets the initial read position to the end of the file
+// initPosition records the file's metadata and, unless the watcher was created
-func (w *fileWatcher) seekToEnd() error {
+// to read from the start, sets the initial read position to the end
+func (w *fileWatcher) initPosition() error {
 	file, err := os.Open(w.directory)
 	if err != nil {
 		if os.IsNotExist(err) {
@@ -322,8 +335,6 @@ func (w *fileWatcher) seekToEnd() error {
 	w.mu.Lock()
 	defer w.mu.Unlock()
 
-	// Keep existing position (including 0)
-	// First time initialization seeks to the end of the file
 	if w.position == -1 {
 		pos, err := file.Seek(0, io.SeekEnd)
 		if err != nil {
@@ -341,6 +352,23 @@ func (w *fileWatcher) seekToEnd() error {
 	return nil
 }
 
+// readTo reports how far this watcher read the given inode: the file it tails
+// now, or the one a rotation renamed out from under it.
+func (w *fileWatcher) readTo(inode uint64) (int64, bool) {
+	if inode == 0 {
+		return 0, false
+	}
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	switch inode {
+	case w.inode:
+		return w.position, true
+	case w.prevInode:
+		return w.prevPosition, true
+	}
+	return 0, false
+}
+
 // isStopped checks if the watcher has been instructed to stop
 func (w *fileWatcher) isStopped() bool {
 	w.mu.Lock()
@@ -348,36 +376,67 @@ func (w *fileWatcher) isStopped() bool {
 	return w.stopped
 }
 
-// parseLine attempts to parse a line as JSON, falling back to plain text
+// parseLine converts a line into an entry, as JSON when nothing would be lost
 func (w *fileWatcher) parseLine(line string) core.LogEntry {
-	var jsonLog struct {
+	if w.raw {
-		Time    string          `json:"time"`
+		// Newline restored: sinks write the payload as it stands
-		Level   string          `json:"level"`
-		Message string          `json:"msg"`
-		Fields  json.RawMessage `json:"fields"`
-	}
-
-	if err := json.Unmarshal([]byte(line), &jsonLog); err == nil {
-		timestamp, err := time.Parse(time.RFC3339Nano, jsonLog.Time)
-		if err != nil {
-			timestamp = time.Now()
-		}
-
 		return core.LogEntry{
-			Time:    timestamp,
+			Time:    time.Now(),
 			Source:  filepath.Base(w.directory),
-			Level:   jsonLog.Level,
+			Level:   source.ExtractLogLevel(line),
-			Message: jsonLog.Message,
+			Message: line + "\n",
-			Fields:  jsonLog.Fields,
 		}
 	}
 
-	level := source.ExtractLogLevel(line)
+	if entry, ok := w.parseJSON(line); ok {
+		return entry
+	}
 
 	return core.LogEntry{
 		Time:    time.Now(),
 		Source:  filepath.Base(w.directory),
-		Level:   level,
+		Level:   source.ExtractLogLevel(line),
 		Message: line,
 	}
 }
+
+// parseJSON decodes a line into the entry envelope. A top-level key LogEntry
+// cannot carry refuses the whole line, so a richer record reaches the pipeline
+// as text rather than silently reduced to the four keys kept here.
+func (w *fileWatcher) parseJSON(line string) (core.LogEntry, bool) {
+	if len(line) == 0 || line[0] != '{' {
+		return core.LogEntry{}, false
+	}
+
+	var obj map[string]json.RawMessage
+	if err := json.Unmarshal([]byte(line), &obj); err != nil || len(obj) == 0 {
+		return core.LogEntry{}, false
+	}
+
+	entry := core.LogEntry{Time: time.Now(), Source: filepath.Base(w.directory)}
+	for key, val := range obj {
+		var err error
+		switch key {
+		case "time":
+			var ts string
+			if json.Unmarshal(val, &ts) == nil {
+				if t, terr := time.Parse(time.RFC3339Nano, ts); terr == nil {
+					entry.Time = t
+				}
+			}
+		case "level":
+			err = json.Unmarshal(val, &entry.Level)
+		case "msg":
+			err = json.Unmarshal(val, &entry.Message)
+		case "fields":
+			entry.Fields = val
+		default:
+			return core.LogEntry{}, false
+		}
+		if err != nil {
+			return core.LogEntry{}, false
+		}
+	}
+
+	return entry, true
+}

test/passthrough-test.sh

@@ -0,0 +1,92 @@
+#!/usr/bin/env bash
+# logwisp file source pass-through test
+#
+#   file src (raw, from=start) --> raw format --> file sink   byte-exact relay
+#   file src (defaults)        --> raw format --> file sink   no key dropped
+#
+# The fixture is a record whose envelope is wider than time/level/msg/fields,
+# which is what the narrow JSON branch used to reduce to an empty message.
+#
+# Usage: ./passthrough-test.sh
+# Requires: bash 5+, coreutils (timeout). Linux dev host only.
+
+set -u
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+BIN="${LOGWISP_BIN:-$SCRIPT_DIR/../bin/logwisp}"
+RUN="$SCRIPT_DIR/run/passthrough"
+
+[[ -x $BIN ]] || { echo "no binary at $BIN; run make build" >&2; exit 1; }
+
+rm -rf "$RUN"
+mkdir -p "$RUN/in" "$RUN/out-raw" "$RUN/out-parsed"
+
+cat > "$RUN/in/wide.jsonl" <<'EOF'
+{"time":"2026-09-08T21:35:44.178372768-04:00","level":"INFO","sub":"app","run":0,"tick":0,"frame":0,"fields":{"msg":"init begin","mode":"play"}}
+{"time":"2026-09-08T21:35:44.178581366-04:00","level":"PROC","run":0,"tick":0,"frame":0,"fields":{"seq":1,"seed":1788917744178374836}}
+plain text line, not JSON at all
+EOF
+
+conf() { # sink_dir raw
+	cat <<EOF
+quiet = true
+status_reporter = false
+[logging]
+output = "stderr"
+level = "error"
+[[pipelines]]
+name = "passthrough"
+[pipelines.flow.format]
+type = "raw"
+sanitizer_policy = "raw"
+[[pipelines.plugin_sources]]
+id = "wide"
+type = "file"
+[pipelines.plugin_sources.config]
+directory = "$RUN/in"
+pattern = "*.jsonl"
+raw = $2
+from = "start"
+[[pipelines.plugin_sinks]]
+id = "out"
+type = "file"
+[pipelines.plugin_sinks.config]
+directory = "$1"
+name = "relay"
+flush_interval_ms = 100
+EOF
+}
+
+conf "$RUN/out-raw" true > "$RUN/raw.toml"
+conf "$RUN/out-parsed" false > "$RUN/parsed.toml"
+
+for c in raw parsed; do
+	timeout 5 "$BIN" -c "$RUN/$c.toml" > "$RUN/$c.out" 2>&1
+done
+
+fail=0
+check() { # label condition_result
+	if (( $2 )); then echo "PASS: $1"; else echo "FAIL: $1"; fail=1; fi
+}
+
+# 1. raw = true relays the file byte for byte
+if diff -q "$RUN/in/wide.jsonl" "$RUN/out-raw/relay.log" > /dev/null; then
+	check "raw = true: output identical to input" 1
+else
+	check "raw = true: output identical to input" 0
+	diff "$RUN/in/wide.jsonl" "$RUN/out-raw/relay.log" | head -6
+fi
+
+# 2. the default parse keeps every key, JSON branch refused on the wide envelope
+n=$(grep -c '"sub":"app"' "$RUN/out-parsed/relay.log")
+check "defaults: wide envelope reaches the sink whole ($n line(s) carry sub)" $(( n == 1 ))
+n=$(grep -c '1788917744178374836' "$RUN/out-parsed/relay.log")
+check "defaults: large integers are not re-encoded through float64 ($n)" $(( n == 1 ))
+
+echo "================================================================"
+if (( fail == 0 )); then
+	echo "RESULT: ALL PASS"
+else
+	echo "RESULT: FAILURES — inspect $RUN/"
+fi
+exit "$fail"