path: root/internal/ior_mode_test.go

package internal

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"io"
	"os"
	"path/filepath"
	"testing"
	"testing/synctest"
	"time"

	"ior/internal/event"
	"ior/internal/file"
	"ior/internal/flags"
	"ior/internal/globalfilter"
	"ior/internal/parquet"
	"ior/internal/tui"
	"ior/internal/tui/eventstream"
	flamegraphtui "ior/internal/tui/flamegraph"
	"ior/internal/types"

	parquetgo "github.com/parquet-go/parquet-go"
)

func TestShouldRunTraceMode(t *testing.T) {
	base := flags.Config{}

	if shouldRunTraceMode(base) {
		t.Fatalf("expected default mode to use TUI")
	}

	withPlain := base
	withPlain.PlainMode = true
	if !shouldRunTraceMode(withPlain) {
		t.Fatalf("expected plain mode to use trace mode")
	}

	withParquet := base
	withParquet.ParquetPath = "trace.parquet"
	if !shouldRunTraceMode(withParquet) {
		t.Fatalf("expected parquet mode to use trace mode")
	}

	withPprof := base
	withPprof.PprofEnable = true
	if shouldRunTraceMode(withPprof) {
		t.Fatalf("expected pprof flag alone to keep TUI mode")
	}

	withTestFlames := base
	withTestFlames.TestFlames = true
	if shouldRunTraceMode(withTestFlames) {
		t.Fatalf("expected --testflames to stay in TUI mode")
	}

	withTestLiveFlames := base
	withTestLiveFlames.TestLiveFlames = true
	if shouldRunTraceMode(withTestLiveFlames) {
		t.Fatalf("expected --testliveflames to stay in TUI mode")
	}
}

func TestShouldAutoStopByDuration(t *testing.T) {
	base := flags.Config{}
	if shouldAutoStopByDuration(base) {
		t.Fatalf("expected default TUI mode not to auto-stop by duration")
	}

	withPlain := base
	withPlain.PlainMode = true
	if !shouldAutoStopByDuration(withPlain) {
		t.Fatalf("expected plain mode to auto-stop by duration")
	}

	withParquet := base
	withParquet.ParquetPath = "trace.parquet"
	if !shouldAutoStopByDuration(withParquet) {
		t.Fatalf("expected parquet mode to auto-stop by duration")
	}

	withPprof := base
	withPprof.PprofEnable = true
	if shouldAutoStopByDuration(withPprof) {
		t.Fatalf("expected pprof flag alone not to auto-stop by duration")
	}

}

func TestDispatchRunUsesTraceModeWhenRequested(t *testing.T) {
	origRunTrace := runTraceFn
	origRunParquet := runParquetFn
	origRunTUI := runTUIFn
	origRunTUITestFlames := runTUITestFlamesFn
	origRunTUITestLiveFlames := runTUITestLiveFlamesFn
	defer func() {
		runTraceFn = origRunTrace
		runParquetFn = origRunParquet
		runTUIFn = origRunTUI
		runTUITestFlamesFn = origRunTUITestFlames
		runTUITestLiveFlamesFn = origRunTUITestLiveFlames
	}()

	traceCalled := false
	tuiCalled := false
	runTraceFn = func(flags.Config) error {
		traceCalled = true
		return nil
	}
	runParquetFn = func(flags.Config) error {
		t.Fatalf("runParquetFn should not be called in plain trace mode")
		return nil
	}
	runTUIFn = func(flags.Config, tui.TraceStarter) error {
		tuiCalled = true
		return nil
	}
	runTUITestFlamesFn = func(flags.Config, tui.TraceStarter) error {
		t.Fatalf("runTUITestFlamesFn should not be called in trace mode")
		return nil
	}
	runTUITestLiveFlamesFn = func(flags.Config, tui.TraceStarter) error {
		t.Fatalf("runTUITestLiveFlamesFn should not be called in trace mode")
		return nil
	}

	cfg := flags.Config{PlainMode: true}
	if err := dispatchRun(cfg); err != nil {
		t.Fatalf("dispatchRun returned error: %v", err)
	}
	if !traceCalled {
		t.Fatalf("expected runTraceFn to be called")
	}
	if tuiCalled {
		t.Fatalf("did not expect runTUIFn to be called")
	}
}

