feat: deepen NMEA 2000 integration

CONTEXT.md

@@ -0,0 +1,64 @@
+# Beacon NMEA 2000 Integration
+
+Beacon observes NMEA 2000 networks and moves messages between marine data endpoints while preserving enough wire truth for diagnosis, replay, and controlled forwarding.
+
+## Language
+
+**Bus endpoint**:
+A physical or remote connection to one NMEA 2000 network segment.
+_Avoid_: Interface, port, adapter when referring to the network connection as a whole
+
+**Device NAME**:
+The stable 64-bit ISO 11783 identity claimed by an NMEA 2000 device independently of its current source address.
+_Avoid_: Device ID, address
+
+**Appliance identity**:
+Beacon's persistent Device NAME and product/configuration information presented on every independently connected bus endpoint.
+_Avoid_: Random client identity
+
+**Envelope**:
+Beacon's canonical representation of one assembled NMEA 2000 message, containing raw wire data plus additive semantic metadata.
+_Avoid_: Packet, event
+
+**Connector route**:
+One configured path from a source through filters and a durable queue to a sink.
+_Avoid_: Pipeline, link
+
+**Pending delivery**:
+Messages after a connector route's delivery checkpoint that have not yet reached that route's declared delivery boundary.
+_Avoid_: Queue depth when retained history is included
+
+**Retained history**:
+Acknowledged and pending queue rows kept for replay until a connector route's retention limits prune them.
+_Avoid_: Backlog
+
+**Delivery class**:
+The sink capability that defines whether the route boundary is confirmed, resumable, or best-effort.
+_Avoid_: Reliability level
+
+**Bridge mode**:
+The connector route policy that chooses observe-only, semantic re-origination, or transparent wire-preserving forwarding.
+_Avoid_: Sink mode
+
+**Commissioning baseline**:
+An operator-approved inventory of expected Device NAMEs and product/configuration details for a vessel installation.
+_Avoid_: Discovered devices
+
+## Relationships
+
+- A **Bus endpoint** observes zero or more **Device NAMEs**
+- One **Appliance identity** participates on zero or more independent **Bus endpoints**
+- A **Connector route** consumes **Envelopes** and has exactly one **Bridge mode** and one **Delivery class**
+- **Pending delivery** is a subset of **Retained history**
+- A **Commissioning baseline** contains zero or more expected **Device NAMEs** per **Bus endpoint**
+
+## Example dialogue
+
+> **Dev:** "The queue shows 20,000 rows. Is the connector route behind?"
+> **Domain expert:** "Check pending delivery, not retained history. Those rows may already be acknowledged and kept only for replay."
+
+## Flagged ambiguities
+
+- "device id" previously meant both source address and ISO NAME — resolved: **Device NAME** is stable identity; source address is the current bus address.
+- "queue depth" previously counted all retained rows — resolved: route health reports **Pending delivery**, while replay storage reports **Retained history**.
+- "CAN forwarding" previously meant semantic re-origination — resolved: **Bridge mode** states whether source identity and unknown PGNs are preserved.

README.md

@@ -24,9 +24,13 @@ source --> [connector: CEL filters + durable buffer] --> sink
   2000 bus through a TCP gateway (`tcp_gateway`).
 - **Connectors** are the only thing that moves data — each names exactly
   one source and one sink, an optional list of CEL filter expressions, and
-  its own durable SQLite-backed buffer that absorbs a slow or disconnected
-  sink without blocking the source. A source or sink with no connector
-  naming it does nothing.
+  its own durable SQLite-backed buffer. Each connector explicitly chooses
+  `semantic`, `transparent`, or `observe` bridge mode and reports a
+  confirmed, resumable, best-effort, or observe-only delivery boundary.
+
+Beacon also persists one stable N2K appliance NAME, exposes the complete PGN
+and field catalog with unit-scaled physical values, inventories devices by
+their stable NAME, and generates commissioning reports with SocketCAN health.
 
 Every configuration write (via the API or the UI) applies immediately —
 validated, persisted, and reconciled against the running system in one
