Publish a live view failure event when the capture pipeline cannot start

server/internal/plugins/telemetry/manager.go

@@ -102,6 +102,22 @@ func (m *Manager) handleDisconnect(id string) {
 	})
 }
 
+// reportLiveViewFailure publishes a capture pipeline that could not start. The
+// code is the one the dashboard classifies on, so the failure is attributable to
+// an org and a session without reading the VM's log.
+func (m *Manager) reportLiveViewFailure(sessionID, code, message string) {
+	m.enqueue(eventPayload{
+		Type:        eventLiveViewFailed,
+		Category:    categoryConnection,
+		SourceEvent: "neko.live_view.pipeline_failed",
+		Data: map[string]any{
+			"session_id": sessionID,
+			"code":       code,
+			"message":    message,
+		},
+	})
+}
+
 func (m *Manager) enqueue(ev eventPayload) {
 	select {
 	case m.eventsCh <- ev:
@@ -134,9 +150,14 @@ func (m *Manager) worker() {
 }
 
 func (m *Manager) publish(ev eventPayload) {
+	category := ev.Category
+	if category == "" {
+		category = "system"
+	}
+
 	body := publishBody{
 		Type:     ev.Type,
-		Category: "system",
+		Category: category,
 		Source: publishSource{
 			Kind:  "local_process",
 			Event: ev.SourceEvent,
@@ -169,6 +190,7 @@ func (m *Manager) publish(ev eventPayload) {
 
 type eventPayload struct {
 	Type        string
+	Category    string
 	SourceEvent string
 	Data        map[string]any
 }

server/internal/plugins/telemetry/manager_test.go

@@ -2,6 +2,7 @@ package telemetry
 
 import (
 	"encoding/json"
+	"errors"
 	"io"
 	"net/http"
 	"net/http/httptest"
@@ -165,3 +166,62 @@ func connect(t *testing.T, sm types.SessionManager, id string) (types.Session, t
 	s.ConnectWebSocketPeer(p)
 	return s, p
 }
+
+func TestManager_ReportsLiveViewFailure(t *testing.T) {
+	srv, captured := newCapturingServer(t)
+	defer srv.Close()
+
+	sm := newSessionManager(t)
+	m := NewManager(sm, &Config{Enabled: true, Endpoint: srv.URL})
+	require.NoError(t, m.Start())
+	defer func() { require.NoError(t, m.Shutdown()) }()
+
+	m.reportLiveViewFailure("session-1", codeGPUEncoderUnavailable, `(pipeline error) no element "nvh264enc"`)
+
+	require.Eventually(t, func() bool {
+		return len(captured.snapshot()) >= 1
+	}, 2*time.Second, 10*time.Millisecond, "expected a failure event")
+
+	ev := captured.snapshot()[0]
+	require.Equal(t, "live_view_failed", ev.Type)
+	require.Equal(t, "connection", ev.Category)
+	require.Equal(t, "local_process", ev.Source.Kind)
+	require.Equal(t, "neko.live_view.pipeline_failed", ev.Source.Event)
+	require.Equal(t, "session-1", ev.Data["session_id"])
+	require.Equal(t, codeGPUEncoderUnavailable, ev.Data["code"])
+	require.NotEmpty(t, ev.Data["message"])
+}
+
+func TestReportLiveViewFailure_WithoutReporter(t *testing.T) {
+	setReporter(nil)
+
+	require.NotPanics(t, func() {
+		ReportLiveViewFailure("session-1", errors.New(`(pipeline error) no element "nvh264enc"`))
+	})
+	require.NotPanics(t, func() {
+		ReportLiveViewFailure("", errors.New("pipeline failed"))
+	})
+	require.NotPanics(t, func() {
+		ReportLiveViewFailure("session-1", nil)
+	})
+}
+
+func TestClassifyPipelineError(t *testing.T) {
+	cases := []struct {
+		name string
+		err  error
+		want string
+	}{
+		{"missing encoder element", errors.New(`(pipeline error) no element "nvh264enc"`), codeGPUEncoderUnavailable},
+		{"cuda context", errors.New("failed to create cuda context, ret: 0x2"), codeGPUEncoderUnavailable},
+		{"plugin load", errors.New("element plugin couldn't be loaded"), codeGPUEncoderUnavailable},
+		{"other pipeline failure", errors.New("webrtc stream not found"), codePipelineFailed},
+		{"no error", nil, codePipelineFailed},
+	}
+
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			require.Equal(t, c.want, classifyPipelineError(c.err))
+		})
+	}
+}

server/internal/plugins/telemetry/plugin.go

@@ -25,9 +25,11 @@ func (p *Plugin) Config() types.PluginConfig {
 
 func (p *Plugin) Start(m types.PluginManagers) error {
 	p.manager = NewManager(m.SessionManager, p.config)
+	setReporter(p.manager.reportLiveViewFailure)
 	return p.manager.Start()
 }
 
 func (p *Plugin) Shutdown() error {
+	setReporter(nil)
 	return p.manager.Shutdown()
 }

server/internal/plugins/telemetry/types.go

@@ -1,3 +1,70 @@
 package telemetry
 
+import (
+	"regexp"
+	"sync"
+)
+
 const PluginName = "telemetry"
+
+const (
+	// eventLiveViewFailed is published when a live-view capture pipeline cannot
+	// start, so the viewer connects and never receives a frame.
+	eventLiveViewFailed = "live_view_failed"
+
+	categoryConnection = "connection"
+
+	// Codes match the vocabulary the dashboard already classifies on, so the
+	// control plane does not need a second taxonomy.
+	codeGPUEncoderUnavailable = "gpu_encoder_unavailable"
+	codePipelineFailed        = "live_view_pipeline_failed"
+)
+
+// Matches the strings GStreamer and the NVIDIA stack produce when the encoder
+// cannot be created, which is what a live view hits when the GPU is already
+// holding an encoder session for the replay recorder.
+var gpuEncoderFailurePattern = regexp.MustCompile(
+	`(?i)cuctxcreate failed|failed to create cuda context|cuda_error_out_of_memory|insufficient memory on nvh264enc|nvh264enc|nvenc|element plugin couldn't be loaded`,
+)
+
+func classifyPipelineError(err error) string {
+	if err == nil {
+		return codePipelineFailed
+	}
+	if gpuEncoderFailurePattern.MatchString(err.Error()) {
+		return codeGPUEncoderUnavailable
+	}
+	return codePipelineFailed
+}
+
+var (
+	reporterMu sync.RWMutex
+	reporter   func(sessionID, code, message string)
+)
+
+func setReporter(fn func(sessionID, code, message string)) {
+	reporterMu.Lock()
+	defer reporterMu.Unlock()
+	reporter = fn
+}
+
+func currentReporter() func(sessionID, code, message string) {
+	reporterMu.RLock()
+	defer reporterMu.RUnlock()
+	return reporter
+}
+
+// ReportLiveViewFailure records a live-view pipeline failure for the session that
+// asked for the stream. The websocket handler owns both the session and the error,
+// so it reports through this hook instead of holding a plugin reference. No-op
+// when the plugin is disabled or has not started.
+func ReportLiveViewFailure(sessionID string, err error) {
+	if sessionID == "" || err == nil {
+		return
+	}
+	fn := currentReporter()
+	if fn == nil {
+		return
+	}
+	fn(sessionID, classifyPipelineError(err), err.Error())
+}

server/internal/websocket/handler/handler.go

@@ -4,6 +4,7 @@ import (
 	"github.com/rs/zerolog"
 	"github.com/rs/zerolog/log"
 
+	"github.com/m1k1o/neko/server/internal/plugins/telemetry"
 	"github.com/m1k1o/neko/server/pkg/types"
 	"github.com/m1k1o/neko/server/pkg/types/event"
 	"github.com/m1k1o/neko/server/pkg/types/message"
@@ -202,6 +203,13 @@ func (h *MessageHandlerCtx) Message(session types.Session, data types.WebSocketM
 			Str("event", data.Event).
 			Str("session_id", session.ID()).
 			Msg("message handler has failed")
+
+		// A failed signal request is how a capture pipeline that cannot start
+		// reaches the viewer: the encoder is unavailable, so no frames render and
+		// the client has nothing to report.
+		if data.Event == event.SIGNAL_REQUEST {
+			telemetry.ReportLiveViewFailure(session.ID(), err)
+		}
 	}
 
 	return true