feat: persist parquet recording across TUI restarts

2 files changed, 108 insertions, 2 deletions

diff --git a/internal/tui/tui.go b/internal/tui/tui.go
index 434e813..29d1fc8 100644
--- a/internal/tui/tui.go
+++ b/internal/tui/tui.go
@@ -8,10 +8,12 @@ import (
 	"strconv"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"ior/internal/flags"
 	"ior/internal/globalfilter"
+	"ior/internal/parquet"
 	"ior/internal/probemanager"
 	"ior/internal/statsengine"
 	common "ior/internal/tui/common"
@@ -64,6 +66,9 @@ type TraceRuntimeBindings interface {
 	SetLiveTrie(liveTrie flamegraphtui.LiveTrieSource)
 	SetProbeManager(manager ProbeManager)
 	StreamBuffer() eventstream.Source
+	Recorder() *parquet.Recorder
+	StreamSequencer() *eventstream.Sequencer
+	FilterEpoch() uint64
 }
 
 type runtimeBindingsContextKey struct{}
@@ -75,8 +80,11 @@ type runtimeBindings struct {
 	snapshotSource SnapshotSource
 	streamSource   eventstream.Source
 	streamBuffer   *eventstream.RingBuffer
+	streamSeq      *eventstream.Sequencer
+	recorder       *parquet.Recorder
 	liveTrieSource flamegraphtui.LiveTrieSource
 	probeManager   ProbeManager
+	filterEpoch    atomic.Uint64
 }
 
 type traceFilters struct {
@@ -88,6 +96,8 @@ func newRuntimeBindings() *runtimeBindings {
 	return &runtimeBindings{
 		streamSource: streamBuffer,
 		streamBuffer: streamBuffer,
+		streamSeq:    eventstream.NewSequencer(0),
+		recorder:     parquet.NewRecorder(parquet.RecorderConfig{}),
 	}
 }
 
@@ -109,6 +119,22 @@ func (r *runtimeBindings) StreamBuffer() eventstream.Source {
 	return r.streamBuffer
 }
 
+func (r *runtimeBindings) Recorder() *parquet.Recorder {
+	r.mu.RLock()
+	defer r.mu.RUnlock()
+	return r.recorder
+}
+
+func (r *runtimeBindings) StreamSequencer() *eventstream.Sequencer {
+	r.mu.RLock()
+	defer r.mu.RUnlock()
+	return r.streamSeq
+}
+
+func (r *runtimeBindings) FilterEpoch() uint64 {
+	return r.filterEpoch.Load()
+}
+
 func (r *runtimeBindings) SetLiveTrie(liveTrie flamegraphtui.LiveTrieSource) {
 	r.mu.Lock()
 	r.liveTrieSource = liveTrie
@@ -155,6 +181,10 @@ func (r *runtimeBindings) resetStreamBuffer() {
 	r.streamSource = r.streamBuffer
 }
 
+func (r *runtimeBindings) advanceFilterEpoch() uint64 {
+	return r.filterEpoch.Add(1)
+}
+
 func (r *runtimeBindings) resetDashboardSnapshotSource() *statsengine.Snapshot {
 	src := r.dashboardSnapshotSource()
 	if src == nil {
@@ -173,13 +203,18 @@ func (r *runtimeBindings) resetDashboardSnapshotSource() *statsengine.Snapshot {
 // RuntimeBindingsFromContext returns model-scoped trace bindings when the
 // context was created by the TUI.
 func RuntimeBindingsFromContext(ctx context.Context) (TraceRuntimeBindings, bool) {
-	bindings, ok := ctx.Value(runtimeBindingsContextKey{}).(*runtimeBindings)
+	bindings, ok := ctx.Value(runtimeBindingsContextKey{}).(TraceRuntimeBindings)
 	if !ok || bindings == nil {
 		return nil, false
 	}
 	return bindings, true
 }
 
+// ContextWithRuntimeBindings stores trace runtime bindings on the context.
+func ContextWithRuntimeBindings(ctx context.Context, bindings TraceRuntimeBindings) context.Context {
+	return context.WithValue(ctx, runtimeBindingsContextKey{}, bindings)
+}
+
 // ContextWithTraceFilters stores the active trace filters for the trace starter.
 func ContextWithTraceFilters(ctx context.Context, filter globalfilter.Filter) context.Context {
 	filters := traceFilters{filter: filter.Clone()}
@@ -724,7 +759,7 @@ func (m Model) cancelPickerToDashboard() (tea.Model, tea.Cmd) {
 func (m *Model) beginTraceCmd() tea.Cmd {
 	ctx, cancel := context.WithCancel(context.Background())
 	m.traceStop = cancel
-	ctx = context.WithValue(ctx, runtimeBindingsContextKey{}, m.runtime)
+	ctx = ContextWithRuntimeBindings(ctx, m.runtime)
 	ctx = ContextWithTraceFilters(ctx, m.globalFilter)
 	return startTraceCmd(m.startTrace, ctx)
 }
@@ -816,6 +851,7 @@ func (m Model) applyGlobalFilter(filter globalfilter.Filter, action string) (tea
 		return m, nil
 	}
 
+	m.runtime.advanceFilterEpoch()
 	m.stopTrace()
 	m.dashboard.PrepareForTraceRestart()
 	m.attaching = true
@@ -837,6 +873,7 @@ func (m Model) undoGlobalFilter() (tea.Model, tea.Cmd) {
 		return m, nil
 	}
 
+	m.runtime.advanceFilterEpoch()
 	m.stopTrace()
 	m.dashboard.PrepareForTraceRestart()
 	m.attaching = true
diff --git a/internal/tui/tui_test.go b/internal/tui/tui_test.go
index 361f69e..70e2b5b 100644
--- a/internal/tui/tui_test.go
+++ b/internal/tui/tui_test.go
@@ -73,6 +73,19 @@ func TestTraceFiltersContextRoundTripClonesPayload(t *testing.T) {
 	}
 }
 
+func TestRuntimeBindingsContextRoundTrip(t *testing.T) {
+	runtime := newRuntimeBindings()
+
+	ctx := ContextWithRuntimeBindings(context.Background(), runtime)
+	got, ok := RuntimeBindingsFromContext(ctx)
+	if !ok {
+		t.Fatalf("expected runtime bindings in context")
+	}
+	if got != runtime {
+		t.Fatalf("expected same runtime bindings instance from context")
+	}
+}
+
 func TestPidSelectedTransitionsToDashboardAndSetsPIDFilter(t *testing.T) {
 	flags.SetPidFilter(-1)
 	flags.SetTidFilter(99)
@@ -466,6 +479,29 @@ func TestRuntimeBindingsProvidePersistentStreamBuffer(t *testing.T) {
 	}
 }
 
+func TestRuntimeBindingsProvidePersistentRecorderAndSequencer(t *testing.T) {
+	runtime := newRuntimeBindings()
+
+	recorder := runtime.Recorder()
+	if recorder == nil {
+		t.Fatalf("expected persistent recorder")
+	}
+	if got := runtime.Recorder(); got != recorder {
+		t.Fatalf("expected recorder pointer to remain stable")
+	}
+
+	seq := runtime.StreamSequencer()
+	if seq == nil {
+		t.Fatalf("expected persistent stream sequencer")
+	}
+	if got := seq.Next(); got != 1 {
+		t.Fatalf("first persistent sequence = %d, want 1", got)
+	}
+	if got := runtime.StreamSequencer().Next(); got != 2 {
+		t.Fatalf("second persistent sequence = %d, want 2", got)
+	}
+}
+
 func TestProbeToggledMsgResetsDashboardStatsSource(t *testing.T) {
 	src := &fakeResettableDashboardSource{snap: &statsengine.Snapshot{TotalSyscalls: 99}}
 
@@ -562,6 +598,39 @@ func TestGlobalFilterApplyPreservesBufferedStreamRowsAcrossRestart(t *testing.T)
 	}
 }
 
+func TestGlobalFilterApplyAdvancesRuntimeFilterEpochAndKeepsRecorder(t *testing.T) {
+	m := NewModelWithConfig(flags.Config{PidFilter: -1, TidFilter: -1, TUIExportEnable: true}, -1, func(context.Context) error { return nil })
+	m.screen = ScreenDashboard
+	m.attaching = false
+
+	initialRecorder := m.runtime.Recorder()
+	if initialRecorder == nil {
+		t.Fatalf("expected runtime recorder")
+	}
+	if got := m.runtime.FilterEpoch(); got != 0 {
+		t.Fatalf("initial filter epoch = %d, want 0", got)
+	}
+
+	next, _ := m.Update(tea.KeyPressMsg{Code: []rune{'f'}[0], Text: string([]rune{'f'})})
+	m = next.(Model)
+	next, _ = m.Update(tea.KeyPressMsg{Code: tea.KeyEnter})
+	m = next.(Model)
+	next, _ = m.Update(tea.KeyPressMsg{Code: []rune("read")[0], Text: string([]rune("read"))})
+	m = next.(Model)
+	next, _ = m.Update(tea.KeyPressMsg{Code: tea.KeyEsc})
+	m = next.(Model)
+
+	if got := m.runtime.FilterEpoch(); got != 1 {
+		t.Fatalf("filter epoch after apply = %d, want 1", got)
+	}
+	if got := m.runtime.Recorder(); got != initialRecorder {
+		t.Fatalf("expected runtime recorder to survive filter restart")
+	}
+	if !m.attaching {
+		t.Fatalf("expected filter apply to restart tracing")
+	}
+}
+
 func TestTracingStartedUsesCurrentViewportForFlameNavigationWithoutResize(t *testing.T) {
 	trie := coreflamegraph.NewLiveTrie([]string{"comm", "path", "tracepoint"}, "count")
 	coreflamegraph.SeedTestFlameData(trie)