func TestDispatchRunUsesHeadlessParquetModeWhenRequested(t *testing.T) {
	origRunTrace := runTraceFn
	origRunParquet := runParquetFn
	origRunTUI := runTUIFn
	origRunTUITestFlames := runTUITestFlamesFn
	origRunTUITestLiveFlames := runTUITestLiveFlamesFn
	defer func() {
		runTraceFn = origRunTrace
		runParquetFn = origRunParquet
		runTUIFn = origRunTUI
		runTUITestFlamesFn = origRunTUITestFlames
		runTUITestLiveFlamesFn = origRunTUITestLiveFlames
	}()

	traceCalled := false
	parquetCalled := false
	tuiCalled := false
	runTraceFn = func(flags.Config) error {
		traceCalled = true
		return nil
	}
	runParquetFn = func(flags.Config) error {
		parquetCalled = true
		return nil
	}
	runTUIFn = func(flags.Config, tui.TraceStarter) error {
		tuiCalled = true
		return nil
	}
	runTUITestFlamesFn = func(flags.Config, tui.TraceStarter) error {
		t.Fatalf("runTUITestFlamesFn should not be called in parquet mode")
		return nil
	}
	runTUITestLiveFlamesFn = func(flags.Config, tui.TraceStarter) error {
		t.Fatalf("runTUITestLiveFlamesFn should not be called in parquet mode")
		return nil
	}

	cfg := flags.Config{ParquetPath: "trace.parquet"}
	if err := dispatchRun(cfg); err != nil {
		t.Fatalf("dispatchRun returned error: %v", err)
	}
	if !parquetCalled {
		t.Fatalf("expected runParquetFn to be called")
	}
	if traceCalled {
		t.Fatalf("did not expect runTraceFn to be called")
	}
	if tuiCalled {
		t.Fatalf("did not expect runTUIFn to be called")
	}
}

func TestDispatchRunUsesTUIWhenOnlyPprofEnabled(t *testing.T) {
	origRunTrace := runTraceFn
	origRunTUI := runTUIFn
	origRunTUITestFlames := runTUITestFlamesFn
	origRunTUITestLiveFlames := runTUITestLiveFlamesFn
	defer func() {
		runTraceFn = origRunTrace
		runTUIFn = origRunTUI
		runTUITestFlamesFn = origRunTUITestFlames
		runTUITestLiveFlamesFn = origRunTUITestLiveFlames
	}()

	traceCalled := false
	tuiCalled := false
	runTraceFn = func(flags.Config) error {
		traceCalled = true
		return nil
	}
	runTUIFn = func(flags.Config, tui.TraceStarter) error {
		tuiCalled = true
		return nil
	}
	runTUITestFlamesFn = func(flags.Config, tui.TraceStarter) error {
		t.Fatalf("runTUITestFlamesFn should not be called for regular TUI mode")
		return nil
	}
	runTUITestLiveFlamesFn = func(flags.Config, tui.TraceStarter) error {
		t.Fatalf("runTUITestLiveFlamesFn should not be called for regular TUI mode")
		return nil
	}

	cfg := flags.Config{PprofEnable: true}
	if err := dispatchRun(cfg); err != nil {
		t.Fatalf("dispatchRun returned error: %v", err)
	}
	if traceCalled {
		t.Fatalf("did not expect runTraceFn when only -pprof is enabled")
	}
	if !tuiCalled {
		t.Fatalf("expected runTUIFn to be called")
	}
}

