feat: ship n2k-native streams and richer source diagnostics

.air.toml

@@ -0,0 +1,22 @@
+root = "."
+tmp_dir = ".air"
+
+[build]
+  cmd = "go build -o ./.air/beacon ./cmd/beacon"
+  entrypoint = ["./.air/beacon"]
+  delay = 300
+  exclude_dir = [".air", ".git", "data", "node_modules", "playwright-report", "test-results"]
+  exclude_regex = ["_test.go"]
+  include_dir = ["cmd", "internal"]
+  include_ext = ["css", "go", "html", "js", "md", "svg"]
+  include_file = ["go.mod", "go.sum"]
+  kill_delay = "1s"
+  send_interrupt = true
+  stop_on_error = true
+
+[misc]
+  clean_on_exit = true
+
+[screen]
+  clear_on_rebuild = false
+  keep_scroll = true

.gitignore

@@ -43,8 +43,10 @@ config.toml
 beacon.db
 
 # Artifacts
+/data/
 /beacon
 /dist/
+/.air/
 
 # Tailwind/daisyUI standalone CLI + plugin bundles (dev-time download, not
 # source); see internal/ui/uisrc/README.md to re-fetch.

README.md

@@ -43,9 +43,9 @@ docker run --rm \
 
 Once it is running:
 
