[kernel-1116] browser events: add external events (#227)

As per plan -> #217


<!-- CURSOR_SUMMARY -->
---

> [!NOTE]
> **Medium Risk**
> Introduces new public event ingestion/streaming endpoints and
refactors the core event/capture-session pipeline (seq semantics and
envelope shape), which could impact existing consumers and event
durability/ordering behavior.
> 
> **Overview**
> Adds **external event support** via new API endpoints: `POST
/events/publish` to inject validated events into the server’s event bus
and `GET /events/stream` to consume envelopes over SSE with
`Last-Event-ID` resume and keepalives.
> 
> Refactors the event pipeline by splitting responsibilities into a
process-wide `events.EventStream` (ring buffer + process-monotonic
sequence) and a new `capturesession.CaptureSession` (session lifecycle +
category filtering + stamping `capture_session_id` into
`event.source.metadata`), removing the old `events.CaptureSession` and
file-writing path. The OpenAPI-generated types/handlers are updated
accordingly (new schemas, renamed capture category enums, SSE responses
now set `Cache-Control: no-cache` and `X-Accel-Buffering: no`), along
with build tooling tweaks to patch the additional SSE method.
> 
> <sup>Reviewed by [Cursor Bugbot](https://cursor.com/bugbot) for commit
bb38112. Bugbot is set up for automated
code reviews on this repo. Configure
[here](https://www.cursor.com/dashboard/bugbot).</sup>
<!-- /CURSOR_SUMMARY -->

Author: archandatta

server/Makefile

@@ -19,7 +19,7 @@ oapi-generate:
 	openapi-down-convert --input openapi.yaml --output openapi-3.0.yaml
 	go tool oapi-codegen -config ./oapi-codegen.yaml ./openapi-3.0.yaml
 	@echo "Fixing oapi-codegen issue https://github.com/oapi-codegen/oapi-codegen/issues/1764..."
-	go run ./scripts/oapi/patch_sse_methods -file ./lib/oapi/oapi.go -expected-replacements 3
+	go run ./scripts/oapi/patch_sse_methods -file ./lib/oapi/oapi.go -expected-replacements 4
 	go fmt ./lib/oapi/oapi.go
 	go mod tidy
 

server/cmd/api/api/api.go

@@ -10,6 +10,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/kernel/kernel-images/server/lib/capturesession"
 	"github.com/kernel/kernel-images/server/lib/cdpmonitor"
 	"github.com/kernel/kernel-images/server/lib/devtoolsproxy"
 	"github.com/kernel/kernel-images/server/lib/events"
@@ -81,7 +82,8 @@ type ApiService struct {
 	xvfbResizeMu sync.Mutex
 
 	// CDP event pipeline and cdpMonitor.
-	captureSession  *events.CaptureSession
+	eventStream     *events.EventStream
+	captureSession  *capturesession.CaptureSession
 	cdpMonitor      cdpMonitorController
 	monitorMu       sync.Mutex
 	lifecycleCtx    context.Context
@@ -96,7 +98,8 @@ func New(
 	upstreamMgr *devtoolsproxy.UpstreamManager,
 	stz scaletozero.Controller,
 	nekoAuthClient *nekoclient.AuthClient,
-	captureSession *events.CaptureSession,
+	captureSession *capturesession.CaptureSession,
+	eventStream    *events.EventStream,
 	displayNum int,
 ) (*ApiService, error) {
 	switch {
@@ -110,6 +113,8 @@ func New(
 		return nil, fmt.Errorf("nekoAuthClient cannot be nil")
 	case captureSession == nil:
 		return nil, fmt.Errorf("captureSession cannot be nil")
+	case eventStream == nil:
+		return nil, fmt.Errorf("eventStream cannot be nil")
 	}
 
 	mon := cdpmonitor.New(upstreamMgr, captureSession.Publish, displayNum, slog.Default())
@@ -125,6 +130,7 @@ func New(
 		stz:               stz,
 		nekoAuthClient:    nekoAuthClient,
 		policy:            &policy.Policy{},
+		eventStream:       eventStream,
 		captureSession:    captureSession,
 		cdpMonitor:        mon,
 		lifecycleCtx:      ctx,
@@ -352,7 +358,6 @@ func (s *ApiService) Shutdown(ctx context.Context) error {
 	s.lifecycleCancel()
 	s.cdpMonitor.Stop()
 	s.captureSession.Stop()
-	_ = s.captureSession.Close()
 	s.monitorMu.Unlock()
 	return s.recordManager.StopAll(ctx)
 }

server/cmd/api/api/api_test.go

@@ -11,6 +11,7 @@ import (
 
 	"log/slog"
 
+	"github.com/kernel/kernel-images/server/lib/capturesession"
 	"github.com/kernel/kernel-images/server/lib/devtoolsproxy"
 	"github.com/kernel/kernel-images/server/lib/events"
 	"github.com/kernel/kernel-images/server/lib/nekoclient"
@@ -26,7 +27,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), newCaptureSession(t), 0)
+		svc, err := newSvc(t, mgr)
 		require.NoError(t, err)
 
 		resp, err := svc.StartRecording(ctx, oapi.StartRecordingRequestObject{})
@@ -40,7 +41,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), newCaptureSession(t), 0)
+		svc, err := newSvc(t, mgr)
 		require.NoError(t, err)
 
 		// First start should succeed
@@ -55,7 +56,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), newCaptureSession(t), 0)
+		svc, err := newSvc(t, mgr)
 		require.NoError(t, err)
 
 		for i := 0; i < 5; i++ {
@@ -88,7 +89,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), newCaptureSession(t), 0)
+		svc, err := newSvc(t, mgr)
 		require.NoError(t, err)
 
 		resp, err := svc.StopRecording(ctx, oapi.StopRecordingRequestObject{})
@@ -101,7 +102,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), newCaptureSession(t), 0)
+		svc, err := newSvc(t, mgr)
 		require.NoError(t, err)
 		resp, err := svc.StopRecording(ctx, oapi.StopRecordingRequestObject{})
 		require.NoError(t, err)