@@ -99,9 +103,9 @@ never risks the surface you use to fix it:
 
 | Surface | Where | Notes |
 |---|---|---|
-| Web UI | admin address, `/ui` (`/` and `/ui` both redirect to `/ui/dashboard`) | dashboard, sources/sinks/connectors CRUD, live stats |
+| Web UI | admin address, `/ui` (`/` and `/ui` both redirect to `/ui/dashboard`) | routing, pending/retained state, bus diagnostics, device commissioning |
 | Manual | admin address, `/ui/docs` (`/docs` 301s there) | getting started, CAN setup, concepts, filters, API, troubleshooting |
-| Config API | admin address, `/api/v1/...` | REST CRUD, filter validation, export/import, live metrics, health, system info |
+| Config API | admin address, `/api/v1/...` | REST CRUD, live route state, PGN catalog, device inventory, commissioning report |
 | API reference | admin address, `/api/docs` (interactive, offline) and `/api/openapi.json` (OpenAPI 3.1) | self-describing — start here for scripting/agents |
 | Health | admin address, `/health` (mirrored at `/api/v1/health`) | JSON status, rolled up from every component |
 | Metrics | admin address, `/metrics` | Prometheus exposition |
@@ -125,6 +129,17 @@ and replay semantics, and a CEL cookbook, see `/ui/docs` (or
 [`internal/ui/docs/`](internal/ui/docs/)) and `/api/docs`; for ready-to-use
 starting points, see [`examples/`](examples/).
 
