feat: add CDP pipeline, cdpmonitor, and wire into API service

Author: archandatta

server/cmd/api/api/api.go

@@ -9,7 +9,9 @@ import (
 	"sync"
 	"time"
 
+	"github.com/onkernel/kernel-images/server/lib/cdpmonitor"
 	"github.com/onkernel/kernel-images/server/lib/devtoolsproxy"
+	"github.com/onkernel/kernel-images/server/lib/events"
 	"github.com/onkernel/kernel-images/server/lib/logger"
 	"github.com/onkernel/kernel-images/server/lib/nekoclient"
 	oapi "github.com/onkernel/kernel-images/server/lib/oapi"
@@ -68,11 +70,24 @@ type ApiService struct {
 	// xvfbResizeMu serializes background Xvfb restarts to prevent races
 	// when multiple CDP fast-path resizes fire in quick succession.
 	xvfbResizeMu sync.Mutex
+
+	// CDP event pipeline and cdpMonitor.
+	eventsPipeline *events.Pipeline
+	cdpMonitor     *cdpmonitor.Monitor
+	monitorMu      sync.Mutex
 }
 
 var _ oapi.StrictServerInterface = (*ApiService)(nil)
 
-func New(recordManager recorder.RecordManager, factory recorder.FFmpegRecorderFactory, upstreamMgr *devtoolsproxy.UpstreamManager, stz scaletozero.Controller, nekoAuthClient *nekoclient.AuthClient) (*ApiService, error) {
+func New(
+	recordManager recorder.RecordManager,
+	factory recorder.FFmpegRecorderFactory,
+	upstreamMgr *devtoolsproxy.UpstreamManager,
+	stz scaletozero.Controller,
+	nekoAuthClient *nekoclient.AuthClient,
+	eventsPipeline *events.Pipeline,
+	displayNum int,
+) (*ApiService, error) {
 	switch {
 	case recordManager == nil:
 		return nil, fmt.Errorf("recordManager cannot be nil")
@@ -82,8 +97,12 @@ func New(recordManager recorder.RecordManager, factory recorder.FFmpegRecorderFa
 		return nil, fmt.Errorf("upstreamMgr cannot be nil")
 	case nekoAuthClient == nil:
 		return nil, fmt.Errorf("nekoAuthClient cannot be nil")
+	case eventsPipeline == nil:
+		return nil, fmt.Errorf("eventsPipeline cannot be nil")
 	}
 
+	mon := cdpmonitor.New(upstreamMgr, eventsPipeline.Publish, displayNum)
+
 	return &ApiService{
 		recordManager:     recordManager,
 		factory:           factory,
@@ -94,6 +113,8 @@ func New(recordManager recorder.RecordManager, factory recorder.FFmpegRecorderFa
 		stz:               stz,
 		nekoAuthClient:    nekoAuthClient,
 		policy:            &policy.Policy{},
+		eventsPipeline:    eventsPipeline,
+		cdpMonitor:        mon,
 	}, nil
 }
 
@@ -313,5 +334,9 @@ func (s *ApiService) ListRecorders(ctx context.Context, _ oapi.ListRecordersRequ
 }
 
 func (s *ApiService) Shutdown(ctx context.Context) error {
+	s.monitorMu.Lock()
+	s.cdpMonitor.Stop()
+	_ = s.eventsPipeline.Close()
+	s.monitorMu.Unlock()
 	return s.recordManager.StopAll(ctx)
 }

server/cmd/api/api/api_test.go

@@ -12,6 +12,7 @@ import (
 	"log/slog"
 
 	"github.com/onkernel/kernel-images/server/lib/devtoolsproxy"
+	"github.com/onkernel/kernel-images/server/lib/events"
 	"github.com/onkernel/kernel-images/server/lib/nekoclient"
 	oapi "github.com/onkernel/kernel-images/server/lib/oapi"
 	"github.com/onkernel/kernel-images/server/lib/recorder"
@@ -25,7 +26,7 @@ func TestApiService_StartRecording(t *testing.T) {
 
 	t.Run("success", func(t *testing.T) {
 		mgr := recorder.NewFFmpegManager()
-		svc, err := New(mgr, newMockFactory(), newTestUpstreamManager(), scaletozero.NewNoopController(), newMockNekoClient(t))
+		svc, err := New(mgr, newMockFactory(), newTestUpstreamManager(), scaletozero.NewNoopController(), newMockNekoClient(t), newEventsPipeline(), 0)
 		require.NoError(t, err)
 
 		resp, err := svc.StartRecording(ctx, oapi.StartRecordingRequestObject{})
@@ -39,7 +40,7 @@ func TestApiService_StartRecording(t *testing.T) {
 
 	t.Run("already recording", func(t *testing.T) {
 		mgr := recorder.NewFFmpegManager()
-		svc, err := New(mgr, newMockFactory(), newTestUpstreamManager(), scaletozero.NewNoopController(), newMockNekoClient(t))
+		svc, err := New(mgr, newMockFactory(), newTestUpstreamManager(), scaletozero.NewNoopController(), newMockNekoClient(t), newEventsPipeline(), 0)
 		require.NoError(t, err)
 
 		// First start should succeed
@@ -54,7 +55,7 @@ func TestApiService_StartRecording(t *testing.T) {
 
 	t.Run("custom ids don't collide", func(t *testing.T) {
 		mgr := recorder.NewFFmpegManager()
-		svc, err := New(mgr, newMockFactory(), newTestUpstreamManager(), scaletozero.NewNoopController(), newMockNekoClient(t))
+		svc, err := New(mgr, newMockFactory(), newTestUpstreamManager(), scaletozero.NewNoopController(), newMockNekoClient(t), newEventsPipeline(), 0)
 		require.NoError(t, err)
 
 		for i := 0; i < 5; i++ {
@@ -87,7 +88,7 @@ func TestApiService_StopRecording(t *testing.T) {
 
 	t.Run("no active recording", func(t *testing.T) {
 		mgr := recorder.NewFFmpegManager()
-		svc, err := New(mgr, newMockFactory(), newTestUpstreamManager(), scaletozero.NewNoopController(), newMockNekoClient(t))
+		svc, err := New(mgr, newMockFactory(), newTestUpstreamManager(), scaletozero.NewNoopController(), newMockNekoClient(t), newEventsPipeline(), 0)
 		require.NoError(t, err)
 
 		resp, err := svc.StopRecording(ctx, oapi.StopRecordingRequestObject{})
@@ -100,7 +101,7 @@ func TestApiService_StopRecording(t *testing.T) {
 		rec := &mockRecorder{id: "default", isRecordingFlag: true}
 		require.NoError(t, mgr.RegisterRecorder(ctx, rec), "failed to register recorder")
 
-		svc, err := New(mgr, newMockFactory(), newTestUpstreamManager(), scaletozero.NewNoopController(), newMockNekoClient(t))
+		svc, err := New(mgr, newMockFactory(), newTestUpstreamManager(), scaletozero.NewNoopController(), newMockNekoClient(t), newEventsPipeline(), 0)
 		require.NoError(t, err)
 		resp, err := svc.StopRecording(ctx, oapi.StopRecordingRequestObject{})
 		require.NoError(t, err)
@@ -115,7 +116,7 @@ func TestApiService_StopRecording(t *testing.T) {
 
 		force := true
 		req := oapi.StopRecordingRequestObject{Body: &oapi.StopRecordingJSONRequestBody{ForceStop: &force}}
-		svc, err := New(mgr, newMockFactory(), newTestUpstreamManager(), scaletozero.NewNoopController(), newMockNekoClient(t))
+		svc, err := New(mgr, newMockFactory(), newTestUpstreamManager(), scaletozero.NewNoopController(), newMockNekoClient(t), newEventsPipeline(), 0)
 		require.NoError(t, err)
 		resp, err := svc.StopRecording(ctx, req)
 		require.NoError(t, err)
@@ -129,7 +130,7 @@ func TestApiService_DownloadRecording(t *testing.T) {
 
 	t.Run("not found", func(t *testing.T) {
 		mgr := recorder.NewFFmpegManager()
-		svc, err := New(mgr, newMockFactory(), newTestUpstreamManager(), scaletozero.NewNoopController(), newMockNekoClient(t))
+		svc, err := New(mgr, newMockFactory(), newTestUpstreamManager(), scaletozero.NewNoopController(), newMockNekoClient(t), newEventsPipeline(), 0)
 		require.NoError(t, err)
 		resp, err := svc.DownloadRecording(ctx, oapi.DownloadRecordingRequestObject{})
 		require.NoError(t, err)
@@ -149,7 +150,7 @@ func TestApiService_DownloadRecording(t *testing.T) {
 		rec := &mockRecorder{id: "default", isRecordingFlag: true, recordingData: randomBytes(minRecordingSizeInBytes - 1)}
 		require.NoError(t, mgr.RegisterRecorder(ctx, rec), "failed to register recorder")
 
-		svc, err := New(mgr, newMockFactory(), newTestUpstreamManager(), scaletozero.NewNoopController(), newMockNekoClient(t))
+		svc, err := New(mgr, newMockFactory(), newTestUpstreamManager(), scaletozero.NewNoopController(), newMockNekoClient(t), newEventsPipeline(), 0)
 		require.NoError(t, err)
 		// will return a 202 when the recording is too small
 		resp, err := svc.DownloadRecording(ctx, oapi.DownloadRecordingRequestObject{})
@@ -179,7 +180,7 @@ func TestApiService_DownloadRecording(t *testing.T) {
 		rec := &mockRecorder{id: "default", recordingData: data}
 		require.NoError(t, mgr.RegisterRecorder(ctx, rec), "failed to register recorder")
 
-		svc, err := New(mgr, newMockFactory(), newTestUpstreamManager(), scaletozero.NewNoopController(), newMockNekoClient(t))
+		svc, err := New(mgr, newMockFactory(), newTestUpstreamManager(), scaletozero.NewNoopController(), newMockNekoClient(t), newEventsPipeline(), 0)
 		require.NoError(t, err)
 		resp, err := svc.DownloadRecording(ctx, oapi.DownloadRecordingRequestObject{})
 		require.NoError(t, err)
@@ -199,7 +200,7 @@ func TestApiService_Shutdown(t *testing.T) {
 	rec := &mockRecorder{id: "default", isRecordingFlag: true}
 	require.NoError(t, mgr.RegisterRecorder(ctx, rec), "failed to register recorder")
 
-	svc, err := New(mgr, newMockFactory(), newTestUpstreamManager(), scaletozero.NewNoopController(), newMockNekoClient(t))
+	svc, err := New(mgr, newMockFactory(), newTestUpstreamManager(), scaletozero.NewNoopController(), newMockNekoClient(t), newEventsPipeline(), 0)
 	require.NoError(t, err)
 
 	require.NoError(t, svc.Shutdown(ctx))
@@ -303,10 +304,16 @@ func newMockNekoClient(t *testing.T) *nekoclient.AuthClient {
 	return client
 }
 
+func newEventsPipeline() *events.Pipeline {
+	ring := events.NewRingBuffer(64)
+	fw := events.NewFileWriter(os.TempDir())
+	return events.NewPipeline(ring, fw)
+}
+
 func TestApiService_PatchChromiumFlags(t *testing.T) {
 	ctx := context.Background()
 	mgr := recorder.NewFFmpegManager()
-	svc, err := New(mgr, newMockFactory(), newTestUpstreamManager(), scaletozero.NewNoopController(), newMockNekoClient(t))
+	svc, err := New(mgr, newMockFactory(), newTestUpstreamManager(), scaletozero.NewNoopController(), newMockNekoClient(t), newEventsPipeline(), 0)
 	require.NoError(t, err)
 
 	// Test with valid flags

server/cmd/api/api/display_test.go

@@ -34,7 +34,7 @@ func testFFmpegFactory(t *testing.T, tempDir string) recorder.FFmpegRecorderFact
 
 func newTestServiceWithFactory(t *testing.T, mgr recorder.RecordManager, factory recorder.FFmpegRecorderFactory) *ApiService {
 	t.Helper()
-	svc, err := New(mgr, factory, newTestUpstreamManager(), scaletozero.NewNoopController(), newMockNekoClient(t))
+	svc, err := New(mgr, factory, newTestUpstreamManager(), scaletozero.NewNoopController(), newMockNekoClient(t), newEventsPipeline(), 0)
 	require.NoError(t, err)
 	return svc
 }

server/cmd/api/api/events.go

@@ -0,0 +1,34 @@
+package api
+
+import (
+	"net/http"
+
+	"github.com/google/uuid"
+	"github.com/onkernel/kernel-images/server/lib/logger"
+)
+
+// StartCapture handles POST /events/start.
+// Generates a new capture session ID, seeds the pipeline, then starts the
+// CDP monitor. If already running, the monitor is stopped and
+// restarted with a fresh session ID
+func (s *ApiService) StartCapture(w http.ResponseWriter, r *http.Request) {
+	s.monitorMu.Lock()
+	defer s.monitorMu.Unlock()
+
+	s.eventsPipeline.Start(uuid.New().String())
+
+	if err := s.cdpMonitor.Start(r.Context()); err != nil {
+		logger.FromContext(r.Context()).Error("failed to start CDP monitor", "err", err)
+		http.Error(w, "failed to start capture", http.StatusInternalServerError)
+		return
+	}
+	w.WriteHeader(http.StatusOK)
+}
+
+// StopCapture handles POST /events/stop
+func (s *ApiService) StopCapture(w http.ResponseWriter, r *http.Request) {
+	s.monitorMu.Lock()
+	defer s.monitorMu.Unlock()
+	s.cdpMonitor.Stop()
+	w.WriteHeader(http.StatusOK)
+}

server/cmd/api/main.go

@@ -24,6 +24,7 @@ import (
 	"github.com/onkernel/kernel-images/server/cmd/config"
 	"github.com/onkernel/kernel-images/server/lib/chromedriverproxy"
 	"github.com/onkernel/kernel-images/server/lib/devtoolsproxy"
+	"github.com/onkernel/kernel-images/server/lib/events"
 	"github.com/onkernel/kernel-images/server/lib/logger"
 	"github.com/onkernel/kernel-images/server/lib/nekoclient"
 	oapi "github.com/onkernel/kernel-images/server/lib/oapi"
@@ -90,12 +91,19 @@ func main() {
 		os.Exit(1)
 	}
 
+	// Construct events pipeline
+	eventsRing := events.NewRingBuffer(1024)
+	eventsFileWriter := events.NewFileWriter("/var/log")
+	eventsPipeline := events.NewPipeline(eventsRing, eventsFileWriter)
+
 	apiService, err := api.New(
 		recorder.NewFFmpegManager(),
 		recorder.NewFFmpegRecorderFactory(config.PathToFFmpeg, defaultParams, stz),
 		upstreamMgr,
 		stz,
 		nekoAuthClient,
+		eventsPipeline,
+		config.DisplayNum,
 	)
 	if err != nil {
 		slogger.Error("failed to create api service", "err", err)
@@ -120,6 +128,10 @@ func main() {
 		w.Header().Set("Content-Type", "application/json")
 		w.Write(jsonData)
 	})
+	// capture events
+	r.Post("/events/start", apiService.StartCapture)
+	r.Post("/events/stop", apiService.StopCapture)
+
 	// PTY attach endpoint (WebSocket) - not part of OpenAPI spec
 	// Uses WebSocket for bidirectional streaming, which works well through proxies.
 	r.Get("/process/{process_id}/attach", func(w http.ResponseWriter, r *http.Request) {

server/lib/cdpmonitor/monitor.go

@@ -0,0 +1,41 @@
+package cdpmonitor
+
+import (
+	"context"
+	"sync/atomic"
+
+	"github.com/onkernel/kernel-images/server/lib/events"
+)
+
+// UpstreamProvider abstracts *devtoolsproxy.UpstreamManager for testability.
+type UpstreamProvider interface {
+	Current() string
+	Subscribe() (<-chan string, func())
+}
+
+// PublishFunc publishes an Event to the pipeline.
+type PublishFunc func(ev events.Event)
+
+// Monitor manages a CDP WebSocket connection with auto-attach session fan-out.
+// Single-use per capture session: call Start to begin, Stop to tear down.
+type Monitor struct {
+	running atomic.Bool
+}
+
+// New creates a Monitor. displayNum is the X display for ffmpeg screenshots.
+func New(_ UpstreamProvider, _ PublishFunc, _ int) *Monitor {
+	return &Monitor{}
+}
+
+// IsRunning reports whether the monitor is actively capturing.
+func (m *Monitor) IsRunning() bool {
+	return m.running.Load()
+}
+
+// Start begins CDP capture. Restarts if already running.
+func (m *Monitor) Start(_ context.Context) error {
+	return nil
+}
+
+// Stop tears down the monitor. Safe to call multiple times.
+func (m *Monitor) Stop() {}

server/lib/events/event.go

@@ -3,6 +3,7 @@ package events
 import (
 	"encoding/json"
 	"log/slog"
+	"strings"
 )
 
 // maxS2RecordBytes is the maximum record size for the S2 event pipeline (1 MB).
@@ -50,7 +51,7 @@ const (
 // Event is the portable event schema. It contains only producer-emitted content;
 // pipeline metadata (seq, capture session) lives on the Envelope.
 type Event struct {
-	Ts          int64           `json:"ts"` // Unix microseconds (µs since epoch)
+	Ts          int64           `json:"ts"`
 	Type        string          `json:"type"`
 	Category    EventCategory   `json:"category"`
 	Source      Source          `json:"source"`
@@ -67,6 +68,26 @@ type Envelope struct {
 	Event            Event  `json:"event"`
 }
 
+// CategoryFor derives an EventCategory from an event type string.
+// It splits on the first underscore and maps the prefix to a category.
+func CategoryFor(eventType string) EventCategory {
+	prefix, _, _ := strings.Cut(eventType, "_")
+	switch prefix {
+	case "console":
+		return CategoryConsole
+	case "network":
+		return CategoryNetwork
+	case "page", "navigation", "dom", "target":
+		return CategoryPage
+	case "interaction", "layout", "scroll":
+		return CategoryInteraction
+	case "screenshot", "monitor":
+		return CategorySystem
+	default:
+		return CategorySystem
+	}
+}
+
 // truncateIfNeeded marshals env and returns the (possibly truncated) envelope.
 // If the envelope still exceeds maxS2RecordBytes after nulling data (e.g. huge
 // url or source.metadata), it is returned as-is — callers must handle nil data.