-- **Dashboard:** [localhost:2112/ui](http://localhost:2112/ui)
-- **Onboard manual:** [localhost:2112/ui/docs](http://localhost:2112/ui/docs)
-- **MCP agent reference:** [localhost:2112/ui/mcp](http://localhost:2112/ui/mcp)
+- **Dashboard:** [localhost:2112](http://localhost:2112)
+- **Onboard manual:** [localhost:2112/docs](http://localhost:2112/docs)
+- **MCP agent reference:** [localhost:2112/mcp/info](http://localhost:2112/mcp/info)
 - **Interactive API reference:** [localhost:2112/api/docs](http://localhost:2112/api/docs)
 - **Health:** [localhost:2112/health](http://localhost:2112/health)
 
@@ -211,21 +211,52 @@ read [Concepts](internal/ui/docs/03-concepts.md) and
 
 ## The envelope
 
-Every source is normalized into one JSON envelope used by buffers, filters, and
-network sinks. Its canonical representation keeps wire truth intact and adds
-semantic metadata rather than replacing raw values.
+MQTT, SSE, WebSocket, TCP, and NDJSON consumers receive exactly three top-level
+keys:
 
-| Field group | What it carries |
-|---|---|
-| Wire identity | `pgn`, `source`, `dest`, `priority`, `timestamp`, `raw` |
-| Route identity | `id`, `connector` after the message enters a route buffer |
-| Provenance | `observed_at`, `ingress`, `origin_ingress` |
-| Stable device identity | `device_name` and JavaScript-safe `device_name_hex` after address claim |
-| Catalog metadata | `pgn_name`, `variant`, `transport`, `manufacturer_code`, `decode` |
-| Values | Lossless raw-tick `payload` plus additive unit-scaled `physical` values |
+```json
+{
+  "payload": {
+    "info": {
+      "timestamp": "2026-07-25T12:00:00Z",
+      "receivedAt": "2026-07-25T12:00:00Z",
+      "adapterId": "socketcan:can0",
+      "networkId": "can0",
+      "direction": "received",
+      "priority": 2,
+      "pgn": 127250,
+      "sourceId": 12,
+      "targetId": null
+    },
+    "heading": 15708
+  },
+  "metadata": {
+    "id": 42,
+    "connector": "heading",
+    "observed_at": "2026-07-25T12:00:00Z",
+    "ingress": "can0",
+    "pgn_name": "Vessel Heading",
+    "decode": {"status": "decoded", "complete": true}
+  },
+  "raw": "XC9///////8="
+}
+```
+
+`payload` is the verbatim JSON representation of the decoded
+[`open-ships/n2k`](https://github.com/open-ships/n2k) Go struct, including every
+exported `MessageInfo` field such as receive time, transport timing, adapter,
+network, and direction. A consumer that knows the PGN can unmarshal `payload`
+directly into the corresponding `pgn` type. The raw-tick values are unchanged.
+
+`metadata` holds only what Beacon adds: queue and connector identity, ingress
+provenance, stable Device NAME, catalog/decode details, and physical values.
+
+`raw` is the assembled CAN payload as base64 bytes. It is top-level data, not
+metadata.
 
 Unknown PGNs still move through HTTP, TCP, MQTT, file, observe, and transparent
-routes with their raw bytes. See
+routes. Their `payload` is the complete `pgn.UnknownPGN` JSON and their original
+bytes remain available at top-level `raw`. See
 [ADR 0004](docs/adr/0004-keep-wire-values-canonical.md) for the compatibility
 rationale and [Concepts](internal/ui/docs/03-concepts.md#the-envelope) for the
 complete field contract.
@@ -307,10 +338,10 @@ take away the control surface used to repair it.
 
 | Surface | Default location | Purpose |
 |---|---|---|
-| Dashboard | `:2112/ui` | Route graph, pending/retained state, bus diagnostics, device commissioning |
-| Manual | `:2112/ui/docs` | Offline getting started, CAN setup, concepts, filters, API, troubleshooting |
-| MCP endpoint | `:2112/mcp` | Streamable HTTP tools for agent configuration, health, delivery, and source PGN statistics |
-| MCP reference | `:2112/ui/mcp` | Embedded connection guide, tool catalog, and call examples |
+| Dashboard | `:2112/dashboard` | Route graph, pending/retained state, bus diagnostics, device commissioning |
+| Manual | `:2112/docs` | Offline getting started, CAN setup, concepts, filters, API, troubleshooting |
+| MCP endpoint | `:2112/mcp` | Streamable HTTP tools for agent configuration, health, delivery, and source PGN metrics |
+| MCP reference | `:2112/mcp/info` | Embedded connection guide, tool catalog, and call examples |
 | REST API | `:2112/api/v1/...` | Configuration, live state, PGN catalog, inventory, commissioning |
 | API reference | `:2112/api/docs` | Interactive, embedded OpenAPI 3.1 documentation |
 | OpenAPI document | `:2112/api/openapi.json` | Machine-readable discovery for SDKs, scripts, and agents |
@@ -329,11 +360,10 @@ An MCP client can connect without a cloud relay or companion process:
 }
 ```
 
-The MCP server exposes twelve tools to read the complete configuration, create or
+The MCP server exposes tools to read the complete configuration, create or
 update sources, sinks, and connector routes, delete each entity type, and read
-health, delivery statistics, or per-source PGN traffic metrics. Operators and
-agents can approve or clear persistent expected-traffic baselines for each
-source. It uses the same validation, SQLite persistence, and hot reconciliation as the UI and REST API.
+health, delivery metrics, or per-source PGN traffic metrics. It uses the same
+validation, SQLite persistence, and hot reconciliation as the UI and REST API.
 The server, tool schemas, and reference page are all embedded in the Beacon
 binary; no internet connection, remote schema, CDN, or hosted MCP service is
 required.
@@ -348,6 +378,20 @@ moves, and out-of-range values visible after a restart. The scrape-safe subset
 is exported as `beacon_source_pgn_*` at `/metrics`; raw payloads and fingerprint
 identifiers stay in the UI/MCP response to avoid unbounded Prometheus labels.
 
+Every source and sink overview also has a stopped-by-default stream inspector.
+Start captures future source-received or sink-sent messages without consuming
+connector queues or blocking routing; Stop freezes the current browser-local
+capture. An optional CEL expression beside Start filters the server-side
+preview using the same `msg` fields as connector filters and can be changed
+while streaming without clearing captured rows. Clicking a JSON key or value
+shows a usable CEL expression for that field. The inspector switches between
+the verbatim decoded n2k JSON and assembled CAN payload bytes in hexadecimal,
+with exactly one message per line in either view. Captures can be copied,
+exported as n2k JSONL, or exported as one hexadecimal CAN payload per line,
+preserving message boundaries. The captured counter tracks the entire browser
+session even though only its latest 200 messages remain available for display,
+copy, and export.
+
 The admin API also exposes the complete PGN and field catalog, best-effort
 CAN/USB hardware discovery, stable Device NAME inventory, commissioning
 baselines, and a machine-readable commissioning report.

VERSION

@@ -1 +1 @@
-1.0.0
+1.1.0

examples/README.md

@@ -15,7 +15,7 @@ stops validating, that test fails.
 - **`navigation.json`** — the same shape as `minimal.json`, but the
   connector's filter allow-lists navigation PGNs only (heading, rapid
   position, COG/SOG, GNSS position: `127250`, `129025`, `129026`,
-  `129029`), served on `/nav`. See the filters page (`/ui/docs/filters`)
+  `129029`), served on `/nav`. See the filters page (`/docs/filters`)
   for how the filter list and CEL expressions work.
 - **`engine-room.json`** — one CAN source feeding *two* connectors, both
   filtered to engine PGNs (`127488`, `127489`, `127493`): one to an SSE
@@ -38,12 +38,12 @@ stops validating, that test fails.
   `file_path` must be an absolute path that exists on disk (the sink opens
   it but does not create missing parent directories) — create `/data/log`
   first, or change both paths to a directory you control. See the concepts
-  page (`/ui/docs/concepts`) for file sink delivery semantics, rotation, and
+  page (`/docs/concepts`) for file sink delivery semantics, rotation, and
   replaying a `candump` log with `canplayer`.
 - **`vcan-dev.json`** — identical to `minimal.json` but pointed at
   `vcan0` instead of a real interface, for developing and testing beacon
   with no CAN hardware attached. Bring the virtual interface up first (see
-  `/ui/docs/can-setup`):
+  `/docs/can-setup`):
 
   ```
   sudo modprobe vcan
@@ -58,7 +58,7 @@ Replace `can0` / `vcan0` with your actual interface name, and adjust sink
 paths/addresses and connector filters as needed — these are starting
 points, not fixed configurations. All buffer limits shown are optional; if
 a connector's `buffer` object sets nothing at all, `max_messages` defaults
-to 100000 (see `/ui/docs/concepts`).
+to 100000 (see `/docs/concepts`).
 
 ## Using an example
 
@@ -72,7 +72,7 @@ leave on the command line permanently.
 ```
 
 **Offline, against an existing database file** (the file must not be held
-open by a running beacon process — see `/ui/docs/api` for why):
+open by a running beacon process — see `/docs/api` for why):
 
 ```
 beacon import --db beacon.db examples/navigation.json           # replaces the whole config

go.mod

@@ -10,7 +10,7 @@ require (
 	github.com/go-chi/chi/v5 v5.3.1
 	github.com/google/cel-go v0.29.2
 	github.com/modelcontextprotocol/go-sdk v1.6.1
-	github.com/open-ships/n2k v0.3.0
+	github.com/open-ships/n2k v1.0.0
 	github.com/prometheus/client_golang v1.23.2
 	github.com/spf13/cobra v1.10.2
 	github.com/yuin/goldmark v1.8.4

go.sum

@@ -64,8 +64,8 @@ github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq
 github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ=
 github.com/ncruces/go-strftime v1.0.0 h1:HMFp8mLCTPp341M/ZnA4qaf7ZlsbTc+miZjCLOFAw7w=
 github.com/ncruces/go-strftime v1.0.0/go.mod h1:Fwc5htZGVVkseilnfgOVb9mKy6w1naJmn9CehxcKcls=
-github.com/open-ships/n2k v0.3.0 h1:7P/HGxAisNo4FIaFU/PCSTTLLR09rMe6ECOieBaGELQ=
-github.com/open-ships/n2k v0.3.0/go.mod h1:9LVVEnTKODTR+g/zIKO9H99S6ckObWy8ZeAMtcCu8vs=
+github.com/open-ships/n2k v1.0.0 h1:bMKVqRJHy9liPMZg9PPyZnM4GCkldeIZLXfCeRKCxJ4=
+github.com/open-ships/n2k v1.0.0/go.mod h1:xYolOtbRoo4IMuOXPsBgS+wyir25a2n5D9doSXcIUMs=
 github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
 github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=

internal/app/app.go

@@ -10,7 +10,6 @@ import (
 	"log/slog"
 	"net"
 	"net/http"
-	"net/url"
 	"os"
 	"sync"
 	"time"
@@ -122,8 +121,8 @@ func Run(ctx context.Context, opts Options) (*App, error) {
 		_ = st.Close()
 		return nil, fmt.Errorf("start data server: %w", err)
 	}
-	// Load approved expectations before sources start so even their first
-	// observed stream/change event is compared and persisted.
+	// Load recent source lifecycle events before sources start, then persist
+	// new events through the registry's bounded asynchronous writer.
 	if err := reg.AttachSourceMetricPersistence(ctx, st.DB()); err != nil {
 		_ = ds.Stop(ctx)
 		_ = st.Close()
@@ -175,24 +174,9 @@ func Run(ctx context.Context, opts Options) (*App, error) {
 	mux.HandleFunc("GET /health", a.handleHealth)
 	mux.Handle(mcpserver.EndpointPath, mcpserver.Handler(cfgSvc, reg, version, log))
 	mux.Handle("/api/", apiHandler)
-	mux.Handle("/ui/", uiHandler)
-	// The exact-path "/ui" mount (alongside the "/ui/" subtree mount above)
-	// is required for GET /ui to reach uiHandler's own "GET /ui" redirect
-	// route at all — see ui.Handler's doc comment for why.
-	mux.Handle("/ui", uiHandler)
-	// "/docs" and "/docs/{slug}" (spec §5) are permanent redirects to their
-	// /ui/docs equivalents (internal/ui/docspages.go), not a second copy of
-	// the manual — "/docs" is one of model.ReservedPathPrefixes, so no HTTP
-	// sink config can ever collide with either pattern.
-	mux.HandleFunc("GET /docs", func(w http.ResponseWriter, r *http.Request) {
-		http.Redirect(w, r, "/ui/docs", http.StatusMovedPermanently)
-	})
-	mux.HandleFunc("GET /docs/{slug}", func(w http.ResponseWriter, r *http.Request) {
-		http.Redirect(w, r, "/ui/docs/"+url.PathEscape(r.PathValue("slug")), http.StatusMovedPermanently)
-	})
-	mux.HandleFunc("GET /{$}", func(w http.ResponseWriter, r *http.Request) {
-		http.Redirect(w, r, "/ui/dashboard", http.StatusFound)
-	})
+	// The UI owns the remaining root-level admin paths. More-specific API,
+	// MCP, health, and metrics patterns above continue to win.
+	mux.Handle("/", uiHandler)
 	a.adminSrv = &http.Server{Handler: mux, ReadHeaderTimeout: 10 * time.Second}
 	ln, err := net.Listen("tcp", opts.AdminAddr)
 	if err != nil {

internal/app/app_test.go

@@ -82,13 +82,7 @@ func TestHandleHealthRollup(t *testing.T) {
 	}
 }
 
-// TestDocsRedirectToUIDocs covers app.go's "/docs" and "/docs/{slug}"
-// permanent redirects (spec §5) to their internal/ui/docspages.go
-// equivalents — registered on the admin mux itself, not internal/ui's own
-// handler, so this is the one place that composition is exercised
-// end-to-end. "/docs" is one of model.ReservedPathPrefixes, so no sink
-// config can ever be written that would collide with either route.
-func TestDocsRedirectToUIDocs(t *testing.T) {
+func TestDocsRoutesAreServedAtRoot(t *testing.T) {
 	a := startTestApp(t)
 	client := &http.Client{
 		CheckRedirect: func(req *http.Request, via []*http.Request) error {
@@ -97,11 +91,12 @@ func TestDocsRedirectToUIDocs(t *testing.T) {
 	}
 
 	cases := []struct {
-		path string
-		want string
+		path       string
+		wantStatus int
+		want       string
 	}{
-		{"/docs", "/ui/docs"},
-		{"/docs/getting-started", "/ui/docs/getting-started"},
+		{"/docs", http.StatusFound, "/docs/getting-started"},
+		{"/docs/getting-started", http.StatusOK, ""},
 	}
 	for _, c := range cases {
 		t.Run(c.path, func(t *testing.T) {
@@ -110,8 +105,45 @@ func TestDocsRedirectToUIDocs(t *testing.T) {
 				t.Fatal(err)
 			}
 			defer func() { _ = resp.Body.Close() }()
-			if resp.StatusCode != http.StatusMovedPermanently {
-				t.Fatalf("status = %d, want %d", resp.StatusCode, http.StatusMovedPermanently)
+			if resp.StatusCode != c.wantStatus {
+				t.Fatalf("status = %d, want %d", resp.StatusCode, c.wantStatus)
+			}
+			if loc := resp.Header.Get("Location"); loc != c.want {
+				t.Fatalf("Location = %q, want %q", loc, c.want)
+			}
+		})
+	}
+}
+
+func TestAdminUIIsMountedAtRoot(t *testing.T) {
+	a := startTestApp(t)
+	client := &http.Client{
+		CheckRedirect: func(req *http.Request, via []*http.Request) error {
+			return http.ErrUseLastResponse
+		},
+	}
+
+	cases := []struct {
+		path       string
+		wantStatus int
+		want       string
+	}{
+		{"/", http.StatusFound, "/dashboard"},
+		{"/dashboard", http.StatusOK, ""},
+		{"/sources", http.StatusOK, ""},
+		{"/mcp/info", http.StatusOK, ""},
+		{"/ui", http.StatusNotFound, ""},
+		{"/ui/dashboard", http.StatusNotFound, ""},
+	}
+	for _, c := range cases {
+		t.Run(c.path, func(t *testing.T) {
+			resp, err := client.Get("http://" + a.AdminAddr() + c.path)
+			if err != nil {
+				t.Fatal(err)
+			}
+			defer func() { _ = resp.Body.Close() }()
+			if resp.StatusCode != c.wantStatus {
+				t.Fatalf("status = %d, want %d", resp.StatusCode, c.wantStatus)
 			}
 			if loc := resp.Header.Get("Location"); loc != c.want {
 				t.Fatalf("Location = %q, want %q", loc, c.want)

internal/e2e/api_test.go

@@ -224,9 +224,8 @@ func TestEndToEndAPIDrivenLifecycle(t *testing.T) {
 	}
 
 	// Step 4: connect an SSE client, inject heading + depth frames, and
-	// assert only the heading (127250) messages arrive, with the "info"
-	// object stripped from the payload (wire-freeze regression).
-	t.Log("step 4: SSE client observes only the filtered PGN, with info stripped")
+	// assert only complete heading (127250) n2k payloads arrive.
+	t.Log("step 4: SSE client observes complete n2k payloads for only the filtered PGN")
 	resp, err := http.Get(dataBase + "/nav")
 	if err != nil {
 		t.Fatal(err)
@@ -241,15 +240,15 @@ func TestEndToEndAPIDrivenLifecycle(t *testing.T) {
 
 	events := sseEvents(t, resp, 2)
 	for _, e := range events {
-		if e["pgn"].(float64) != 127250 {
-			t.Fatalf("filter leaked pgn %v", e["pgn"])
+		if consumerEnvelopePGN(t, e) != 127250 {
+			t.Fatalf("filter leaked event %v", e)
 		}
-		payload, ok := e["payload"].(map[string]any)
-		if !ok {
-			t.Fatalf("payload not an object: %v", e["payload"])
+		payload, metadata := consumerEnvelopeParts(t, e)
+		if payload["heading"] == nil {
+			t.Fatalf("SSE payload lost the n2k VesselHeading fields: %v", payload)
 		}
-		if _, hasInfo := payload["info"]; hasInfo {
-			t.Fatalf("SSE payload still contains info: %v", payload)
+		if metadata["connector"] != "heading" {
+			t.Fatalf("connector metadata is not nested under metadata: %v", metadata)
 		}
 	}
 	_ = resp.Body.Close()