feat: add rich source traffic observability

README.md

@@ -309,13 +309,13 @@ take away the control surface used to repair it.
 |---|---|---|
 | 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, and delivery statistics |
+| 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 |
 | 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 |
 | Health | `:2112/health` | Rolled-up component health; mirrored at `/api/v1/health` |
-| Metrics | `:2112/metrics` | Prometheus exposition |
+| Metrics | `:2112/metrics` | Prometheus exposition, including per-source/sender PGN traffic and value distributions |
 | SSE / WebSocket sinks | `:8080/<configured-path>` | Data and replay endpoints |
 | TCP sinks | Configured listener address | Live-only NDJSON stream |
 
@@ -329,12 +329,24 @@ An MCP client can connect without a cloud relay or companion process:
 }
 ```
 
-The MCP server exposes nine tools to read the complete configuration, create or
+The MCP server exposes twelve tools to read the complete configuration, create or
 update sources, sinks, and connector routes, delete each entity type, and read
-health or delivery statistics. 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.
+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.
+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.
+
+Each source overview groups traffic by source address and PGN, including PGNs
+Beacon cannot decode. It reports learned frequency and jitter, gaps and bursts,
+last-seen age, addressing and decode outcomes, traffic share and estimated CAN
+load, payload lengths, decoded-field quantiles/availability/rate-of-change, and
+bounded raw-wire fingerprints, entropy, byte ranges, and change masks. Approved
+baselines make missing streams, frequency drift, payload/decode changes, address
+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.
 
 The admin API also exposes the complete PGN and field catalog, best-effort
 CAN/USB hardware discovery, stable Device NAME inventory, commissioning

internal/app/app.go

@@ -95,12 +95,12 @@ func Run(ctx context.Context, opts Options) (*App, error) {
 		}
 	}
 
-	met, promHandler, err := metrics.New()
+	reg := stats.NewRegistry()
+	met, promHandler, err := metrics.New(reg)
 	if err != nil {
 		_ = st.Close()
 		return nil, fmt.Errorf("init metrics: %w", err)
 	}
-	reg := stats.NewRegistry()
 	appliance, err := identity.LoadOrCreate(ctx, st)
 	if err != nil {
 		_ = st.Close()
@@ -122,11 +122,19 @@ 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.
+	if err := reg.AttachSourceMetricPersistence(ctx, st.DB()); err != nil {
+		_ = ds.Stop(ctx)
+		_ = st.Close()
+		return nil, fmt.Errorf("load source metric history: %w", err)
+	}
 
 	sup := supervisor.New(st, busMgr, ds, log, met, reg)
 	if err := sup.Reconcile(ctx); err != nil {
 		sup.Stop()
 		_ = ds.Stop(ctx)
+		_ = reg.CloseSourceMetricPersistence(ctx)
 		_ = st.Close()
 		return nil, fmt.Errorf("initial reconcile: %w", err)
 	}
@@ -135,7 +143,6 @@ func Run(ctx context.Context, opts Options) (*App, error) {
 	if err := inv.Refresh(ctx); err != nil {
 		log.Warn("load N2K inventory", "err", err)
 	}
-
 	a := &App{log: log, st: st, ds: ds, sup: sup, reg: reg, cfgSvc: cfgSvc, identity: appliance, inv: inv}
 	invCtx, invCancel := context.WithCancel(context.Background())
 	a.invCancel = invCancel
@@ -293,5 +300,7 @@ func (a *App) Close(ctx context.Context) error {
 	}
 	a.sup.Stop() // connectors flush final checkpoints
 	_ = a.ds.Stop(ctx)
-	return a.st.Close()
+	persistenceErr := a.reg.CloseSourceMetricPersistence(ctx)
+	storeErr := a.st.Close()
+	return errors.Join(persistenceErr, storeErr)
 }

internal/mcpserver/server.go

@@ -42,6 +42,9 @@ var toolCatalog = []ToolInfo{
 	{Name: "delete_connector", Access: "delete", Description: "Delete a connector route and its route-owned queue."},
 	{Name: "get_health", Access: "read", Description: "Read rolled-up health and live source, sink, and connector states."},
 	{Name: "get_delivery_statistics", Access: "read", Description: "Read delivery rates, totals, pending delivery, retained history, limits, drops, and errors."},
+	{Name: "get_source_metrics", Access: "read", Description: "Inspect all PGNs by source and sender, including unknown raw payloads, timing, load, decode quality, addressing, gaps, anomalies, field distributions, approved baselines, and lifecycle events."},
+	{Name: "commit_source_traffic_baseline", Access: "write", Description: "Replace one source's persistent expected-traffic baseline with every PGN stream observed since Beacon started."},
+	{Name: "clear_source_traffic_baseline", Access: "delete", Description: "Clear one source's approved traffic baseline without deleting its observations or event history."},
 }
 
 // Catalog returns a copy so callers cannot mutate the registered tool list.
@@ -288,6 +291,29 @@ type deliveryStatisticsInput struct {
 	ConnectorID string `json:"connector_id,omitempty" jsonschema:"Optional connector route id. Omit to return every configured connector."`
 }
 
+type sourceMetricsInput struct {
+	SourceID      string  `json:"source_id,omitempty" jsonschema:"Optional configured source id. Omit to inspect every source."`
+	PGN           *uint32 `json:"pgn,omitempty" jsonschema:"Optional PGN number filter."`
+	SourceAddress *uint8  `json:"source_address,omitempty" jsonschema:"Optional NMEA 2000 source-address filter."`
+	EventLimit    int     `json:"event_limit,omitempty" jsonschema:"Recent lifecycle events per source to include; zero defaults to 50."`
+}
+
+type sourceMetricsOutput struct {
+	GeneratedAt time.Time                                `json:"generated_at"`
+	Sources     map[string][]stats.SourcePGNMetric       `json:"sources"`
+	Baselines   map[string][]stats.SourceTrafficBaseline `json:"baselines"`
+	Events      map[string][]stats.SourceMetricEvent     `json:"events"`
+}
+
+type sourceTrafficBaselineInput struct {
+	SourceID string `json:"source_id" jsonschema:"Configured source id."`
+}
+
+type sourceTrafficBaselineOutput struct {
+	SourceID  string                        `json:"source_id"`
+	Baselines []stats.SourceTrafficBaseline `json:"baselines"`
+}
+
 type deliveryStatisticsOutput struct {
 	Connectors map[string]deliveryStatistics `json:"connectors"`
 }
@@ -462,6 +488,105 @@ func registerTools(server *sdkmcp.Server, svc *config.Service, reg *stats.Regist
 			}
 			return nil, out, nil
 		})
+
+	sdkmcp.AddTool(server, tool("get_source_metrics", "Get source PGN metrics", readAnnotations),
+		func(ctx context.Context, _ *sdkmcp.CallToolRequest, in sourceMetricsInput) (*sdkmcp.CallToolResult, sourceMetricsOutput, error) {
+			out := sourceMetricsOutput{
+				GeneratedAt: time.Now().UTC(),
+				Sources:     map[string][]stats.SourcePGNMetric{},
+				Baselines:   map[string][]stats.SourceTrafficBaseline{},
+				Events:      map[string][]stats.SourceMetricEvent{},
+			}
+			eventLimit := in.EventLimit
+			if eventLimit <= 0 || eventLimit > 200 {
+				eventLimit = 50
+			}
+			if in.SourceID != "" {
+				if _, err := svc.GetSource(ctx, in.SourceID); err != nil {
+					return nil, sourceMetricsOutput{}, publicError(log, err)
+				}
+				out.Sources[in.SourceID] = filterSourceMetrics(reg.SourcePGNMetrics(in.SourceID), in)
+				out.Baselines[in.SourceID] = filterSourceBaselines(reg.SourceTrafficBaselines(in.SourceID), in)
+				out.Events[in.SourceID] = filterSourceMetricEvents(reg.SourceMetricEvents(in.SourceID, eventLimit), in)
+				return nil, out, nil
+			}
+			sources, err := svc.ListSources(ctx)
+			if err != nil {
+				return nil, sourceMetricsOutput{}, publicError(log, err)
+			}
+			all := reg.AllSourcePGNMetrics()
+			for _, source := range sources {
+				out.Sources[source.ID] = filterSourceMetrics(all[source.ID], in)
+				out.Baselines[source.ID] = filterSourceBaselines(reg.SourceTrafficBaselines(source.ID), in)
+				out.Events[source.ID] = filterSourceMetricEvents(reg.SourceMetricEvents(source.ID, eventLimit), in)
+			}
+			return nil, out, nil
+		})
+
+	sdkmcp.AddTool(server, tool("commit_source_traffic_baseline", "Set expected traffic baseline", writeAnnotations),
+		func(ctx context.Context, _ *sdkmcp.CallToolRequest, in sourceTrafficBaselineInput) (*sdkmcp.CallToolResult, sourceTrafficBaselineOutput, error) {
+			if _, err := svc.GetSource(ctx, in.SourceID); err != nil {
+				return nil, sourceTrafficBaselineOutput{}, publicError(log, err)
+			}
+			baselines, err := reg.CommitSourceTrafficBaseline(ctx, in.SourceID)
+			if err != nil {
+				return nil, sourceTrafficBaselineOutput{}, publicError(log, err)
+			}
+			return nil, sourceTrafficBaselineOutput{SourceID: in.SourceID, Baselines: baselines}, nil
+		})
+
+	sdkmcp.AddTool(server, tool("clear_source_traffic_baseline", "Clear source traffic baseline", deleteAnnotations),
+		func(ctx context.Context, _ *sdkmcp.CallToolRequest, in sourceTrafficBaselineInput) (*sdkmcp.CallToolResult, sourceTrafficBaselineOutput, error) {
+			if _, err := svc.GetSource(ctx, in.SourceID); err != nil {
+				return nil, sourceTrafficBaselineOutput{}, publicError(log, err)
+			}
+			if err := reg.ClearSourceTrafficBaseline(ctx, in.SourceID); err != nil {
+				return nil, sourceTrafficBaselineOutput{}, publicError(log, err)
+			}
+			return nil, sourceTrafficBaselineOutput{SourceID: in.SourceID, Baselines: []stats.SourceTrafficBaseline{}}, nil
+		})
+}
+
+func filterSourceMetrics(metrics []stats.SourcePGNMetric, in sourceMetricsInput) []stats.SourcePGNMetric {
+	out := make([]stats.SourcePGNMetric, 0, len(metrics))
+	for _, metric := range metrics {
+		if in.PGN != nil && metric.PGN != *in.PGN {
+			continue
+		}
+		if in.SourceAddress != nil && metric.SourceAddress != *in.SourceAddress {
+			continue
+		}
+		out = append(out, metric)
+	}
+	return out
+}
+
+func filterSourceBaselines(baselines []stats.SourceTrafficBaseline, in sourceMetricsInput) []stats.SourceTrafficBaseline {
+	out := make([]stats.SourceTrafficBaseline, 0, len(baselines))
+	for _, baseline := range baselines {
+		if in.PGN != nil && baseline.PGN != *in.PGN {
+			continue
+		}
+		if in.SourceAddress != nil && baseline.SourceAddress != *in.SourceAddress {
+			continue
+		}
+		out = append(out, baseline)
+	}
+	return out
+}
+
+func filterSourceMetricEvents(events []stats.SourceMetricEvent, in sourceMetricsInput) []stats.SourceMetricEvent {
+	out := make([]stats.SourceMetricEvent, 0, len(events))
+	for _, event := range events {
+		if in.PGN != nil && event.PGN != *in.PGN {
+			continue
+		}
+		if in.SourceAddress != nil && event.SourceAddress != *in.SourceAddress {
+			continue
+		}
+		out = append(out, event)
+	}
+	return out
 }
 
 func tool(name, title string, annotations *sdkmcp.ToolAnnotations) *sdkmcp.Tool {

internal/mcpserver/server_test.go

@@ -17,6 +17,7 @@ import (
 
 	"github.com/open-ships/beacon/internal/config"
 	"github.com/open-ships/beacon/internal/model"
+	"github.com/open-ships/beacon/internal/msg"
 	"github.com/open-ships/beacon/internal/stats"
 	"github.com/open-ships/beacon/internal/store"
 	"github.com/open-ships/beacon/internal/supervisor"
@@ -65,6 +66,10 @@ func newTestMCP(t *testing.T) testMCP {
 	}}
 	svc := config.NewService(st, rec, slog.New(slog.NewTextHandler(io.Discard, nil)))
 	reg := stats.NewRegistry()
+	if err := reg.AttachSourceMetricPersistence(context.Background(), st.DB()); err != nil {
+		_ = st.Close()
+		t.Fatal(err)
+	}
 	server := NewServer(svc, reg, "test", slog.New(slog.NewTextHandler(io.Discard, nil)))
 	clientTransport, serverTransport := sdkmcp.NewInMemoryTransports()
 	serverSession, err := server.Connect(context.Background(), serverTransport, nil)
@@ -82,6 +87,7 @@ func newTestMCP(t *testing.T) testMCP {
 	t.Cleanup(func() {
 		_ = clientSession.Close()
 		_ = serverSession.Close()
+		_ = reg.CloseSourceMetricPersistence(context.Background())
 		_ = st.Close()
 	})
 	return testMCP{svc: svc, stats: reg, client: clientSession, server: serverSession, store: st, rec: rec}
@@ -235,6 +241,87 @@ func TestConfigureInspectAndDeleteThroughTools(t *testing.T) {
 	}
 }
 
+func TestGetSourceMetricsReturnsAndFiltersSharedPGNStore(t *testing.T) {
+	tm := newTestMCP(t)
+	mustPutSource := model.Source{ID: "can0", Name: "CAN", Type: model.SourceSocketCAN, Interface: "can0"}
+	if err := tm.svc.PutSource(context.Background(), mustPutSource, true); err != nil {
+		t.Fatal(err)
+	}
+	tm.stats.RecordSource("can0", &msg.Envelope{
+		PGN: 127250, PGNName: "Vessel Heading", Source: 12,
+		Raw: []byte{1, 2, 3, 4, 5, 6, 7, 8}, Payload: json.RawMessage(`{"heading":1.5}`),
+	})
+	tm.stats.RecordSource("can0", &msg.Envelope{PGN: 128259, Source: 44, Raw: []byte{1, 2}})
+
+	result := callTool(t, tm.client, "get_source_metrics", map[string]any{
+		"source_id": "can0", "pgn": 127250, "source_address": 12,
+	})
+	if result.IsError {
+		t.Fatalf("get_source_metrics: %s", toolErrorText(result))
+	}
+	out := decodeStructured[sourceMetricsOutput](t, result)
+	streams, ok := out.Sources["can0"]
+	if !ok || len(streams) != 1 {
+		t.Fatalf("source metrics = %+v", out)
+	}
+	stream := streams[0]
+	if stream.PGN != 127250 || stream.SourceAddress != 12 || stream.PayloadBytesLast != 8 || stream.Messages != 1 {
+		t.Fatalf("filtered stream = %+v", stream)
+	}
+	if out.GeneratedAt.IsZero() {
+		t.Fatal("generated_at was not populated")
+	}
+
+	result = callTool(t, tm.client, "get_source_metrics", map[string]any{"source_id": "missing"})
+	if !result.IsError || !strings.Contains(toolErrorText(result), config.ErrNotFound.Error()) {
+		t.Fatalf("unknown source result = isError %v, content %q", result.IsError, toolErrorText(result))
+	}
+}
+
+func TestSourceTrafficBaselineToolsPersistAndReportEvents(t *testing.T) {
+	tm := newTestMCP(t)
+	if err := tm.svc.PutSource(context.Background(), model.Source{
+		ID: "can0", Name: "CAN", Type: model.SourceSocketCAN, Interface: "can0",
+	}, true); err != nil {
+		t.Fatal(err)
+	}
+	for i := 0; i < 4; i++ {
+		tm.stats.RecordSource("can0", &msg.Envelope{
+			PGN: 127250, PGNName: "Vessel Heading", Source: 12, Dest: 255, Priority: 2,
+			Raw: []byte{1, 2, 3, byte(i)}, Decode: msg.DecodeInfo{Status: "decoded", Complete: true},
+		})
+	}
+
+	result := callTool(t, tm.client, "commit_source_traffic_baseline", map[string]any{"source_id": "can0"})
+	if result.IsError {
+		t.Fatalf("commit baseline: %s", toolErrorText(result))
+	}
+	committed := decodeStructured[sourceTrafficBaselineOutput](t, result)
+	if len(committed.Baselines) != 1 || committed.Baselines[0].PGN != 127250 {
+		t.Fatalf("committed baseline = %+v", committed)
+	}
+
+	result = callTool(t, tm.client, "get_source_metrics", map[string]any{"source_id": "can0", "event_limit": 10})
+	if result.IsError {
+		t.Fatalf("get metrics after baseline: %s", toolErrorText(result))
+	}
+	out := decodeStructured[sourceMetricsOutput](t, result)
+	if len(out.Baselines["can0"]) != 1 || len(out.Events["can0"]) == 0 {
+		t.Fatalf("baseline/event output = %+v", out)
+	}
+	if got := out.Sources["can0"][0].BaselineStatus; got != "matching" {
+		t.Fatalf("baseline status = %q, want matching", got)
+	}
+
+	result = callTool(t, tm.client, "clear_source_traffic_baseline", map[string]any{"source_id": "can0"})
+	if result.IsError {
+		t.Fatalf("clear baseline: %s", toolErrorText(result))
+	}
+	if baselines := tm.stats.SourceTrafficBaselines("can0"); len(baselines) != 0 {
+		t.Fatalf("baselines after clear = %+v", baselines)
+	}
+}
+
 func TestValidationAndDependencyFailuresAreToolErrors(t *testing.T) {
 	tm := newTestMCP(t)