func TestDispatchRunUsesTUIStarterWhenNotPlain(t *testing.T) {
	origRunTraceWithContext := runTraceWithContextFn
	origRunTUI := runTUIFn
	origRunTUITestFlames := runTUITestFlamesFn
	origRunTUITestLiveFlames := runTUITestLiveFlamesFn
	defer func() {
		runTraceWithContextFn = origRunTraceWithContext
		runTUIFn = origRunTUI
		runTUITestFlamesFn = origRunTUITestFlames
		runTUITestLiveFlamesFn = origRunTUITestLiveFlames
	}()

	traceDone := make(chan struct{}, 1)
	runTraceWithContextFn = func(_ context.Context, _ flags.Config, started chan<- struct{}, configure func(*eventLoop)) error {
		if configure != nil {
			configure(&eventLoop{})
		}
		close(started)
		traceDone <- struct{}{}
		return nil
	}

	tuiCalled := false
	runTUIFn = func(_ flags.Config, starter tui.TraceStarter) error {
		tuiCalled = true
		if starter == nil {
			t.Fatalf("expected non-nil starter")
		}
		if err := starter(context.Background()); err != nil {
			t.Fatalf("starter returned error: %v", err)
		}
		return nil
	}
	runTUITestFlamesFn = func(flags.Config, tui.TraceStarter) error {
		t.Fatalf("runTUITestFlamesFn should not be called for normal starter path")
		return nil
	}
	runTUITestLiveFlamesFn = func(flags.Config, tui.TraceStarter) error {
		t.Fatalf("runTUITestLiveFlamesFn should not be called for normal starter path")
		return nil
	}

	cfg := flags.Config{}
	if err := dispatchRun(cfg); err != nil {
		t.Fatalf("dispatchRun returned error: %v", err)
	}
	if !tuiCalled {
		t.Fatalf("expected runTUIFn to be called")
	}

	select {
	case <-traceDone:
	case <-time.After(200 * time.Millisecond):
		t.Fatalf("expected starter to launch runTraceWithContextFn")
	}
}

func TestDispatchRunUsesTestFlamesModeWhenRequested(t *testing.T) {
	origRunTrace := runTraceFn
	origRunTUI := runTUIFn
	origRunTUITestFlames := runTUITestFlamesFn
	origRunTUITestLiveFlames := runTUITestLiveFlamesFn
	defer func() {
		runTraceFn = origRunTrace
		runTUIFn = origRunTUI
		runTUITestFlamesFn = origRunTUITestFlames
		runTUITestLiveFlamesFn = origRunTUITestLiveFlames
	}()

	traceCalled := false
	regularTUICalled := false
	testFlamesCalled := false
	runTraceFn = func(flags.Config) error {
		traceCalled = true
		return nil
	}
	runTUIFn = func(flags.Config, tui.TraceStarter) error {
		regularTUICalled = true
		return nil
	}
	runTUITestFlamesFn = func(_ flags.Config, starter tui.TraceStarter) error {
		testFlamesCalled = true
		if starter == nil {
			t.Fatalf("expected non-nil starter for test flames mode")
		}
		return starter(context.Background())
	}
	runTUITestLiveFlamesFn = func(flags.Config, tui.TraceStarter) error {
		t.Fatalf("runTUITestLiveFlamesFn should not be called for --testflames")
		return nil
	}

	cfg := flags.Config{TestFlames: true}
	if err := dispatchRun(cfg); err != nil {
		t.Fatalf("dispatchRun returned error: %v", err)
	}
	if traceCalled {
		t.Fatalf("did not expect runTraceFn for test flames mode")
	}
	if regularTUICalled {
		t.Fatalf("did not expect runTUIFn for test flames mode")
	}
	if !testFlamesCalled {
		t.Fatalf("expected runTUITestFlamesFn to be called")
	}
}

func TestDispatchRunUsesTestLiveFlamesModeWhenRequested(t *testing.T) {
	origRunTrace := runTraceFn
	origRunTUI := runTUIFn
	origRunTUITestFlames := runTUITestFlamesFn
	origRunTUITestLiveFlames := runTUITestLiveFlamesFn
	defer func() {
		runTraceFn = origRunTrace
		runTUIFn = origRunTUI
		runTUITestFlamesFn = origRunTUITestFlames
		runTUITestLiveFlamesFn = origRunTUITestLiveFlames
	}()

	traceCalled := false
	regularTUICalled := false
	testLiveFlamesCalled := false
	runTraceFn = func(flags.Config) error {
		traceCalled = true
		return nil
	}
	runTUIFn = func(flags.Config, tui.TraceStarter) error {
		regularTUICalled = true
		return nil
	}
	runTUITestFlamesFn = func(flags.Config, tui.TraceStarter) error {
		t.Fatalf("runTUITestFlamesFn should not be called for --testliveflames")
		return nil
	}
	runTUITestLiveFlamesFn = func(_ flags.Config, starter tui.TraceStarter) error {
		testLiveFlamesCalled = true
		if starter == nil {
			t.Fatalf("expected non-nil starter for test live flames mode")
		}
		return starter(context.Background())
	}

	cfg := flags.Config{TestLiveFlames: true}
	if err := dispatchRun(cfg); err != nil {
		t.Fatalf("dispatchRun returned error: %v", err)
	}
	if traceCalled {
		t.Fatalf("did not expect runTraceFn for test live flames mode")
	}
	if regularTUICalled {
		t.Fatalf("did not expect runTUIFn for test live flames mode")
	}
	if !testLiveFlamesCalled {
		t.Fatalf("expected runTUITestLiveFlamesFn to be called")
	}
}

