Worktree traces prometheus scraper

.dockerignore

@@ -0,0 +1,13 @@
+.DS_Store
+.claude/
+.cyclonedds/
+.draco/
+.zenoh-c/
+__pycache__/
+bin/
+internal/ddsgen/
+myenv/
+recordings/
+.env
+docker-compose.override.yml
+.git

README.md

@@ -10,6 +10,7 @@ A Go application that synchronously records multi-modal sensor data:
 - **Depth** frames from a CycloneDDS topic (RVL-compressed, lossless)
 - **Odometry** messages from a CycloneDDS topic
 - **Lowstate** (IMU/joints/battery/etc.) from a CycloneDDS topic
+- **LLM traces** (prompt/response records) polled from a co-located OM1 process's Prometheus trace-export endpoint
 
 All streams are timestamped and organized into session directories for easy alignment and analysis.
 
@@ -87,6 +88,33 @@ The profile has it enabled because this fleet's G1 has a D435i fitted
 (measured 14.9 Hz, 480x270 `16UC1`, RVL-compressed to ~34% of raw, lossless).
 On a G1 without one, set `ENABLE_DEPTH=false`.
 
+#### DDS readers reconnect on their own after a sensor restart
+
+`lidar`, `pointcloud`, `depth`, `odom`, and `lowstate` each hold one
+long-lived DDS reader for the life of the process -- a session rotation only
+swaps their output files (see `persistentStreams` in `cmd/main/main.go`), it
+never resubscribes.
+
+Plain DDS discovery does not always recover cleanly if the underlying sensor
+or its driver process restarts (a physical power-cycle, say): a writer
+announces itself with a burst of discovery packets right at startup and then
+falls back to a much slower steady-state interval, so a reader that's been
+sitting idle for a while can simply miss that burst and never learn the new
+writer exists -- nothing retries that specific handshake afterward. A reader
+created fresh (a full process restart, or a brand-new subscriber from a tool
+like `ros2 topic hz`) sends its own discovery burst immediately and tends to
+match right away, which is what makes this failure mode easy to miss until a
+stream has been silently empty for hours.
+
+Each of the five streams above is registered with
+`heartbeat.Monitor.RegisterRecoverable` instead of plain `Register`: once a
+stream is found broken (the same "recorder NOT WORKING" condition that gets
+logged), the monitor calls that stream's `Reconnect()` once per check
+interval until it recovers. `Reconnect()` tears down and recreates just that
+stream's DDS participant/reader -- forcing a fresh discovery burst -- without
+touching output files, so no manual restart should be needed after a sensor
+bounces.
+
 #### The G1's down camera is the RealSense RGB view
 
 There is no third USB camera on a G1 -- only `OM1FRONTCAM` and `OM1REARCAM`.
@@ -139,6 +167,9 @@ Any of the following override the selected profile / defaults:
 - `ODOM_DDS_TOPIC` - DDS topic for odometry data (default: `rt/odom`)
 - `LOWSTATE_DDS_DOMAIN` - CycloneDDS domain ID for lowstate (default: `0`)
 - `LOWSTATE_DDS_TOPIC` - DDS topic for lowstate data (default: `rt/lowstate`)
+- `TRACES_ENABLED` - Poll a co-located OM1 process's Prometheus trace-export endpoint and record its LLM traces (default: `false` — most deployments have no OM1 process to poll; see "LLM trace export" below)
+- `TRACES_URL` - OM1's metrics endpoint (default: `http://localhost:9090/metrics`)
+- `TRACES_POLL_INTERVAL` - How often to poll `TRACES_URL` (default: `30s`)
 - `RECORDINGS_DIR` - Base directory for recordings (default: `recordings`)
 - `SESSION_ROTATE_INTERVAL` - Close the current session and open a fresh one on this cadence, so each segment can be uploaded without waiting for the whole run to end (default: `5m`; `0` disables rotation — one session for the life of the process, as before this feature existed)
 - `UPLOAD_ENABLED` - Upload each rotated-away session to the openmind-api (default: `true`; actual uploading additionally requires `OPENMIND_API_URL` and `OPENMIND_API_KEY` — see "Cloud upload" below)
