feat: add filtering and full-archive paging to browsers telemetry events

README.md

@@ -321,9 +321,12 @@ Per-category updates are partial — only categories you name are changed; other
   - `-o, --output json` - Output newline-delimited JSON envelopes
   - Default output: tab-separated `<time>\t[<category>]\t<type>`, e.g. `15:04:05  [network]  network_response`
 - `kernel browsers telemetry events <id>` - Read historical telemetry events (paged)
-  - `--limit <n>` - Maximum number of events per page (default 20)
+  - `--limit <n>` - Maximum number of events per page (1-100, default 20)
   - `--offset <cursor>` - Pagination cursor: pass the `X-Next-Offset` from a previous response
   - `--since <ts|dur>` / `--until <ts|dur>` - Time window (RFC-3339 timestamp or duration like `5m`). `--since` is ignored when `--offset` is set; `--until` still bounds the page
+  - `--categories <list>` - Filter by event category (`console`, `network`, `page`, `interaction`, `control`, `connection`, `system`, `screenshot`, `captcha`, `monitor`); filtered server-side
+  - `--types <list>` - Filter by event type (e.g. `network_response`, `console_error`); filtered client-side, so this walks every page in the window for complete results
+  - `--all` - Walk every page in the window instead of just the first (ignores `--offset`; no `next_offset` is returned)
   - `-o, --output json` - Output `{ "events": [...], "next_offset": "..." }` (omit `next_offset` when there is no next page)
 
 ### Browser Process Control

cmd/browsers.go