func TestValidateRunConfigRejectsTestFlamesWithTraceFlags(t *testing.T) {
	cfg := flags.Config{TestFlames: true, PlainMode: true}
	err := validateRunConfig(cfg)
	if err == nil {
		t.Fatalf("expected error for --testflames with trace-mode flags")
	}
	if err.Error() != "--testflames cannot be combined with -plain" {
		t.Fatalf("unexpected error: %v", err)
	}
}

func TestValidateRunConfigRejectsTestLiveFlamesWithTraceFlags(t *testing.T) {
	cfg := flags.Config{TestLiveFlames: true, PlainMode: true}
	err := validateRunConfig(cfg)
	if err == nil {
		t.Fatalf("expected error for --testliveflames with trace-mode flags")
	}
	if err.Error() != "--testliveflames cannot be combined with -plain" {
		t.Fatalf("unexpected error: %v", err)
	}
}

func TestValidateRunConfigRejectsBothTestModes(t *testing.T) {
	cfg := flags.Config{TestFlames: true, TestLiveFlames: true}
	err := validateRunConfig(cfg)
	if err == nil {
		t.Fatalf("expected error when both test flame modes are enabled")
	}
	if err.Error() != "--testflames and --testliveflames are mutually exclusive" {
		t.Fatalf("unexpected error: %v", err)
	}
}

func TestValidateRunConfigRejectsParquetWithPlain(t *testing.T) {
	cfg := flags.Config{ParquetPath: "trace.parquet", PlainMode: true}
	err := validateRunConfig(cfg)
	if err == nil {
		t.Fatalf("expected error for -parquet with -plain")
	}
	if err.Error() != "-parquet and -plain are mutually exclusive" {
		t.Fatalf("unexpected error: %v", err)
	}
}

func TestValidateRunConfigRejectsParquetWithContentFilters(t *testing.T) {
	cfg := flags.Config{
		ParquetPath: "trace.parquet",
		CommFilter:  "nginx",
	}
	err := validateRunConfig(cfg)
	if err == nil {
		t.Fatalf("expected error for -parquet with content filters")
	}
	if err.Error() != "-parquet cannot be combined with content filters (-comm, -path, -pid, -tid)" {
		t.Fatalf("unexpected error: %v", err)
	}
}

func TestValidateRunConfigRejectsParquetWithGlobalFilter(t *testing.T) {
	cfg := flags.Config{
		ParquetPath: "trace.parquet",
		GlobalFilter: globalfilter.Filter{
			Syscall: &globalfilter.StringFilter{Pattern: "read"},
		},
	}
	err := validateRunConfig(cfg)
	if err == nil {
		t.Fatalf("expected error for -parquet with global filter")
	}
	if err.Error() != "-parquet cannot be combined with content filters (-comm, -path, -pid, -tid)" {
		t.Fatalf("unexpected error: %v", err)
	}
}

func TestBuildTestFlamesRuntimeSeedsLiveTrie(t *testing.T) {
	cfg := flags.NewFlags()
	_, streamBuf, liveTrie := buildTestFlamesRuntime(cfg)
	if streamBuf == nil {
		t.Fatalf("expected stream buffer in test flames runtime")
	}
	if liveTrie == nil {
		t.Fatalf("expected live trie in test flames runtime")
	}
	if liveTrie.Version() == 0 {
		t.Fatalf("expected seeded live trie version to be non-zero")
	}

	payload, _ := liveTrie.SnapshotJSON()
	var snap map[string]any
	if err := json.Unmarshal(payload, &snap); err != nil {
		t.Fatalf("decode snapshot: %v", err)
	}
	total, ok := snap["t"].(float64)
	if !ok || total <= 0 {
		t.Fatalf("expected seeded snapshot total > 0, got %v", snap["t"])
	}
}

