perf(desktop): batch session telemetry checkpoints

desktop/app.go

@@ -968,7 +968,6 @@ func (a *App) shutdown(context.Context) {
 	// Close every shared plugin host before releasing the lifecycle barrier,
 	// even if a tab cleanup panics.
 	defer a.closeAllSharedHosts()
-
 	a.mu.RLock()
 	tabs := a.runtimeTabsLocked()
 	type shutdownItem struct {
@@ -988,6 +987,7 @@ func (a *App) shutdown(context.Context) {
 			if err := it.ctrl.SnapshotForShutdown(); err != nil {
 				slog.Warn("desktop: shutdown snapshot failed", "tab", it.tab.ID, "err", err)
 			}
+			it.tab.checkpointTelemetryForShutdown(it.ctrl.SessionPath())
 		}
 		it.ctrl.Close()
 		it.tab.releaseSessionLease()

desktop/tabs.go

@@ -1214,9 +1214,9 @@ func (t *WorkspaceTab) resetTelemetry(sessionPath string) {
 // session. When the runtime rotated to a different session underneath the tab
 // (typed /new routes through Controller.Submit and never reaches App.NewSession),
 // the previous session's totals must not bleed into the new one: swap in the
-// new session's persisted sidecar, or start from zero when none exists. The
-// sidecar is rewritten on every recorded event, so a reload never loses more
-// than the sub-second in-memory delta of an in-flight record.
+// new session's persisted sidecar, or start from zero when none exists.
+// The sink checkpoints at turn completion and by event count; the age limit is
+// evaluated on the next Usage or read_file mutation, while out-of-turn writes stay immediate.
 func (t *WorkspaceTab) syncTelemetryToSession(sessionPath string) {
 	key := sessionRuntimeKey(sessionPath)
 	if key == "" {
@@ -1536,6 +1536,10 @@ type tabEventSink struct {
 	botSink       event.Sink // optional: when set, events are also forwarded here
 	botSinkGen    uint64
 	turnInFlight  bool // stays true through the end of TurnDone fan-out
+
+	telemetryCheckpointMu sync.Mutex
+	telemetryCheckpoint   telemetryCheckpointState
+	telemetrySaveHook     func(string, tabTelemetrySnapshot) error // deterministic per-sink test override
 }
 
 type closeableEventSink interface {
@@ -1979,9 +1983,7 @@ func (s *tabEventSink) recordReadTelemetry(e event.Event) {
 		Limit:     limit,
 		Truncated: truncated,
 	})
-	if sp != "" {
-		_ = saveTelemetry(sp+".telemetry.json", tab.telemetrySnapshot())
-	}
+	s.checkpointTelemetry(tab, sp, !s.turnInFlightSnapshot())
 }
 
 func (s *tabEventSink) recordTurnStarted() {
@@ -1993,9 +1995,6 @@ func (s *tabEventSink) recordTurnStarted() {
 		tab.syncTelemetryToSession(sp)
 	}
 	tab.recordTurnStarted(time.Now().UnixMilli())
-	if sp != "" {
-		_ = saveTelemetry(sp+".telemetry.json", tab.telemetrySnapshot())
-	}
 }
 
 func (s *tabEventSink) recordTurnDone() {
@@ -2007,9 +2006,7 @@ func (s *tabEventSink) recordTurnDone() {
 		tab.syncTelemetryToSession(sp)
 	}
 	tab.recordTurnDone(time.Now().UnixMilli())
-	if sp != "" {
-		_ = saveTelemetry(sp+".telemetry.json", tab.telemetrySnapshot())
-	}
+	s.checkpointTelemetry(tab, sp, true)
 }
 
 func (s *tabEventSink) recordUsageTelemetry(e event.Event) {
@@ -2021,9 +2018,7 @@ func (s *tabEventSink) recordUsageTelemetry(e event.Event) {
 		tab.syncTelemetryToSession(sp)
 	}
 	tab.recordUsage(e)
-	if sp != "" {
-		_ = saveTelemetry(sp+".telemetry.json", tab.telemetrySnapshot())
-	}
+	s.checkpointTelemetry(tab, sp, !s.turnInFlightSnapshot())
 }
 
 func (s *tabEventSink) resetDisplayTurn() {

desktop/telemetry_checkpoint.go

@@ -0,0 +1,81 @@
+package main
+
+import (
+	"strings"
+	"time"
+)
+
+// Limits are evaluated only when the next Usage or read_file event arrives; no timer is armed.
+const (
+	tabTelemetryCheckpointEventLimit = 16
+	tabTelemetryCheckpointMaxAge     = 30 * time.Second
+)
+
+type telemetryCheckpointState struct {
+	events     int
+	dirtySince time.Time
+	retry      bool
+	draining   bool
+}
+
+func (s *tabEventSink) turnInFlightSnapshot() bool {
+	if s == nil {
+		return false
+	}
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	return s.turnInFlight
+}
+
+func (s *tabEventSink) checkpointTelemetry(tab *WorkspaceTab, sessionPath string, force bool) {
+	s.checkpointTelemetryAt(tab, sessionPath, force, time.Now())
+}
+
+func (t *WorkspaceTab) checkpointTelemetryForShutdown(sessionPath string) {
+	if t == nil || t.sink == nil {
+		return
+	}
+	t.sink.beginTelemetryDrain()
+	t.sink.checkpointTelemetry(t, sessionPath, true)
+}
+
+func (s *tabEventSink) beginTelemetryDrain() {
+	s.telemetryCheckpointMu.Lock()
+	s.telemetryCheckpoint.draining = true
+	s.telemetryCheckpointMu.Unlock()
+}
+
+// checkpointTelemetryAt synchronously saves at turn and event-count boundaries,
+// or on the next Usage or read_file mutation after the age limit. Injected time keeps
+// age tests deterministic; failed saves remain due for the next mutation.
+func (s *tabEventSink) checkpointTelemetryAt(tab *WorkspaceTab, sessionPath string, force bool, now time.Time) {
+	if s == nil || tab == nil || strings.TrimSpace(sessionPath) == "" {
+		return
+	}
+	s.telemetryCheckpointMu.Lock()
+	defer s.telemetryCheckpointMu.Unlock()
+
+	state := &s.telemetryCheckpoint
+	if !force {
+		state.events++
+		if state.dirtySince.IsZero() {
+			state.dirtySince = now
+		}
+		if !state.draining && !state.retry &&
+			state.events < tabTelemetryCheckpointEventLimit &&
+			now.Sub(state.dirtySince) < tabTelemetryCheckpointMaxAge {
+			return
+		}
+	}
+
+	save := saveTelemetry
+	if s.telemetrySaveHook != nil {
+		save = s.telemetrySaveHook
+	}
+	if err := save(sessionPath+".telemetry.json", tab.telemetrySnapshot()); err != nil {
+		state.retry = true
+		return
+	}
+	draining := state.draining
+	*state = telemetryCheckpointState{draining: draining}
+}