@@ -116,7 +117,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), newCaptureSession(t), 0)
+		svc, err := newSvc(t, mgr)
 		require.NoError(t, err)
 		resp, err := svc.StopRecording(ctx, req)
 		require.NoError(t, err)
@@ -130,7 +131,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), newCaptureSession(t), 0)
+		svc, err := newSvc(t, mgr)
 		require.NoError(t, err)
 		resp, err := svc.DownloadRecording(ctx, oapi.DownloadRecordingRequestObject{})
 		require.NoError(t, err)
@@ -150,7 +151,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), newCaptureSession(t), 0)
+		svc, err := newSvc(t, mgr)
 		require.NoError(t, err)
 		// will return a 202 when the recording is too small
 		resp, err := svc.DownloadRecording(ctx, oapi.DownloadRecordingRequestObject{})
@@ -180,7 +181,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), newCaptureSession(t), 0)
+		svc, err := newSvc(t, mgr)
 		require.NoError(t, err)
 		resp, err := svc.DownloadRecording(ctx, oapi.DownloadRecordingRequestObject{})
 		require.NoError(t, err)
@@ -200,7 +201,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), newCaptureSession(t), 0)
+	svc, err := newSvc(t, mgr)
 	require.NoError(t, err)
 
 	require.NoError(t, svc.Shutdown(ctx))
@@ -304,23 +305,26 @@ func newMockNekoClient(t *testing.T) *nekoclient.AuthClient {
 	return client
 }
 