func TestBuildTestLiveFlamesRuntimeContinuouslyUpdatesLiveTrie(t *testing.T) {
	synctest.Test(t, func(t *testing.T) {
		cfg := flags.NewFlags()
		cfg.LiveInterval = 15 * time.Millisecond

		ctx, cancel := context.WithCancel(context.Background())
		defer cancel()

		_, streamBuf, liveTrie := buildTestLiveFlamesRuntime(ctx, cfg)
		if streamBuf == nil {
			t.Fatalf("expected stream buffer in test live flames runtime")
		}
		if liveTrie == nil {
			t.Fatalf("expected live trie in test live flames runtime")
		}

		initialVersion := liveTrie.Version()
		if initialVersion == 0 {
			t.Fatalf("expected seeded live trie version to be non-zero")
		}
		initialSnapshot, _ := liveTrie.SnapshotJSON()

		time.Sleep(cfg.LiveInterval + time.Nanosecond)
		synctest.Wait()

		if liveTrie.Version() <= initialVersion {
			t.Fatalf("expected live trie version to advance beyond %d", initialVersion)
		}
		currentSnapshot, _ := liveTrie.SnapshotJSON()
		if bytes.Equal(initialSnapshot, currentSnapshot) {
			t.Fatalf("expected test live flames snapshot shape to change over time")
		}
	})
}

func TestRunTraceWithContextRequiresRoot(t *testing.T) {
	origGetEUID := getEUID
	defer func() { getEUID = origGetEUID }()

	getEUID = func() int { return 1000 }
	err := runTraceWithContext(context.Background(), flags.NewFlags(), nil, nil)
	if !errors.Is(err, errRootPrivilegesRequired) {
		t.Fatalf("expected root-required error, got %v", err)
	}
}

func TestTuiTraceStarterFromRunTracePropagatesError(t *testing.T) {
	starter := tuiTraceStarterFromRunTrace(
		flags.NewFlags(),
		func(context.Context, flags.Config, chan<- struct{}, func(*eventLoop)) error {
			return errors.New("startup failed")
		},
	)

	err := starter(context.Background())
	if err == nil || err.Error() != "startup failed" {
		t.Fatalf("expected startup error, got %v", err)
	}
}

func TestTuiTraceStarterFromRunTraceUsesContextFilters(t *testing.T) {
	base := flags.NewFlags()
	base.PidFilter = 11
	base.TidFilter = 12

	var gotCfg flags.Config
	starter := tuiTraceStarterFromRunTrace(
		base,
		func(_ context.Context, cfg flags.Config, started chan<- struct{}, _ func(*eventLoop)) error {
			gotCfg = cfg
			close(started)
			return nil
		},
	)

	ctx := tui.ContextWithTraceFilters(context.Background(), globalfilter.Filter{
		PID:     &globalfilter.NumericFilter{Op: globalfilter.OpEq, Value: 2222},
		TID:     &globalfilter.NumericFilter{Op: globalfilter.OpEq, Value: 3333},
		Comm:    &globalfilter.StringFilter{Pattern: "nginx"},
		File:    &globalfilter.StringFilter{Pattern: "/var/log"},
		Syscall: &globalfilter.StringFilter{Pattern: "read"},
		FD:      &globalfilter.NumericFilter{Op: globalfilter.OpEq, Value: 7},
	})
	if err := starter(ctx); err != nil {
		t.Fatalf("starter returned error: %v", err)
	}
	if gotCfg.PidFilter != 2222 {
		t.Fatalf("expected pid filter from context, got %d", gotCfg.PidFilter)
	}
	if gotCfg.TidFilter != 3333 {
		t.Fatalf("expected tid filter from context, got %d", gotCfg.TidFilter)
	}
	if gotCfg.CommFilter != "" {
		t.Fatalf("expected legacy comm filter to remain unused, got %q", gotCfg.CommFilter)
	}
	if gotCfg.PathFilter != "" {
		t.Fatalf("expected legacy path filter to remain unused, got %q", gotCfg.PathFilter)
	}
	if gotCfg.GlobalFilter.Comm == nil || gotCfg.GlobalFilter.Comm.Pattern != "nginx" {
		t.Fatalf("expected comm preserved in global filter payload, got %+v", gotCfg.GlobalFilter.Comm)
	}
	if gotCfg.GlobalFilter.File == nil || gotCfg.GlobalFilter.File.Pattern != "/var/log" {
		t.Fatalf("expected file preserved in global filter payload, got %+v", gotCfg.GlobalFilter.File)
	}
	if gotCfg.GlobalFilter.Syscall == nil || gotCfg.GlobalFilter.Syscall.Pattern != "read" {
		t.Fatalf("expected syscall preserved in global filter payload, got %+v", gotCfg.GlobalFilter.Syscall)
	}
	if gotCfg.GlobalFilter.FD == nil || gotCfg.GlobalFilter.FD.Value != 7 {
		t.Fatalf("expected fd preserved in global filter payload, got %+v", gotCfg.GlobalFilter.FD)
	}
}