@@ -248,7 +279,8 @@ recordings/
         ├── depth_frames.bin           # RVL-compressed depth frames (lossless)
         ├── depth_timestamps.csv       # unix_ns,seq,byte_offset,byte_length,method,width,height,encoding,mono_ns
         ├── odom_frames.bin            # Raw odometry messages
-        └── odom_timestamps.csv        # Timestamps: unix_ns,seq,byte_offset,mono_ns
+        ├── odom_timestamps.csv        # Timestamps: unix_ns,seq,byte_offset,mono_ns
+        └── traces.jsonl                # LLM trace records polled from OM1 (if TRACES_ENABLED set)
 ```
 
 Every timestamps CSV carries a **`mono_ns`** column alongside its wall-clock
@@ -474,6 +506,33 @@ its own copy of it (a snapshot taken when that segment closes), so each
 segment's directory stays self-contained for `align_recording.py` /
 `fix_session_time.py` without needing the boot session alongside it.
 
+## LLM trace export
+
+Set `TRACES_ENABLED=true` to poll a co-located [OM1](https://github.com/OpenMind/OM1)
+process's Prometheus trace-export endpoint (`use_tracer.prometheus_export.enabled`
+in OM1's config) and record each LLM interaction trace into the current
+session's `traces.jsonl` -- picked up by the normal upload pipeline like
+every other stream, with no extra bucket-path configuration needed.
+
+This is a persistent stream: it keeps one poll loop and one dedup cursor
+(the newest trace timestamp already written) running for the life of the
+process, only swapping which session's `traces.jsonl` it writes to on
+rotation -- so OM1 re-serving its whole buffer on every poll never produces
+duplicate lines, even across a session rotation. The cursor is also
+persisted to `RECORDINGS_DIR/.traces_cursor`, so a full process restart
+resumes from it too, instead of re-ingesting OM1's whole buffered backlog
+into whatever session happens to be open at that moment.
+
+OM1's exporter only buffers its most recent 200 trace records; poll more
+often than that window fills at your deployment's trace rate
+(`TRACES_POLL_INTERVAL`, default `30s`) or older records will be evicted on
+the OM1 side before this recorder ever sees them.
+
+`TRACES_URL` points at OM1's main `/metrics` endpoint -- `om1_trace_info`
+is broadcast there alongside every other metric, so this poller (like OM1's
+own fleet-wide Prometheus, if one scrapes the same endpoint) will also see
+full prompt/response text as unbounded label values on that metric.
+
 ## Scheduling
 
 `SCHEDULE_FILE` points at an optional YAML file describing a daily

cmd/main/main.go

@@ -25,13 +25,21 @@ import (
 	"om1-telemetry/internal/retention"
 	"om1-telemetry/internal/schedule"
 	"om1-telemetry/internal/session"
+	"om1-telemetry/internal/traces"
 	"om1-telemetry/internal/upload"
 	"om1-telemetry/internal/video"
 )
 
 // scheduleTickInterval is how often the schedule reconciler checks for a state change.
 const scheduleTickInterval = 30 * time.Second
 
+// stuckStreamThreshold is consecutive 30s heartbeat failures (~5 minutes)
+// before a recoverable stream escalates from reconnecting to a full restart.
+const stuckStreamThreshold = 10
+
+// shutdownUploadTimeout bounds the synchronous upload of the live session at shutdown.
+const shutdownUploadTimeout = 10 * time.Minute
+
 func main() {
 	// Sampled before anything else, so the session's monotonic anchor covers
 	// the whole run.
@@ -63,7 +71,24 @@ func main() {
 	}
 
 	mon := heartbeat.NewMonitor(30 * time.Second)
-	registerHeartbeats(mon, cfg, videoHeartbeatNames)
+
+	restartSignal := make(chan string, 1)
+	mon.SetStuckThreshold(stuckStreamThreshold, func(stream string) {
+		select {
+		case restartSignal <- stream:
+		default:
+		}
+	})
+
+	var rs *recorderSet
+	streamsFn := func() *persistentStreams {
+		if rs == nil {
+			return nil
+		}
+		return &rs.persistent
+	}
+
+	registerHeartbeats(mon, cfg, videoHeartbeatNames, streamsFn)
 
 	hbCtx, hbCancel := context.WithCancel(context.Background())
 	go mon.Run(hbCtx)
@@ -118,9 +143,9 @@ func main() {
 
 	// Uploading (never cap enforcement) is gated on ctl.Uploading -- see retention.RunSweeps.
 	sweepCtx, sweepCancel := context.WithCancel(context.Background())
-	go retention.RunSweeps(sweepCtx, uploader, recordingsDir, bootTimebasePath, currentDirFn, cfg.Retention, ctl)
+	go retention.RunSweeps(sweepCtx, uploader, recordingsDir, bootTimebasePath, currentDirFn, cfg.Retention, ctl, cfg.Upload.DeleteAfterUpload)
 
-	rs := startRecorders(cfg, mon, recordingsDir, sess, videoHeartbeatNames)
+	rs = startRecorders(cfg, mon, recordingsDir, sess, videoHeartbeatNames)
 
 	videoURLs := make([]string, 0, len(cfg.Video))
 	for _, vc := range cfg.Video {
@@ -191,7 +216,10 @@ func main() {
 		retention.SnapshotTimebase(bootTimebasePath, finished.RealDir())
 		ctl.SetRecording(false)
 		slog.Info("recording paused by schedule", "session", finished.RealDir())
-		retention.UploadFinishedSessionAsync(&uploadWG, uploader, ctl, recordingsDir, bootTimebasePath, finished, uploadDelete)
+		persistent := rs.persistent
+		finishedStart := time.Unix(0, finished.StartUnixNs())
+		retention.UploadFinishedSessionAsync(&uploadWG, uploader, ctl, recordingsDir, bootTimebasePath, finished, uploadDelete,
+			func(ctx context.Context) { persistent.awaitSegments(ctx, finishedStart) })
 	}
 
 	// resumeRecording is the schedule's open-half: same as a rotation's open-half.
@@ -208,7 +236,7 @@ func main() {
 		currentMu.Unlock()
 
 		sess, cfg = next, nextCfg
-		registerHeartbeats(mon, cfg, videoHeartbeatNames)
+		registerHeartbeats(mon, cfg, videoHeartbeatNames, streamsFn)
 		rs = startRecorders(cfg, mon, recordingsDir, sess, videoHeartbeatNames)
 		ctl.SetRecording(true)
 		slog.Info("recording resumed by schedule", "session", sess.RealDir())
@@ -242,12 +270,19 @@ func main() {
 		reconcileSchedule()
 	}
 
+	var restartRequested bool
+
 loop:
 	for {
 		select {
 		case <-shutdownSignal:
 			break loop
 
+		case stream := <-restartSignal:
+			slog.Error("restarting process to recover a stuck stream", "stream", stream)
+			restartRequested = true
+			break loop
+
 		case <-scheduleC:
 			reconcileSchedule()
 
@@ -276,7 +311,7 @@ loop:
 			currentMu.Unlock()
 
 			sess, cfg = next, nextCfg
-			registerHeartbeats(mon, cfg, videoHeartbeatNames)
+			registerHeartbeats(mon, cfg, videoHeartbeatNames, streamsFn)
 
 			persistent := rs.persistent
 			rs = startEphemeral(cfg, mon)
@@ -285,7 +320,9 @@ loop:
 
 			slog.Info("session rotated", "session", sess.RealDir())
 
-			retention.UploadFinishedSessionAsync(&uploadWG, uploader, ctl, recordingsDir, bootTimebasePath, finished, uploadDelete)
+			finishedStart := time.Unix(0, finished.StartUnixNs())
+			retention.UploadFinishedSessionAsync(&uploadWG, uploader, ctl, recordingsDir, bootTimebasePath, finished, uploadDelete,
+				func(ctx context.Context) { persistent.awaitSegments(ctx, finishedStart) })
 		}
 	}
 
@@ -302,7 +339,11 @@ loop:
 
 		if uploader != nil && ctl.Uploading() {
 			opts := retention.UploadOptions(bootTimebasePath, sess.RealDir())
-			retention.UploadSession(ctl, uploader, sess.RealDir(), retention.APISessionDir(recordingsDir, sess.RealDir()), time.Unix(0, sess.StartUnixNs()), uploadDelete, opts)
+			// No awaitReady needed: rs.Stop() above already drains ffmpeg's
+			// final segment (and its relocation) before returning.
+			ctx, cancel := context.WithTimeout(context.Background(), shutdownUploadTimeout)
+			retention.UploadSession(ctx, ctl, uploader, sess.RealDir(), retention.APISessionDir(recordingsDir, sess.RealDir()), time.Unix(0, sess.StartUnixNs()), uploadDelete, opts, nil)
+			cancel()
 		}
 	}
 	uploadWG.Wait()
@@ -314,29 +355,58 @@ loop:
 			"dir", sess.RealDir(),
 			"note", "it will be dated on a later start if it ever syncs")
 	}
+
+	if restartRequested {
+		os.Exit(1)
+	}
 }
 
-func registerHeartbeats(mon *heartbeat.Monitor, cfg config.Config, videoHeartbeatNames []string) {
+// registerHeartbeats registers every stream's heartbeat check, wiring the
+// five DDS-backed streams up with a reconnect callback.
+func registerHeartbeats(mon *heartbeat.Monitor, cfg config.Config, videoHeartbeatNames []string, streams func() *persistentStreams) {
 	if cfg.EnableLidar {
-		mon.Register(lidar.HeartbeatName, 10) // ~10 Hz nominal
+		mon.RegisterRecoverable(lidar.HeartbeatName, 10, func() {
+			if p := streams(); p != nil && p.lidarStream != nil {
+				p.lidarStream.Reconnect()
+			}
+		})
 	}
 	if cfg.EnablePointCloud {
-		mon.Register(pointcloud.HeartbeatName, 10) // ~10 Hz nominal
+		mon.RegisterRecoverable(pointcloud.HeartbeatName, 10, func() {
+			if p := streams(); p != nil && p.pointCloudStream != nil {
+				p.pointCloudStream.Reconnect()
+			}
+		})
 	}
 	if cfg.EnableDepth {
-		mon.Register(depth.HeartbeatName, 10) // measured 14.9 Hz on a G1; floor at 10
+		mon.RegisterRecoverable(depth.HeartbeatName, 10, func() {
+			if p := streams(); p != nil && p.depthStream != nil {
+				p.depthStream.Reconnect()
+			}
+		})
 	}
 	if cfg.EnableOdom {
-		mon.Register(odom.HeartbeatName, 30) // Go2 ~150, G1 ~495; 30 = safe floor
+		mon.RegisterRecoverable(odom.HeartbeatName, 30, func() {
+			if p := streams(); p != nil && p.odomStream != nil {
+				p.odomStream.Reconnect()
+			}
+		})
 	}
 	if cfg.EnableLowstate {
-		mon.Register(lowstate.HeartbeatName, 400) // Go2 ~500, G1 ~1053; 400 = safe floor
+		mon.RegisterRecoverable(lowstate.HeartbeatName, 400, func() {
+			if p := streams(); p != nil && p.lowstateStream != nil {
+				p.lowstateStream.Reconnect()
+			}
+		})
 	}
 	mon.Register(network.HeartbeatName, 0) // 0 = liveness check only
 	mon.Register(audio.HeartbeatName, 0)   // 0 = liveness check only
 	if cfg.Features.Enabled() {
 		mon.Register(features.HeartbeatName, 0) // 0 = liveness check only
 	}
+	if cfg.Traces.Enabled {
+		mon.Register(traces.HeartbeatName, 0)
+	}
 	for _, name := range videoHeartbeatNames {
 		mon.Register(name, 0)
 	}
@@ -364,6 +434,9 @@ func unregisterHeartbeats(mon *heartbeat.Monitor, cfg config.Config, videoHeartb
 	if cfg.Features.Enabled() {
 		mon.Unregister(features.HeartbeatName)
 	}
+	if cfg.Traces.Enabled {
+		mon.Unregister(traces.HeartbeatName)
+	}
 	for _, name := range videoHeartbeatNames {
 		mon.Unregister(name)
 	}
@@ -383,6 +456,7 @@ type persistentStreams struct {
 	lowstateStream   *lowstate.LowstateStream
 	videoStreams     []*video.VideoRTSPStream
 	audioStream      *audio.AudioRTSPStream
+	traceStream      *traces.TraceStream
 }
 
 func newPersistentStreams(cfg config.Config, mon *heartbeat.Monitor, recordingsDir string, sess *session.Session, videoHeartbeatNames []string) persistentStreams {
@@ -415,6 +489,11 @@ func newPersistentStreams(cfg config.Config, mon *heartbeat.Monitor, recordingsD
 		lowstateCfg.Monitor = mon
 		p.lowstateStream = lowstate.New(lowstateCfg)
 	}
+	if cfg.Traces.Enabled {
+		tracesCfg := cfg.Traces.TraceStreamConfig()
+		tracesCfg.Monitor = mon
+		p.traceStream = traces.New(tracesCfg)
+	}
 
 	p.videoStreams = make([]*video.VideoRTSPStream, 0, len(cfg.Video))
 	for i, vc := range cfg.Video {
@@ -467,6 +546,9 @@ func (p *persistentStreams) start() {
 	if p.lowstateStream != nil {
 		p.lowstateStream.Start()
 	}
+	if p.traceStream != nil {
+		p.traceStream.Start()
+	}
 	for _, vs := range p.videoStreams {
 		vs.Start()
 	}
@@ -489,6 +571,9 @@ func (p *persistentStreams) stop() {
 	if p.lowstateStream != nil {
 		p.lowstateStream.Stop()
 	}
+	if p.traceStream != nil {
+		p.traceStream.Stop()
+	}
 	for _, vs := range p.videoStreams {
 		vs.Stop()
 	}
@@ -525,6 +610,11 @@ func (p *persistentStreams) rotate(cfg config.Config, sess *session.Session) {
 			slog.Error("lowstate: rotate failed", "err", err)
 		}
 	}
+	if p.traceStream != nil {
+		if err := p.traceStream.Rotate(cfg.Traces.OutputFile); err != nil {
+			slog.Error("traces: rotate failed", "err", err)
+		}
+	}
 	for i, vs := range p.videoStreams {
 		vc := cfg.Video[i]
 		out := realPath(vc.OutputFile, sessionRealDir)
@@ -536,6 +626,25 @@ func (p *persistentStreams) rotate(cfg config.Config, sess *session.Session) {
 	p.audioStream.Rotate(sessionStart, audioTs, framesFileFor(audioOut, "_packets.csv"))
 }
 
+// awaitSegments waits for every video/audio stream to relocate
+// sessionStart's segment before upload -- ffmpeg closes segments a full rotation interval late, so this must run first.
+func (p *persistentStreams) awaitSegments(ctx context.Context, sessionStart time.Time) {
+	var wg sync.WaitGroup
+	for _, vs := range p.videoStreams {
+		wg.Add(1)
+		go func(vs *video.VideoRTSPStream) {
+			defer wg.Done()
+			vs.WaitSegment(ctx, sessionStart)
+		}(vs)
+	}
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		p.audioStream.WaitSegment(ctx, sessionStart)
+	}()
+	wg.Wait()
+}
+
 // realPath rewrites path to sit in realDir instead of whatever directory it
 // was built from.
 func realPath(path, realDir string) string {