@@ -2753,10 +2753,13 @@ followed automatically by Chromium.`,
 	telemetryRoot.AddCommand(telemetryStream)
 
 	telemetryEvents := &cobra.Command{Use: "events <id>", Short: "Read historical telemetry events (paged)", Args: cobra.ExactArgs(1), RunE: runBrowsersTelemetryEvents}
-	telemetryEvents.Flags().Int64("limit", 0, "Maximum number of events per page (default 20)")
+	telemetryEvents.Flags().Int64("limit", 0, "Maximum number of events per page (1-100, default 20)")
 	telemetryEvents.Flags().Int64("offset", 0, "Pagination cursor: pass the X-Next-Offset from a previous response")
 	telemetryEvents.Flags().String("since", "", "Window start: RFC-3339 timestamp or a duration like 5m (default 5m). Ignored when --offset is set")
 	telemetryEvents.Flags().String("until", "", "Window end (exclusive): RFC-3339 timestamp or a duration like 5m")
+	telemetryEvents.Flags().StringSlice("categories", []string{}, "Filter by event category (console,network,page,interaction,control,connection,system,screenshot,captcha,monitor)")
+	telemetryEvents.Flags().StringSlice("types", []string{}, "Filter by event type (e.g. network_response,console_error); walks every page in the window")
+	telemetryEvents.Flags().Bool("all", false, "Walk every page in the window instead of just the first (ignores --offset)")
 	addJSONOutputFlag(telemetryEvents)
 	telemetryRoot.AddCommand(telemetryEvents)
 	browsersCmd.AddCommand(telemetryRoot)

cmd/browsers_telemetry.go

@@ -27,6 +27,7 @@ import (
 type BrowserTelemetryService interface {
 	StreamStreaming(ctx context.Context, id string, query kernel.BrowserTelemetryStreamParams, opts ...option.RequestOption) (stream *ssestream.Stream[kernel.BrowserTelemetryStreamResponse])
 	Events(ctx context.Context, id string, query kernel.BrowserTelemetryEventsParams, opts ...option.RequestOption) (res *pagination.OffsetPagination[kernel.BrowserTelemetryEventsResponse], err error)
+	EventsAutoPaging(ctx context.Context, id string, query kernel.BrowserTelemetryEventsParams, opts ...option.RequestOption) *pagination.OffsetPaginationAutoPager[kernel.BrowserTelemetryEventsResponse]
 }
 
 type BrowsersTelemetryStreamInput struct {
@@ -44,6 +45,9 @@ type BrowsersTelemetryEventsInput struct {
 	Offset     int64
 	Since      string
 	Until      string
+	Categories []string
+	Types      []string
+	All        bool
 	Output     string
 }
 
@@ -262,45 +266,82 @@ func (b BrowsersCmd) TelemetryEvents(ctx context.Context, in BrowsersTelemetryEv
 	if err := validateJSONOutput(in.Output); err != nil {
 		return err
 	}
+	if in.Limit != 0 && (in.Limit < 1 || in.Limit > 100) {
+		return fmt.Errorf("invalid --limit value %d: must be between 1 and 100", in.Limit)
+	}
+	for _, c := range in.Categories {
+		if !slices.Contains(streamFilterCategories, c) {
+			return fmt.Errorf("invalid --categories value %q: must be one of %s", c, strings.Join(streamFilterCategories, ", "))
+		}
+	}
 
-	br, err := b.browsers.Get(ctx, in.Identifier, kernel.BrowserGetParams{})
-	if err != nil {
-		return util.CleanedUpSdkError{Err: err}
+	// Resolve a name to a session ID when possible, but fall back to the identifier
+	// as-is: the events archive outlives the session, so Get can 404 for an ended
+	// session whose telemetry is still readable.
+	sessionID := in.Identifier
+	if br, gerr := b.browsers.Get(ctx, in.Identifier, kernel.BrowserGetParams{}); gerr == nil {
+		sessionID = br.SessionID
 	}
 
+	// A --types filter is client-side (the archive endpoint filters only by
+	// category), so it must see every page to be complete. Walk the whole window
+	// whenever --all or a --types filter is set; otherwise read a single page and
+	// surface the X-Next-Offset cursor for manual --offset paging.
+	fullScan := in.All || len(in.Types) > 0
+
 	params := kernel.BrowserTelemetryEventsParams{}
 	if in.Limit > 0 {
 		params.Limit = kernel.Opt(in.Limit)
 	}
-	if in.Offset > 0 {
+	if in.Offset > 0 && !fullScan {
 		params.Offset = kernel.Opt(in.Offset)
 	} else if in.Since != "" {
 		// Offset is an opaque cursor that encodes the window start, so --since is
-		// ignored once paging by offset; only send it for the first page.
+		// ignored once paging by offset; only send it for the first page. A full
+		// scan ignores --offset entirely and walks the window from --since.
 		params.Since = kernel.Opt(in.Since)
 	}
 	// --until still bounds the page even when paging by offset.
 	if in.Until != "" {
 		params.Until = kernel.Opt(in.Until)
 	}
-
-	var raw *http.Response
-	page, err := b.telemetry.Events(ctx, br.SessionID, params, option.WithResponseInto(&raw))
-	if err != nil {
-		return util.CleanedUpSdkError{Err: err}
+	// Send each category as a repeated query param. The SDK serializes a []string
+	// field as a single comma-joined value, but the endpoint expects the parameter
+	// repeated, so a comma-joined value matches no category.
+	opts := make([]option.RequestOption, 0, len(in.Categories)+1)
+	for _, c := range in.Categories {
+		opts = append(opts, option.WithQueryAdd("category", c))
 	}
 
 	var items []kernel.BrowserTelemetryEventsResponse
-	if page != nil {
-		items = page.Items
-	}
-
-	// Pagination: the API sets X-Has-More=true while more pages remain;
-	// X-Next-Offset is the cursor to pass as --offset for the next page. Surface
-	// it (in JSON and as the table hint) only when there is actually a next page.
 	nextOffset := ""
-	if raw != nil && strings.EqualFold(raw.Header.Get("X-Has-More"), "true") {
-		nextOffset = raw.Header.Get("X-Next-Offset")
+
+	if fullScan {
+		pager := b.telemetry.EventsAutoPaging(ctx, sessionID, params, opts...)
+		for pager.Next() {
+			it := pager.Current()
+			if shouldEmit(it.Event.Category, it.Event.Type, nil, in.Types) {
+				items = append(items, it)
+			}
+		}
+		if err := pager.Err(); err != nil {
+			return util.CleanedUpSdkError{Err: err}
+		}
+	} else {
+		var raw *http.Response
+		page, err := b.telemetry.Events(ctx, sessionID, params, append(opts, option.WithResponseInto(&raw))...)
+		if err != nil {
+			return util.CleanedUpSdkError{Err: err}
+		}
+		if page != nil {
+			items = page.Items
+		}
+		// The API sets X-Has-More=true while more pages remain; X-Next-Offset is
+		// the cursor to pass as --offset for the next page. Surface it (in JSON and
+		// as the table hint) only when there is actually a next page.
+		if raw != nil && strings.EqualFold(raw.Header.Get("X-Has-More"), "true") {
+			nextOffset = raw.Header.Get("X-Next-Offset")
+		}
 	}
 
 	if in.Output == "json" {
@@ -354,13 +395,19 @@ func runBrowsersTelemetryEvents(cmd *cobra.Command, args []string) error {
 	offset, _ := cmd.Flags().GetInt64("offset")
 	since, _ := cmd.Flags().GetString("since")
 	until, _ := cmd.Flags().GetString("until")
+	categories, _ := cmd.Flags().GetStringSlice("categories")
+	types, _ := cmd.Flags().GetStringSlice("types")
+	all, _ := cmd.Flags().GetBool("all")
 	b := BrowsersCmd{browsers: &svc, telemetry: &svc.Telemetry}
 	return b.TelemetryEvents(cmd.Context(), BrowsersTelemetryEventsInput{
 		Identifier: args[0],
 		Limit:      limit,
 		Offset:     offset,
 		Since:      since,
 		Until:      until,
+		Categories: categories,
+		Types:      types,
+		All:        all,
 		Output:     out,
 	})
 }

cmd/browsers_telemetry_test.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"context"
 	"encoding/json"
+	"fmt"
 	"io"
 	"os"
 	"testing"
@@ -34,8 +35,9 @@ func captureStdout(t *testing.T, fn func()) string {
 }
 
 type FakeBrowserTelemetryService struct {
-	StreamFunc func() *ssestream.Stream[kernel.BrowserTelemetryStreamResponse]
-	EventsFunc func(ctx context.Context, id string, query kernel.BrowserTelemetryEventsParams, opts ...option.RequestOption) (*pagination.OffsetPagination[kernel.BrowserTelemetryEventsResponse], error)
+	StreamFunc           func() *ssestream.Stream[kernel.BrowserTelemetryStreamResponse]
+	EventsFunc           func(ctx context.Context, id string, query kernel.BrowserTelemetryEventsParams, opts ...option.RequestOption) (*pagination.OffsetPagination[kernel.BrowserTelemetryEventsResponse], error)
+	EventsAutoPagingFunc func(id string, query kernel.BrowserTelemetryEventsParams, opts ...option.RequestOption) *pagination.OffsetPaginationAutoPager[kernel.BrowserTelemetryEventsResponse]
 }
 
 func (f *FakeBrowserTelemetryService) StreamStreaming(ctx context.Context, id string, query kernel.BrowserTelemetryStreamParams, opts ...option.RequestOption) *ssestream.Stream[kernel.BrowserTelemetryStreamResponse] {
@@ -52,6 +54,13 @@ func (f *FakeBrowserTelemetryService) Events(ctx context.Context, id string, que
 	return &pagination.OffsetPagination[kernel.BrowserTelemetryEventsResponse]{}, nil
 }
 
+func (f *FakeBrowserTelemetryService) EventsAutoPaging(ctx context.Context, id string, query kernel.BrowserTelemetryEventsParams, opts ...option.RequestOption) *pagination.OffsetPaginationAutoPager[kernel.BrowserTelemetryEventsResponse] {
+	if f.EventsAutoPagingFunc != nil {
+		return f.EventsAutoPagingFunc(id, query, opts...)
+	}
+	return pagination.NewOffsetPaginationAutoPager(&pagination.OffsetPagination[kernel.BrowserTelemetryEventsResponse]{}, nil)
+}
+
 func TestTelemetryStream_NilTelemetryErrors(t *testing.T) {
 	b := BrowsersCmd{browsers: &FakeBrowsersService{}}
 
@@ -425,3 +434,125 @@ func TestTelemetryEvents_OffsetIgnoresSinceKeepsUntil(t *testing.T) {
 	assert.NoError(t, err)
 	_ = buf
 }
+
+func TestTelemetryEvents_UnknownCategoryErrors(t *testing.T) {
+	b := BrowsersCmd{browsers: &FakeBrowsersService{}, telemetry: &FakeBrowserTelemetryService{}}
+
+	err := b.TelemetryEvents(context.Background(), BrowsersTelemetryEventsInput{Identifier: "br-1", Categories: []string{"netowrk"}})
+
+	assert.Error(t, err)
+	assert.Contains(t, err.Error(), "invalid --categories value")
+}
+
+func TestTelemetryEvents_InvalidLimitErrors(t *testing.T) {
+	b := BrowsersCmd{browsers: &FakeBrowsersService{}, telemetry: &FakeBrowserTelemetryService{}}
+
+	for _, lim := range []int64{-1, 101} {
+		err := b.TelemetryEvents(context.Background(), BrowsersTelemetryEventsInput{Identifier: "br-1", Limit: lim})
+		assert.Error(t, err)
+		assert.Contains(t, err.Error(), "invalid --limit value")
+	}
+}
+
+// Categories are filtered server-side, but the SDK comma-joins a []string field
+// into one value the endpoint won't match, so they go out as repeated query
+// params instead of the typed Category field.
+func TestTelemetryEvents_CategoriesSentAsRepeatedQueryParams(t *testing.T) {
+	buf := capturePtermOutput(t)
+	fakeBrowsers := &FakeBrowsersService{GetFunc: func(ctx context.Context, id string, query kernel.BrowserGetParams, opts ...option.RequestOption) (*kernel.BrowserGetResponse, error) {
+		return &kernel.BrowserGetResponse{SessionID: "sess-1"}, nil
+	}}
+	var gotQuery kernel.BrowserTelemetryEventsParams
+	var gotOpts []option.RequestOption
+	fakeTelemetry := &FakeBrowserTelemetryService{
+		EventsFunc: func(ctx context.Context, id string, query kernel.BrowserTelemetryEventsParams, opts ...option.RequestOption) (*pagination.OffsetPagination[kernel.BrowserTelemetryEventsResponse], error) {
+			gotQuery, gotOpts = query, opts
+			return &pagination.OffsetPagination[kernel.BrowserTelemetryEventsResponse]{}, nil
+		},
+	}
+	b := BrowsersCmd{browsers: fakeBrowsers, telemetry: fakeTelemetry}
+
+	err := b.TelemetryEvents(context.Background(), BrowsersTelemetryEventsInput{Identifier: "br-1", Categories: []string{"console", "network"}})
+
+	assert.NoError(t, err)
+	assert.Empty(t, gotQuery.Category, "categories must not use the comma-joined typed field")
+	// Two category query params plus the response-capture option for the cursor.
+	assert.Len(t, gotOpts, 3)
+	_ = buf
+}
+
+// The events archive outlives the session, so a 404 from Get (e.g. an ended
+// session) must not stop the read: the command falls back to the raw identifier.
+func TestTelemetryEvents_FallsBackToIdentifierWhenGetFails(t *testing.T) {
+	buf := capturePtermOutput(t)
+	fakeBrowsers := &FakeBrowsersService{GetFunc: func(ctx context.Context, id string, query kernel.BrowserGetParams, opts ...option.RequestOption) (*kernel.BrowserGetResponse, error) {
+		return nil, fmt.Errorf("not found")
+	}}
+	var gotID string
+	fakeTelemetry := &FakeBrowserTelemetryService{
+		EventsFunc: func(ctx context.Context, id string, query kernel.BrowserTelemetryEventsParams, opts ...option.RequestOption) (*pagination.OffsetPagination[kernel.BrowserTelemetryEventsResponse], error) {
+			gotID = id
+			return &pagination.OffsetPagination[kernel.BrowserTelemetryEventsResponse]{}, nil
+		},
+	}
+	b := BrowsersCmd{browsers: fakeBrowsers, telemetry: fakeTelemetry}
+
+	err := b.TelemetryEvents(context.Background(), BrowsersTelemetryEventsInput{Identifier: "ended-session-id"})
+
+	assert.NoError(t, err)
+	assert.Equal(t, "ended-session-id", gotID)
+	_ = buf
+}
+
+// A --types filter is client-side, so it must scan every page in the window to be
+// complete. Setting --types (without --all) must therefore route through the
+// auto-pager, not the single-page fetch that could drop matches on later pages.
+func TestTelemetryEvents_TypesFilterWalksAllPages(t *testing.T) {
+	buf := capturePtermOutput(t)
+	fakeBrowsers := &FakeBrowsersService{GetFunc: func(ctx context.Context, id string, query kernel.BrowserGetParams, opts ...option.RequestOption) (*kernel.BrowserGetResponse, error) {
+		return &kernel.BrowserGetResponse{SessionID: "sess-1"}, nil
+	}}
+	autoPaged := false
+	fakeTelemetry := &FakeBrowserTelemetryService{
+		EventsFunc: func(ctx context.Context, id string, query kernel.BrowserTelemetryEventsParams, opts ...option.RequestOption) (*pagination.OffsetPagination[kernel.BrowserTelemetryEventsResponse], error) {
+			t.Fatalf("single-page Events must not be called when --types is set")
+			return nil, nil
+		},
+		EventsAutoPagingFunc: func(id string, query kernel.BrowserTelemetryEventsParams, opts ...option.RequestOption) *pagination.OffsetPaginationAutoPager[kernel.BrowserTelemetryEventsResponse] {
+			autoPaged = true
+			return pagination.NewOffsetPaginationAutoPager(&pagination.OffsetPagination[kernel.BrowserTelemetryEventsResponse]{}, nil)
+		},
+	}
+	b := BrowsersCmd{browsers: fakeBrowsers, telemetry: fakeTelemetry}
+
+	err := b.TelemetryEvents(context.Background(), BrowsersTelemetryEventsInput{Identifier: "br-1", Types: []string{"network_response"}})
+
+	assert.NoError(t, err)
+	assert.True(t, autoPaged, "--types must walk every page so the client-side filter is complete")
+	_ = buf
+}
+
+// A full-window scan (--all/--types) must ignore the manual --offset cursor and
+// walk from --since; forwarding the offset would start mid-window and drop
+// earlier pages, contradicting the documented behavior.
+func TestTelemetryEvents_FullScanIgnoresOffsetUsesSince(t *testing.T) {
+	buf := capturePtermOutput(t)
+	fakeBrowsers := &FakeBrowsersService{GetFunc: func(ctx context.Context, id string, query kernel.BrowserGetParams, opts ...option.RequestOption) (*kernel.BrowserGetResponse, error) {
+		return &kernel.BrowserGetResponse{SessionID: "sess-1"}, nil
+	}}
+	var gotQuery kernel.BrowserTelemetryEventsParams
+	fakeTelemetry := &FakeBrowserTelemetryService{
+		EventsAutoPagingFunc: func(id string, query kernel.BrowserTelemetryEventsParams, opts ...option.RequestOption) *pagination.OffsetPaginationAutoPager[kernel.BrowserTelemetryEventsResponse] {
+			gotQuery = query
+			return pagination.NewOffsetPaginationAutoPager(&pagination.OffsetPagination[kernel.BrowserTelemetryEventsResponse]{}, nil)
+		},
+	}
+	b := BrowsersCmd{browsers: fakeBrowsers, telemetry: fakeTelemetry}
+
+	err := b.TelemetryEvents(context.Background(), BrowsersTelemetryEventsInput{Identifier: "br-1", All: true, Offset: 100, Since: "5m"})
+
+	assert.NoError(t, err)
+	assert.False(t, gotQuery.Offset.Valid(), "--all must not forward --offset")
+	assert.Equal(t, "5m", gotQuery.Since.Value, "--all walks the window from --since")
+	_ = buf
+}