func TestShouldIngestTracePairAppliesFullGlobalFilter(t *testing.T) {
	pair := &event.Pair{
		EnterEv:        &types.RetEvent{TraceId: types.SYS_ENTER_READ, Pid: 1234, Tid: 1235},
		ExitEv:         &types.RetEvent{TraceId: types.SYS_EXIT_READ, Pid: 1234, Tid: 1235, Ret: -1},
		Comm:           "nginx",
		File:           file.NewFd(7, "/var/log/access.log", 0),
		Duration:       1_500_000,
		DurationToPrev: 12_000,
		Bytes:          4_096,
	}

	filter := globalfilter.Filter{
		Syscall:    &globalfilter.StringFilter{Pattern: "rea"},
		Comm:       &globalfilter.StringFilter{Pattern: "ngi"},
		File:       &globalfilter.StringFilter{Pattern: "access"},
		PID:        &globalfilter.NumericFilter{Op: globalfilter.OpEq, Value: 1234},
		TID:        &globalfilter.NumericFilter{Op: globalfilter.OpEq, Value: 1235},
		FD:         &globalfilter.NumericFilter{Op: globalfilter.OpEq, Value: 7},
		LatencyNs:  &globalfilter.NumericFilter{Op: globalfilter.OpGt, Value: 1_000_000},
		GapNs:      &globalfilter.NumericFilter{Op: globalfilter.OpLte, Value: 12_000},
		Bytes:      &globalfilter.NumericFilter{Op: globalfilter.OpLt, Value: 8_192},
		RetVal:     &globalfilter.NumericFilter{Op: globalfilter.OpEq, Value: -1},
		ErrorsOnly: true,
	}

	if !shouldIngestTracePair(filter, pair) {
		t.Fatalf("expected full filter to accept matching pair")
	}
	if shouldIngestTracePair(globalfilter.Filter{Syscall: &globalfilter.StringFilter{Pattern: "write"}}, pair) {
		t.Fatalf("expected syscall mismatch to reject pair")
	}
	if shouldIngestTracePair(globalfilter.Filter{FD: &globalfilter.NumericFilter{Op: globalfilter.OpEq, Value: 99}}, pair) {
		t.Fatalf("expected fd mismatch to reject pair")
	}
}

func TestProfilingFilesForMode(t *testing.T) {
	cpu, mem, execTrace, duration := profilingFilesForMode(false)
	if cpu != "ior.cpuprofile" || mem != "ior.memprofile" {
		t.Fatalf("unexpected trace-mode profiling file names: cpu=%q mem=%q", cpu, mem)
	}
	if execTrace != "" || duration != 0 {
		t.Fatalf("expected trace-mode execution tracing to be disabled, got trace=%q duration=%s", execTrace, duration)
	}

	cpu, mem, execTrace, duration = profilingFilesForMode(true)
	if cpu != "ior-tui-cpu.prof" || mem != "ior-tui-mem.prof" || execTrace != "ior-tui-trace.out" {
		t.Fatalf("unexpected TUI profiling file names: cpu=%q mem=%q trace=%q", cpu, mem, execTrace)
	}
	if duration != 10*time.Second {
		t.Fatalf("expected 10s TUI execution trace duration, got %s", duration)
	}
}

func TestTuiTraceStarterFromRunTraceRespectsCancel(t *testing.T) {
	starter := tuiTraceStarterFromRunTrace(
		flags.NewFlags(),
		func(ctx context.Context, _ flags.Config, _ chan<- struct{}, _ func(*eventLoop)) error {
			<-ctx.Done()
			return ctx.Err()
		},
	)

	ctx, cancel := context.WithCancel(context.Background())
	cancel()

	err := starter(ctx)
	if !errors.Is(err, context.Canceled) {
		t.Fatalf("expected context canceled, got %v", err)
	}
}