+Bridge mode is selected on each connector. Omitted mode means `semantic`:
+
+```json
+{"id":"bridge","name":"CAN 0 to CAN 1","source_id":"can0","sink_id":"can1","mode":"transparent","forward_management":false,"filters":[],"buffer":{},"enabled":true}
+```
+
+Transparent mode currently requires a SocketCAN sink, rejects a source and
+sink on the same interface, preserves source/destination/priority/raw payload
+including unknown PGNs, and blocks address-claim and other management PGNs
+unless `forward_management` is deliberately enabled.
+
 Gateway and replay source examples:
 
 ```json

docs/adr/0001-route-delivery-boundaries.md

@@ -0,0 +1,3 @@
+# Make connector delivery boundaries explicit
+
+Beacon reports pending delivery relative to each connector checkpoint and reports retained replay history separately. Confirmed sinks advance after a successful write, resumable sinks advance after the message enters their replayable stream, and best-effort sinks advance after dispatch while recording that downstream receipt is not known; this preserves useful fan-out behavior without labeling dispatch as confirmed delivery.

docs/adr/0002-persist-one-appliance-identity.md

@@ -0,0 +1,3 @@
+# Persist one appliance identity
+
+Beacon generates its experimental ISO 11783 identity number once in SQLite and reuses the resulting appliance NAME across restarts and independently connected bus endpoints. Assigned manufacturer/product codes and certification claims remain explicit configuration concerns rather than values Beacon fabricates, so development builds are stable without impersonating a certified product.

docs/adr/0003-separate-semantic-and-transparent-bridges.md

@@ -0,0 +1,3 @@
+# Separate semantic and transparent bridges
+
+Semantic bridge mode decodes and re-originates messages through Beacon's claimed N2K identity, while transparent bridge mode writes reconstructed CAN frames with the original priority, PGN, source, destination, and payload. Transparent forwarding is initially limited to SocketCAN sinks, rejects same-interface loops, and excludes network-management PGNs unless explicitly enabled because claiming and transport-control traffic cannot safely be copied between arbitrary segments by default.

docs/adr/0004-keep-wire-values-canonical.md

@@ -0,0 +1,3 @@
+# Keep wire values canonical
+
+The envelope retains raw bytes and raw-tick JSON payloads as its canonical lossless representation, then adds PGN catalog identity, decode status, ingress provenance, stable Device NAME, and unit-scaled physical values. This keeps replay and filters backward compatible while giving operators and integrations ergonomic engineering values without destructive conversion.

internal/api/api.go

@@ -16,10 +16,21 @@ import (
 	"github.com/danielgtaylor/huma/v2/adapters/humachi"
 	"github.com/go-chi/chi/v5"
 
+	"github.com/open-ships/beacon/internal/bus"
 	"github.com/open-ships/beacon/internal/config"
+	"github.com/open-ships/beacon/internal/identity"
+	"github.com/open-ships/beacon/internal/inventory"
 	"github.com/open-ships/beacon/internal/stats"
+	"github.com/open-ships/beacon/internal/supervisor"
 )
 
+type RuntimeInfo struct {
+	Identity  identity.Appliance
+	Devices   func() []bus.DeviceInfo
+	Inventory *inventory.Registry
+	Statuses  func() []supervisor.Status
+}
+
 // New builds beacon's config REST API: a chi router with huma registered
 // on it. Every operation is registered with its full "/api/v1/..." path
 // (rather than relying on router-mount prefix stripping), so the returned
@@ -35,7 +46,7 @@ import (
 // log receives the underlying error whenever a handler is about to answer
 // 500 (the client only ever sees a sanitized "internal error" body); nil
 // defaults to slog.Default(), the same convention as config.NewService.
-func New(svc *config.Service, reg *stats.Registry, version string, log *slog.Logger) (http.Handler, huma.API) {
+func New(svc *config.Service, reg *stats.Registry, version string, log *slog.Logger, runtimeInfo ...RuntimeInfo) (http.Handler, huma.API) {
 	if log == nil {
 		log = slog.Default()
 	}
@@ -59,12 +70,18 @@ func New(svc *config.Service, reg *stats.Registry, version string, log *slog.Log
 	humaConfig.SchemasPath = "/api/schemas"
 
 	humaAPI := humachi.New(router, humaConfig)
+	var runtime RuntimeInfo
+	if len(runtimeInfo) > 0 {
+		runtime = runtimeInfo[0]
+	}
 
 	registerSourceRoutes(humaAPI, svc, log)
 	registerSinkRoutes(humaAPI, svc, log)
 	registerConnectorRoutes(humaAPI, svc, log)
 	registerFilterRoutes(humaAPI, svc, log)
-	registerSystemInfoRoutes(humaAPI, version)
+	registerCatalogRoutes(humaAPI)
+	registerSystemInfoRoutes(humaAPI, version, runtime)
+	registerCommissioningRoutes(humaAPI, runtime, reg)
 	registerMetricsRoutes(humaAPI, svc, reg, log)
 	registerConfigIORoutes(humaAPI, svc, log)
 	registerHealthRoutes(humaAPI, svc)

internal/api/catalog.go

@@ -0,0 +1,45 @@
+package api
+
+import (
+	"context"
+	"net/http"
+
+	"github.com/danielgtaylor/huma/v2"
+
+	"github.com/open-ships/beacon/internal/n2kcatalog"
+)
+
+type catalogOutput struct {
+	Body struct {
+		PGNs []n2kcatalog.PGN `json:"pgns" doc:"Known NMEA 2000 PGNs and every decoded variant."`
+	}
+}
+
+type catalogPGNInput struct {
+	PGN uint32 `path:"pgn" doc:"NMEA 2000 Parameter Group Number."`
+}
+
+type catalogPGNOutput struct{ Body n2kcatalog.PGN }
+
+func registerCatalogRoutes(api huma.API) {
+	huma.Register(api, huma.Operation{
+		OperationID: "list-n2k-pgns", Method: http.MethodGet,
+		Path: "/api/v1/n2k/pgns", Summary: "List the runtime NMEA 2000 PGN and field catalog",
+	}, func(ctx context.Context, _ *struct{}) (*catalogOutput, error) {
+		out := &catalogOutput{}
+		out.Body.PGNs = n2kcatalog.List()
+		return out, nil
+	})
+
+	huma.Register(api, huma.Operation{
+		OperationID: "get-n2k-pgn", Method: http.MethodGet,
+		Path: "/api/v1/n2k/pgns/{pgn}", Summary: "Get one NMEA 2000 PGN and its variants",
+		Errors: []int{http.StatusNotFound},
+	}, func(ctx context.Context, in *catalogPGNInput) (*catalogPGNOutput, error) {
+		item, ok := n2kcatalog.Lookup(in.PGN)
+		if !ok {
+			return nil, huma.Error404NotFound("PGN not found")
+		}
+		return &catalogPGNOutput{Body: item}, nil
+	})
+}

internal/api/system.go

@@ -4,10 +4,15 @@ import (
 	"context"
 	"log/slog"
 	"net/http"
+	"strconv"
+	"time"
 
 	"github.com/danielgtaylor/huma/v2"
 
+	"github.com/open-ships/beacon/internal/bus"
 	"github.com/open-ships/beacon/internal/config"
+	"github.com/open-ships/beacon/internal/identity"
+	"github.com/open-ships/beacon/internal/inventory"
 	"github.com/open-ships/beacon/internal/model"
 	"github.com/open-ships/beacon/internal/stats"
 	"github.com/open-ships/beacon/internal/supervisor"
@@ -52,9 +57,12 @@ func registerFilterRoutes(api huma.API, svc *config.Service, log *slog.Logger) {
 
 type systemOutput struct {
 	Body struct {
-		Version       string   `json:"version" doc:"beacon server version."`
-		CANInterfaces []string `json:"can_interfaces" doc:"Detected SocketCAN network interfaces (Linux only; empty elsewhere)."`
-		SerialPorts   []string `json:"serial_ports" doc:"Detected USB-serial device paths (typical USB-CAN/NMEA-0183 adapters)."`
+		Version       string                 `json:"version" doc:"beacon server version."`
+		CANInterfaces []string               `json:"can_interfaces" doc:"Detected SocketCAN network interfaces (Linux only; empty elsewhere)."`
+		SerialPorts   []string               `json:"serial_ports" doc:"Detected USB-serial device paths (typical USB-CAN/NMEA-0183 adapters)."`
+		CANDetails    []sysinfo.CANInterface `json:"can_details" doc:"SocketCAN controller state, bitrate, counters, and sampled bus load."`
+		Identity      identity.Appliance     `json:"n2k_identity" doc:"Beacon's persistent ISO 11783/NMEA 2000 appliance identity."`
+		Devices       []bus.DeviceInfo       `json:"n2k_devices" doc:"Devices currently observed on attached NMEA 2000 networks."`
 	}
 }
 
@@ -65,7 +73,7 @@ type systemOutput struct {
 // directly for that same inventory rather than going through this package —
 // see internal/sysinfo's package doc comment for why discovery lives in its
 // own leaf package instead of here.
-func registerSystemInfoRoutes(api huma.API, version string) {
+func registerSystemInfoRoutes(api huma.API, version string, runtime RuntimeInfo) {
 	huma.Register(api, huma.Operation{
 		OperationID: "get-system",
 		Method:      http.MethodGet,
@@ -75,11 +83,113 @@ func registerSystemInfoRoutes(api huma.API, version string) {
 		out := &systemOutput{}
 		out.Body.Version = version
 		out.Body.CANInterfaces = sysinfo.DiscoverCAN()
+		out.Body.CANDetails = sysinfo.DiscoverCANDetails()
 		out.Body.SerialPorts = sysinfo.DiscoverSerial()
+		out.Body.Identity = runtime.Identity
+		if runtime.Devices != nil {
+			out.Body.Devices = runtime.Devices()
+		} else {
+			out.Body.Devices = []bus.DeviceInfo{}
+		}
 		return out, nil
 	})
 }
 
+type inventoryOutput struct {
+	Body struct {
+		Devices []inventory.Record `json:"devices"`
+	}
+}
+type baselineOutput struct {
+	Body struct {
+		Devices []inventory.Record `json:"devices"`
+		Status  string             `json:"status"`
+	}
+}
+type labelInput struct {
+	Endpoint string `query:"endpoint"`
+	Name     string `path:"name"`
+	Body     struct {
+		Label string `json:"label"`
+	}
+}
+type commissioningOutput struct {
+	Body struct {
+		GeneratedAt   time.Time                 `json:"generated_at"`
+		Identity      identity.Appliance        `json:"identity"`
+		CANInterfaces []sysinfo.CANInterface    `json:"can_interfaces"`
+		Devices       []inventory.Record        `json:"devices"`
+		Components    []supervisor.Status       `json:"components"`
+		Connectors    map[string]stats.Snapshot `json:"connectors"`
+	}
+}
+
+func registerCommissioningRoutes(api huma.API, runtime RuntimeInfo, reg *stats.Registry) {
+	huma.Register(api, huma.Operation{OperationID: "list-n2k-inventory", Method: http.MethodGet,
+		Path: "/api/v1/n2k/inventory", Summary: "List discovered devices and commissioning baseline status"},
+		func(ctx context.Context, _ *struct{}) (*inventoryOutput, error) {
+			out := &inventoryOutput{}
+			if runtime.Inventory != nil {
+				out.Body.Devices = runtime.Inventory.Records()
+			} else {
+				out.Body.Devices = []inventory.Record{}
+			}
+			return out, nil
+		})
+	huma.Register(api, huma.Operation{OperationID: "commit-n2k-baseline", Method: http.MethodPost,
+		Path: "/api/v1/n2k/inventory/baseline", Summary: "Accept currently online devices as the commissioning baseline"},
+		func(ctx context.Context, _ *struct{}) (*baselineOutput, error) {
+			if runtime.Inventory == nil {
+				return nil, huma.Error503ServiceUnavailable("inventory is unavailable")
+			}
+			if err := runtime.Inventory.CommitBaseline(ctx); err != nil {
+				return nil, huma.Error500InternalServerError("inventory baseline failed")
+			}
+			out := &baselineOutput{}
+			out.Body.Status = "committed"
+			out.Body.Devices = runtime.Inventory.Records()
+			return out, nil
+		})
+	huma.Register(api, huma.Operation{OperationID: "label-n2k-device", Method: http.MethodPut,
+		Path: "/api/v1/n2k/inventory/{name}/label", Summary: "Set an operator label for a stable Device NAME"},
+		func(ctx context.Context, in *labelInput) (*baselineOutput, error) {
+			if runtime.Inventory == nil {
+				return nil, huma.Error503ServiceUnavailable("inventory is unavailable")
+			}
+			name, err := strconv.ParseUint(in.Name, 16, 64)
+			if err != nil {
+				return nil, huma.Error422UnprocessableEntity("name must be a hexadecimal Device NAME")
+			}
+			if err := runtime.Inventory.SetLabel(ctx, in.Endpoint, name, in.Body.Label); err != nil {
+				return nil, huma.Error404NotFound("device not found")
+			}
+			out := &baselineOutput{}
+			out.Body.Status = "updated"
+			out.Body.Devices = runtime.Inventory.Records()
+			return out, nil
+		})
+	huma.Register(api, huma.Operation{OperationID: "get-n2k-commissioning-report", Method: http.MethodGet,
+		Path: "/api/v1/n2k/commissioning-report", Summary: "Generate a machine-readable NMEA 2000 commissioning report"},
+		func(ctx context.Context, _ *struct{}) (*commissioningOutput, error) {
+			out := &commissioningOutput{}
+			out.Body.GeneratedAt = time.Now().UTC()
+			out.Body.Identity = runtime.Identity
+			out.Body.CANInterfaces = sysinfo.DiscoverCANDetails()
+			out.Body.Connectors = reg.All()
+			if runtime.Inventory != nil {
+				out.Body.Devices = runtime.Inventory.Records()
+			} else {
+				out.Body.Devices = []inventory.Record{}
+			}
+			if runtime.Statuses != nil {
+				out.Body.Components = runtime.Statuses()
+			} else {
+				out.Body.Components = []supervisor.Status{}
+			}
+			return out, nil
+		})
+}
+
 // --- Live metrics ---
 
 type connectorMetricsOutput struct {