-func newCaptureSession(t *testing.T) *events.CaptureSession {
+func newCaptureSession(t *testing.T) (*capturesession.CaptureSession, *events.EventStream) {
 	t.Helper()
-	cs, err := events.NewCaptureSession(events.CaptureSessionConfig{
-		LogDir:       t.TempDir(),
-		RingCapacity: 64,
-	})
+	es, err := events.NewEventStream(events.EventStreamConfig{RingCapacity: 64})
 	if err != nil {
 		t.Fatal(err)
 	}
-	t.Cleanup(func() { cs.Close() })
-	return cs
+	return capturesession.NewCaptureSession(es), es
+}
+
+// newSvc constructs an ApiService with a fresh capture session and event stream.
+func newSvc(t *testing.T, mgr recorder.RecordManager) (*ApiService, error) {
+	t.Helper()
+	cs, es := newCaptureSession(t)
+	return New(mgr, newMockFactory(), newTestUpstreamManager(), scaletozero.NewNoopController(), newMockNekoClient(t), cs, es, 0)
 }
 
 func TestApiService_PatchChromiumFlags(t *testing.T) {
 	ctx := context.Background()
 	mgr := recorder.NewFFmpegManager()
-	svc, err := New(mgr, newMockFactory(), newTestUpstreamManager(), scaletozero.NewNoopController(), newMockNekoClient(t), newCaptureSession(t), 0)
+	svc, err := newSvc(t, mgr)
 	require.NoError(t, err)
 
 	// Test with valid flags

server/cmd/api/api/capture_session.go

@@ -8,6 +8,7 @@ import (
 	"github.com/nrednav/cuid2"
 	oapi "github.com/kernel/kernel-images/server/lib/oapi"
 
+	"github.com/kernel/kernel-images/server/lib/capturesession"
 	"github.com/kernel/kernel-images/server/lib/events"
 	"github.com/kernel/kernel-images/server/lib/logger"
 )
@@ -119,26 +120,26 @@ func (s *ApiService) buildSessionResponse() oapi.CaptureSession {
 }
 
 // captureConfigFrom converts the optional StartCaptureSessionRequest body
-// into an events.CaptureConfig.
-func captureConfigFrom(body *oapi.StartCaptureSessionRequest) (events.CaptureConfig, error) {
+// into a capturesession.CaptureConfig.
+func captureConfigFrom(body *oapi.StartCaptureSessionRequest) (capturesession.CaptureConfig, error) {
 	if body == nil {
-		return events.CaptureConfig{}, nil
+		return capturesession.CaptureConfig{}, nil
 	}
 	return captureConfigFromOAPI(body.Config)
 }
 
-// captureConfigFromOAPI converts an oapi.CaptureConfig to events.CaptureConfig.
-func captureConfigFromOAPI(cfg *oapi.CaptureConfig) (events.CaptureConfig, error) {
+// captureConfigFromOAPI converts an oapi.CaptureConfig to capturesession.CaptureConfig.
+func captureConfigFromOAPI(cfg *oapi.CaptureConfig) (capturesession.CaptureConfig, error) {
 	if cfg == nil || cfg.Categories == nil {
-		return events.CaptureConfig{}, nil
+		return capturesession.CaptureConfig{}, nil
 	}
-	out := events.CaptureConfig{
+	out := capturesession.CaptureConfig{
 		Categories: make([]events.EventCategory, 0, len(*cfg.Categories)),
 	}
 	for _, c := range *cfg.Categories {
 		cat := events.EventCategory(c)
 		if !events.ValidCategory(cat) {
-			return events.CaptureConfig{}, fmt.Errorf("unknown category: %q", c)
+			return capturesession.CaptureConfig{}, fmt.Errorf("unknown category: %q", c)
 		}
 		out.Categories = append(out.Categories, cat)
 	}

server/cmd/api/api/capture_session_test.go

@@ -19,7 +19,7 @@ func TestCaptureConfigFrom(t *testing.T) {
 	})
 
 	t.Run("valid categories", func(t *testing.T) {
-		cats := []oapi.CaptureConfigCategories{oapi.Console, oapi.Network}
+		cats := []oapi.CaptureConfigCategories{oapi.CaptureConfigCategoriesConsole, oapi.CaptureConfigCategoriesNetwork}
 		body := &oapi.StartCaptureSessionRequest{
 			Config: &oapi.CaptureConfig{Categories: &cats},
 		}
@@ -64,7 +64,7 @@ func TestStartCaptureSession(t *testing.T) {
 
 	t.Run("success with config", func(t *testing.T) {
 		svc := newTestService(t, newMockRecordManager())
-		cats := []oapi.CaptureConfigCategories{oapi.Console}
+		cats := []oapi.CaptureConfigCategories{oapi.CaptureConfigCategoriesConsole}
 		resp, err := svc.StartCaptureSession(ctx, oapi.StartCaptureSessionRequestObject{
 			Body: &oapi.StartCaptureSessionRequest{
 				Config: &oapi.CaptureConfig{Categories: &cats},
@@ -141,7 +141,7 @@ func TestUpdateCaptureSession(t *testing.T) {
 		_, err := svc.StartCaptureSession(ctx, oapi.StartCaptureSessionRequestObject{})
 		require.NoError(t, err)
 
-		cats := []oapi.CaptureConfigCategories{oapi.Console}
+		cats := []oapi.CaptureConfigCategories{oapi.CaptureConfigCategoriesConsole}
 		resp, err := svc.UpdateCaptureSession(ctx, oapi.UpdateCaptureSessionRequestObject{
 			Body: &oapi.UpdateCaptureSessionRequest{
 				Config: &oapi.CaptureConfig{Categories: &cats},
@@ -152,7 +152,7 @@ func TestUpdateCaptureSession(t *testing.T) {
 		require.True(t, ok)
 		require.NotNil(t, r200.Config.Categories)
 		assert.Len(t, *r200.Config.Categories, 1)
-		assert.Equal(t, oapi.Console, (*r200.Config.Categories)[0])
+		assert.Equal(t, oapi.CaptureConfigCategoriesConsole, (*r200.Config.Categories)[0])
 	})
 
 	t.Run("empty body is no-op", func(t *testing.T) {
@@ -246,7 +246,8 @@ func (m *mockRecordManager) StopAll(_ context.Context) error
 // newTestService builds an ApiService with minimal dependencies for capture session tests.
 func newTestService(t *testing.T, mgr recorder.RecordManager) *ApiService {
 	t.Helper()
-	svc, err := New(mgr, newMockFactory(), newTestUpstreamManager(), scaletozero.NewNoopController(), newMockNekoClient(t), newCaptureSession(t), 0)
+	cs, es := newCaptureSession(t)
+	svc, err := New(mgr, newMockFactory(), newTestUpstreamManager(), scaletozero.NewNoopController(), newMockNekoClient(t), cs, es, 0)
 	require.NoError(t, err)
 	svc.cdpMonitor = &stubCdpMonitor{}
 	return svc

server/cmd/api/api/display_test.go

@@ -34,7 +34,8 @@ 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), newCaptureSession(t), 0)
+	cs, es := newCaptureSession(t)
+	svc, err := New(mgr, factory, newTestUpstreamManager(), scaletozero.NewNoopController(), newMockNekoClient(t), cs, es, 0)
 	require.NoError(t, err)
 	return svc
 }