func TestHeadlessParquetTraceConfigClearsContentFilters(t *testing.T) {
	cfg := flags.Config{
		ParquetPath: "trace.parquet",
		PidFilter:   1234,
		TidFilter:   5678,
		CommFilter:  "nginx",
		PathFilter:  "/var/log",
		GlobalFilter: globalfilter.Filter{
			Syscall: &globalfilter.StringFilter{Pattern: "read"},
		},
	}

	got := headlessParquetTraceConfig(cfg)
	if got.PidFilter != -1 || got.TidFilter != -1 {
		t.Fatalf("pid/tid filters = %d/%d, want -1/-1", got.PidFilter, got.TidFilter)
	}
	if got.CommFilter != "" || got.PathFilter != "" {
		t.Fatalf("comm/path filters = %q/%q, want empty", got.CommFilter, got.PathFilter)
	}
	if got.GlobalFilter.IsActive() {
		t.Fatalf("expected sanitized global filter to be empty, got %+v", got.GlobalFilter)
	}
}

func TestHeadlessParquetSinkRecordsRows(t *testing.T) {
	recorder := parquet.NewRecorder(parquet.RecorderConfig{
		BatchSize:     1,
		FlushInterval: time.Hour,
	})
	path := filepath.Join(t.TempDir(), "headless.parquet")
	if err := recorder.Start(path, parquet.StartOptions{
		Metadata: parquet.FileMetadata{Mode: "headless"},
	}); err != nil {
		t.Fatalf("recorder.Start() error = %v", err)
	}

	_, cancel := context.WithCancel(context.Background())
	defer cancel()

	sink := newHeadlessParquetSink(recorder, cancel)
	el := &eventLoop{}
	sink.configure(el)

	el.printCb(testTracePair(1, "keep"))
	el.printCb(testTracePair(2, "keep"))

	if err := recorder.Stop(); err != nil {
		t.Fatalf("recorder.Stop() error = %v", err)
	}
	if err := sink.err(); err != nil {
		t.Fatalf("sink.err() = %v, want nil", err)
	}

	rows := readRecordedParquet(t, path)
	if len(rows) != 2 {
		t.Fatalf("recorded rows = %d, want 2", len(rows))
	}
	if rows[0].Seq != 1 || rows[1].Seq != 2 {
		t.Fatalf("recorded seq = %d,%d, want 1,2", rows[0].Seq, rows[1].Seq)
	}
	if rows[0].FilterEpoch != 0 || rows[1].FilterEpoch != 0 {
		t.Fatalf("recorded filter epochs = %d,%d, want 0,0", rows[0].FilterEpoch, rows[1].FilterEpoch)
	}
	if rows[0].Comm != "keep" || rows[1].Syscall != "openat" {
		t.Fatalf("unexpected recorded rows: %+v %+v", rows[0], rows[1])
	}
}

func TestTuiTraceStarterFromRunTracePersistsRecorderAcrossRestarts(t *testing.T) {
	recorder := parquet.NewRecorder(parquet.RecorderConfig{
		BatchSize:     1,
		FlushInterval: time.Hour,
	})
	if err := recorder.Start(filepath.Join(t.TempDir(), "trace"), parquet.StartOptions{
		Metadata: parquet.FileMetadata{Mode: "tui"},
	}); err != nil {
		t.Fatalf("recorder.Start() error = %v", err)
	}

	bindings := &traceRuntimeBindingsStub{
		streamBuffer: eventstream.NewRingBuffer(),
		streamSeq:    eventstream.NewSequencer(0),
		recorder:     recorder,
	}
	base := flags.NewFlags()
	base.GlobalFilter = globalfilter.Filter{Comm: &globalfilter.StringFilter{Pattern: "keep"}}

	runs := [][]*event.Pair{
		{
			testTracePair(1, "keep"),
			testTracePair(99, "drop"),
		},
		{
			testTracePair(2, "keep"),
		},
	}
	runIndex := 0
	starter := tuiTraceStarterFromRunTrace(
		base,
		func(_ context.Context, _ flags.Config, started chan<- struct{}, configure func(*eventLoop)) error {
			el := &eventLoop{}
			configure(el)
			for _, pair := range runs[runIndex] {
				el.printCb(pair)
			}
			runIndex++
			close(started)
			return nil
		},
	)

	ctx := tui.ContextWithRuntimeBindings(context.Background(), bindings)
	if err := starter(ctx); err != nil {
		t.Fatalf("first starter() error = %v", err)
	}
	waitForStreamRows(t, bindings.streamBuffer, 1)

	bindings.filterEpoch = 1
	if err := starter(ctx); err != nil {
		t.Fatalf("second starter() error = %v", err)
	}
	waitForStreamRows(t, bindings.streamBuffer, 2)

	if err := recorder.Stop(); err != nil {
		t.Fatalf("recorder.Stop() error = %v", err)
	}

	status := recorder.Status()
	if status.LastError != nil {
		t.Fatalf("recorder status error = %v, want nil", status.LastError)
	}

	got := readRecordedParquet(t, status.Path)
	if len(got) != 2 {
		t.Fatalf("recorded rows = %d, want 2", len(got))
	}
	if got[0].Seq != 1 || got[1].Seq != 2 {
		t.Fatalf("recorded seq = %d,%d, want 1,2", got[0].Seq, got[1].Seq)
	}
	if got[0].FilterEpoch != 0 || got[1].FilterEpoch != 1 {
		t.Fatalf("recorded filter epochs = %d,%d, want 0,1", got[0].FilterEpoch, got[1].FilterEpoch)
	}
	if snapshot := bindings.streamBuffer.Snapshot(); len(snapshot) != 2 {
		t.Fatalf("stream buffer rows = %d, want 2", len(snapshot))
	}
}

type traceRuntimeBindingsStub struct {
	streamBuffer *eventstream.RingBuffer
	streamSource eventstream.Source
	streamSeq    *eventstream.Sequencer
	recorder     *parquet.Recorder
	filterEpoch  uint64
}

func (b *traceRuntimeBindingsStub) SetDashboardSnapshotSource(tui.SnapshotSource) {}

func (b *traceRuntimeBindingsStub) SetEventStreamSource(source eventstream.Source) {
	b.streamSource = source
}

func (b *traceRuntimeBindingsStub) SetLiveTrie(flamegraphtui.LiveTrieSource) {}

func (b *traceRuntimeBindingsStub) SetProbeManager(tui.ProbeManager) {}

func (b *traceRuntimeBindingsStub) StreamBuffer() eventstream.Source {
	return b.streamBuffer
}

func (b *traceRuntimeBindingsStub) Recorder() *parquet.Recorder {
	return b.recorder
}

func (b *traceRuntimeBindingsStub) StreamSequencer() *eventstream.Sequencer {
	return b.streamSeq
}

func (b *traceRuntimeBindingsStub) FilterEpoch() uint64 {
	return b.filterEpoch
}

func testTracePair(seq uint64, comm string) *event.Pair {
	enter := &types.OpenEvent{TraceId: types.SYS_ENTER_OPENAT, Time: seq * 10, Pid: 42, Tid: 84}
	exit := &types.RetEvent{TraceId: types.SYS_EXIT_OPENAT, Time: seq*10 + 1, Ret: int64(seq), Pid: 42, Tid: 84}
	pair := event.NewPair(enter)
	pair.ExitEv = exit
	pair.File = file.NewFd(int32(seq), "/tmp/test", 0)
	pair.Comm = comm
	pair.Duration = seq
	pair.DurationToPrev = seq + 1
	pair.Bytes = seq + 2
	return pair
}

func waitForStreamRows(t *testing.T, buffer *eventstream.RingBuffer, want int) {
	t.Helper()
	deadline := time.Now().Add(2 * time.Second)
	for time.Now().Before(deadline) {
		if buffer.Len() == want {
			return
		}
		time.Sleep(time.Millisecond)
	}
	t.Fatalf("stream buffer len = %d, want %d", buffer.Len(), want)
}

func readRecordedParquet(t *testing.T, path string) []parquet.Record {
	t.Helper()

	f, err := os.Open(path)
	if err != nil {
		t.Fatalf("open parquet %q: %v", path, err)
	}
	defer f.Close()

	reader := parquetgo.NewGenericReader[parquet.Record](f)
	defer reader.Close()

	var rows []parquet.Record
	buf := make([]parquet.Record, 4)
	for {
		n, err := reader.Read(buf)
		if n > 0 {
			rows = append(rows, buf[:n]...)
		}
		if err == nil {
			continue
		}
		if err == io.EOF {
			return rows
		}
		t.Fatalf("read parquet rows: %v", err)
	}
}