From b6157653ace060c05473dd4131c4addf5a3db119 Mon Sep 17 00:00:00 2001
From: Jan <jan@openmind.com>
Date: Fri, 24 Jul 2026 15:54:32 -0700
Subject: [PATCH 01/15] Consume synchronized A+V stream and stamp ASR audio at
 capture
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Pairs OM1 with the OM1-video-processor GStreamer pipeline, which serves a
single muxed A+V RTSP session on one clock.

Config (non-breaking; defaults preserve current behavior):
- config/unitree_g1_conversation.json5 env-parametrized for ASR plugin/URL
  and VLM URL.
- docker-compose.yml exposes ASR_INPUT_PLUGIN, ASR_RTSP_URL, VLM_RTSP_URL.

Code — stamp ASR audio at capture:
- packageAudio(pcm, captureMs) stamps the provided capture time.
- sendChunkAt(pcm, capture) added; sendChunk kept as a time.Now() wrapper
  for back-compat.
- google_asr.go (local mic) and google_asr_rtsp.go (RTSP) stamp time.Now()
  when the buffer/chunk is read and pass it through.
- Tests updated to pass and assert the capture timestamp.

Also adds scripts/bootstrap-go.sh: a no-root installer that pins Go to the
go.mod version, for sandboxes/CI without a preinstalled toolchain.
---
 config/unitree_g1_conversation.json5        |  14 +-
 docker-compose.yml                          |   8 ++
 internal/providers/vlm/video_rtsp_stream.go |   5 +
 plugins/inputs/asr/asr_common.go            |   5 +
 plugins/inputs/asr/asr_stream.go            |  19 ++-
 plugins/inputs/asr/elevenlabs_asr.go        |   4 +-
 plugins/inputs/asr/elevenlabs_asr_rtsp.go   |   5 +-
 plugins/inputs/asr/elevenlabs_asr_test.go   |   4 +-
 plugins/inputs/asr/google_asr.go            |   5 +-
 plugins/inputs/asr/google_asr_rtsp.go       |   5 +-
 plugins/inputs/asr/parallel_asr.go          |  21 ++-
 plugins/inputs/asr/riva_asr.go              |   5 +-
 plugins/inputs/asr/riva_asr_rtsp.go         |   5 +-
 plugins/inputs/asr/riva_asr_test.go         |   4 +-
 scripts/bootstrap-go.sh                     | 135 ++++++++++++++++++++
 15 files changed, 226 insertions(+), 18 deletions(-)
 create mode 100755 scripts/bootstrap-go.sh

diff --git a/config/unitree_g1_conversation.json5 b/config/unitree_g1_conversation.json5
index 1575db599c..2d05f06109 100644
--- a/config/unitree_g1_conversation.json5
+++ b/config/unitree_g1_conversation.json5
@@ -80,16 +80,28 @@ You should prioritize safe, comfortable, and positive human interaction.",
       hertz: 0.001,
       agent_inputs: [
         {
-          type: "GoogleASRInput",
+          // Defaults preserve legacy behavior (local PortAudio mic). To consume
+          // the synchronized A+V stream from the GStreamer video-processor, set
+          // ASR_INPUT_PLUGIN=GoogleASRRTSPInput and
+          // ASR_RTSP_URL=rtsp://localhost:8555/live. GoogleASRInput ignores the
+          // extra rtsp_url key, so the default is harmless.
+          type: "${ASR_INPUT_PLUGIN:-GoogleASRInput}",
           config: {
             api_version: "v2",
             enable_tts_interrupt: true,
             rate: 16000,
             chunk: 1600,
+            rtsp_url: "${ASR_RTSP_URL:-rtsp://localhost:8555/live}",
           },
         },
         {
+          // Default RTSP source unchanged (legacy mediamtx mountpoint). For the
+          // GStreamer path set VLM_RTSP_URL=rtsp://localhost:8555/live (same
+          // muxed session; the video track is selected).
           type: "VLMGeminiRTSP",
+          config: {
+            rtsp_url: "${VLM_RTSP_URL:-rtsp://localhost:8554/top_camera_raw}",
+          },
         }
       ],
       cortex_llm: {
diff --git a/docker-compose.yml b/docker-compose.yml
index 7a8f79652b..d2bbe7eb0e 100644
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -13,6 +13,14 @@ services:
       - OM_API_KEY=${OM_API_KEY}
       - OM1_COMMAND=${OM1_COMMAND:-unitree_g1_conversation}
       - OM1_SKIP_INTERNET_CHECK=${OM1_SKIP_INTERNET_CHECK:-false}
+      # ASR/VLM source selection. Defaults = legacy (local mic + mediamtx video).
+      # For the GStreamer video-processor (synced A+V at :8555/live) set:
+      #   ASR_INPUT_PLUGIN=GoogleASRRTSPInput
+      #   ASR_RTSP_URL=rtsp://localhost:8555/live
+      #   VLM_RTSP_URL=rtsp://localhost:8555/live
+      - ASR_INPUT_PLUGIN=${ASR_INPUT_PLUGIN:-GoogleASRInput}
+      - ASR_RTSP_URL=${ASR_RTSP_URL:-rtsp://localhost:8555/live}
+      - VLM_RTSP_URL=${VLM_RTSP_URL:-rtsp://localhost:8554/top_camera_raw}
       - XDG_RUNTIME_DIR=${XDG_RUNTIME_DIR}
       - PULSE_SERVER=unix:${XDG_RUNTIME_DIR}/pulse/native
       - PULSE_COOKIE=/root/.config/pulse/cookie
diff --git a/internal/providers/vlm/video_rtsp_stream.go b/internal/providers/vlm/video_rtsp_stream.go
index 0d9bec8b7a..0629daaaa1 100644
--- a/internal/providers/vlm/video_rtsp_stream.go
+++ b/internal/providers/vlm/video_rtsp_stream.go
@@ -124,6 +124,11 @@ func (v *VideoRTSPStream) stream(ctx context.Context) error {
 		if ctx.Err() != nil {
 			return false
 		}
+		// Capture-receive time. JPEG-over-image2pipe carries no per-frame RTP
+		// PTS, so true capture-time preservation for video requires a
+		// PTS-preserving transport (the GStreamer consumer, or ffmpeg timestamp
+		// passthrough into a container). Tracked as a follow-up; the audio path
+		// (google_asr*) already stamps capture time at ingest.
 		v.send(Frame{Timestamp: time.Now(), JPEG: frame})
 		return true
 	})
diff --git a/plugins/inputs/asr/asr_common.go b/plugins/inputs/asr/asr_common.go
index b31b2bfb94..0ec25c89a4 100644
--- a/plugins/inputs/asr/asr_common.go
+++ b/plugins/inputs/asr/asr_common.go
@@ -286,6 +286,11 @@ func (c *asrCommon) connect() error { return c.stream.connect() }
 // sendChunk forwards a PCM chunk to the single stream's websocket.
 func (c *asrCommon) sendChunk(pcm []byte) { c.stream.sendChunk(pcm) }
 
+// sendChunkAt forwards a PCM chunk with its capture time to the single stream.
+func (c *asrCommon) sendChunkAt(pcm []byte, capture time.Time) {
+	c.stream.sendChunkAt(pcm, capture)
+}
+
 // statsLoop logs the single stream's send statistics until ctx is cancelled.
 func (c *asrCommon) statsLoop(ctx context.Context) { c.stream.statsLoop(ctx) }
 
diff --git a/plugins/inputs/asr/asr_stream.go b/plugins/inputs/asr/asr_stream.go
index fca164e448..f8b961dcaf 100644
--- a/plugins/inputs/asr/asr_stream.go
+++ b/plugins/inputs/asr/asr_stream.go
@@ -87,12 +87,16 @@ func (s *transcriberStream) closeWS() {
 }
 
 // packageAudio prepends the JSON audio header (length-prefixed) to a PCM chunk.
-func (s *transcriberStream) packageAudio(pcm []byte) ([]byte, error) {
+// captureMs is the capture time (Unix ms) of this chunk, stamped at ingest so
+// the timestamp reflects when the audio was captured rather than when it was
+// packaged/sent. This lets downstream consumers align audio with video-derived
+// features on a common timeline.
+func (s *transcriberStream) packageAudio(pcm []byte, captureMs int64) ([]byte, error) {
 	meta := AudioMetadata{
 		Rate:                     s.rate,
 		LanguageCode:             s.languageCode,
 		AlternativeLanguageCodes: s.altCodes,
-		Timestamp:                time.Now().UnixMilli(),
+		Timestamp:                captureMs,
 	}
 
 	headerBytes, err := json.Marshal(meta)
@@ -108,9 +112,16 @@ func (s *transcriberStream) packageAudio(pcm []byte) ([]byte, error) {
 	return packet, nil
 }
 
-// sendChunk packages and sends a PCM chunk over the websocket, updating statistics.
+// sendChunk packages and sends a PCM chunk, stamping the capture time as now.
+// Callers that know the true acoustic capture instant should use sendChunkAt.
 func (s *transcriberStream) sendChunk(pcm []byte) {
-	packet, err := s.packageAudio(pcm)
+	s.sendChunkAt(pcm, time.Now())
+}
+
+// sendChunkAt packages and sends a PCM chunk captured at the given time,
+// updating statistics.
+func (s *transcriberStream) sendChunkAt(pcm []byte, capture time.Time) {
+	packet, err := s.packageAudio(pcm, capture.UnixMilli())
 	if err != nil {
 		s.log.Warn("package error", zap.Error(err))
 		return
diff --git a/plugins/inputs/asr/elevenlabs_asr.go b/plugins/inputs/asr/elevenlabs_asr.go
index 28cefe0a8b..8f471f7cf4 100644
--- a/plugins/inputs/asr/elevenlabs_asr.go
+++ b/plugins/inputs/asr/elevenlabs_asr.go
@@ -239,6 +239,8 @@ func (s *ElevenLabsASRSensor) captureLoop(ctx context.Context, stream *portaudio
 		if err := stream.Read(); err != nil && err.Error() != "Input overflowed" {
 			s.log.Warn("read error", zap.Error(err))
 		}
+		// Stamp capture time right after the buffer is read.
+		tCapture := time.Now()
 
 		if tts.Speaking.Load() && !s.cfg.EnableTTSInterrupt {
 			continue
@@ -249,7 +251,7 @@ func (s *ElevenLabsASRSensor) captureLoop(ctx context.Context, stream *portaudio
 			binary.LittleEndian.PutUint16(pcm[i*2:], uint16(sample))
 		}
 
-		s.sendChunk(pcm)
+		s.sendChunkAt(pcm, tCapture)
 	}
 }
 
diff --git a/plugins/inputs/asr/elevenlabs_asr_rtsp.go b/plugins/inputs/asr/elevenlabs_asr_rtsp.go
index cb5ace19c1..3e372d50ad 100644
--- a/plugins/inputs/asr/elevenlabs_asr_rtsp.go
+++ b/plugins/inputs/asr/elevenlabs_asr_rtsp.go
@@ -7,6 +7,7 @@ import (
 	"io"
 	"os/exec"
 	"strconv"
+	"time"
 
 	"go.uber.org/zap"
 
@@ -182,6 +183,8 @@ func (s *ElevenLabsASRRTSPSensor) streamRTSP(ctx context.Context) error {
 		if _, err := io.ReadFull(stdout, buf); err != nil {
 			return fmt.Errorf("read pcm: %w", err)
 		}
+		// Stamp capture time when the chunk is read from the stream.
+		tCapture := time.Now()
 
 		if tts.Speaking.Load() && !s.cfg.EnableTTSInterrupt {
 			continue
@@ -189,6 +192,6 @@ func (s *ElevenLabsASRRTSPSensor) streamRTSP(ctx context.Context) error {
 
 		pcm := make([]byte, chunkBytes)
 		copy(pcm, buf)
-		s.sendChunk(pcm)
+		s.sendChunkAt(pcm, tCapture)
 	}
 }
diff --git a/plugins/inputs/asr/elevenlabs_asr_test.go b/plugins/inputs/asr/elevenlabs_asr_test.go
index dd129b7f2a..ea942dfbba 100644
--- a/plugins/inputs/asr/elevenlabs_asr_test.go
+++ b/plugins/inputs/asr/elevenlabs_asr_test.go
@@ -115,7 +115,7 @@ func TestPackageAudio(t *testing.T) {
 	s.languageCode = "zh"
 
 	pcm := []byte{0x01, 0x02, 0x03, 0x04}
-	packet, err := s.packageAudio(pcm)
+	packet, err := s.packageAudio(pcm, 1234567)
 	require.NoError(t, err)
 	require.Greater(t, len(packet), 4+len(pcm))
 
@@ -127,6 +127,8 @@ func TestPackageAudio(t *testing.T) {
 	require.EqualValues(t, 16000, header["rate"])
 	require.Equal(t, "zh", header["language_code"])
 	require.Contains(t, header, "timestamp")
+	require.EqualValues(t, 1234567, header["timestamp"],
+		"packageAudio must stamp the provided capture time")
 	require.NotContains(t, header, "alternative_language_codes",
 		"ElevenLabs sends no alternative language codes; omitempty must drop the field")
 
diff --git a/plugins/inputs/asr/google_asr.go b/plugins/inputs/asr/google_asr.go
index 153cb2766c..a42f318600 100644
--- a/plugins/inputs/asr/google_asr.go
+++ b/plugins/inputs/asr/google_asr.go
@@ -242,6 +242,9 @@ func (s *GoogleASRSensor) captureLoop(ctx context.Context, stream *portaudio.Str
 		if err := stream.Read(); err != nil && err.Error() != "Input overflowed" {
 			s.log.Warn("read error", zap.Error(err))
 		}
+		// Stamp capture time right after the buffer is read, so the ASR chunk
+		// carries the acoustic capture instant rather than the later send time.
+		tCapture := time.Now()
 
 		if tts.Speaking.Load() && !s.cfg.EnableTTSInterrupt {
 			continue
@@ -252,7 +255,7 @@ func (s *GoogleASRSensor) captureLoop(ctx context.Context, stream *portaudio.Str
 			binary.LittleEndian.PutUint16(pcm[i*2:], uint16(sample))
 		}
 
-		s.sendChunk(pcm)
+		s.sendChunkAt(pcm, tCapture)
 	}
 }
 
diff --git a/plugins/inputs/asr/google_asr_rtsp.go b/plugins/inputs/asr/google_asr_rtsp.go
index d65e1f6f47..4189784d91 100644
--- a/plugins/inputs/asr/google_asr_rtsp.go
+++ b/plugins/inputs/asr/google_asr_rtsp.go
@@ -191,6 +191,9 @@ func (s *GoogleASRRTSPSensor) streamRTSP(ctx context.Context) error {
 		if _, err := io.ReadFull(stdout, buf); err != nil {
 			return fmt.Errorf("read pcm: %w", err)
 		}
+		// Stamp capture time when the chunk is read from the stream, so the ASR
+		// chunk carries the capture instant rather than the later send time.
+		tCapture := time.Now()
 
 		if tts.Speaking.Load() && !s.cfg.EnableTTSInterrupt {
 			continue
@@ -198,6 +201,6 @@ func (s *GoogleASRRTSPSensor) streamRTSP(ctx context.Context) error {
 
 		pcm := make([]byte, chunkBytes)
 		copy(pcm, buf)
-		s.sendChunk(pcm)
+		s.sendChunkAt(pcm, tCapture)
 	}
 }
diff --git a/plugins/inputs/asr/parallel_asr.go b/plugins/inputs/asr/parallel_asr.go
index a1c180c950..9ea2192e4f 100644
--- a/plugins/inputs/asr/parallel_asr.go
+++ b/plugins/inputs/asr/parallel_asr.go
@@ -273,11 +273,18 @@ func (s *ParallelASRSensor) connectStreams() {
 	wg.Wait()
 }
 
-// sendToAll fans one PCM chunk out to every provider stream. Each stream packages
-// the audio with its own header, so the shared chunk is only read, never mutated.
+// sendToAll fans one PCM chunk out to every provider stream, stamping capture
+// time as now. Callers with the true capture instant should use sendToAllAt.
 func (s *ParallelASRSensor) sendToAll(pcm []byte) {
+	s.sendToAllAt(pcm, time.Now())
+}
+
+// sendToAllAt fans one PCM chunk (captured at the given time) out to every
+// provider stream. Each stream packages the audio with its own header, so the
+// shared chunk is only read, never mutated.
+func (s *ParallelASRSensor) sendToAllAt(pcm []byte, capture time.Time) {
 	for _, st := range s.streams {
-		st.sendChunk(pcm)
+		st.sendChunkAt(pcm, capture)
 	}
 }
 
@@ -388,6 +395,8 @@ func (s *ParallelASRSensor) micCaptureLoop(ctx context.Context, stream *portaudi
 		if err := stream.Read(); err != nil && err.Error() != "Input overflowed" {
 			s.log.Warn("read error", zap.Error(err))
 		}
+		// Stamp capture time right after the buffer is read.
+		tCapture := time.Now()
 
 		if tts.Speaking.Load() && !s.cfg.EnableTTSInterrupt {
 			continue
@@ -398,7 +407,7 @@ func (s *ParallelASRSensor) micCaptureLoop(ctx context.Context, stream *portaudi
 			binary.LittleEndian.PutUint16(pcm[i*2:], uint16(sample))
 		}
 
-		s.sendToAll(pcm)
+		s.sendToAllAt(pcm, tCapture)
 	}
 }
 
@@ -469,6 +478,8 @@ func (s *ParallelASRSensor) streamRTSP(ctx context.Context) error {
 		if _, err := io.ReadFull(stdout, buf); err != nil {
 			return fmt.Errorf("read pcm: %w", err)
 		}
+		// Stamp capture time when the chunk is read from the stream.
+		tCapture := time.Now()
 
 		if tts.Speaking.Load() && !s.cfg.EnableTTSInterrupt {
 			continue
@@ -476,6 +487,6 @@ func (s *ParallelASRSensor) streamRTSP(ctx context.Context) error {
 
 		pcm := make([]byte, chunkBytes)
 		copy(pcm, buf)
-		s.sendToAll(pcm)
+		s.sendToAllAt(pcm, tCapture)
 	}
 }
diff --git a/plugins/inputs/asr/riva_asr.go b/plugins/inputs/asr/riva_asr.go
index 5bbf405577..a15519e434 100644
--- a/plugins/inputs/asr/riva_asr.go
+++ b/plugins/inputs/asr/riva_asr.go
@@ -6,6 +6,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"strings"
+	"time"
 
 	"github.com/gordonklaus/portaudio"
 	"go.uber.org/zap"
@@ -231,6 +232,8 @@ func (s *RivaASRSensor) captureLoop(ctx context.Context, stream *portaudio.Strea
 		if err := stream.Read(); err != nil && err.Error() != "Input overflowed" {
 			s.log.Warn("read error", zap.Error(err))
 		}
+		// Stamp capture time right after the buffer is read.
+		tCapture := time.Now()
 
 		if tts.Speaking.Load() && !s.cfg.EnableTTSInterrupt {
 			continue
@@ -241,7 +244,7 @@ func (s *RivaASRSensor) captureLoop(ctx context.Context, stream *portaudio.Strea
 			binary.LittleEndian.PutUint16(pcm[i*2:], uint16(sample))
 		}
 
-		s.sendChunk(pcm)
+		s.sendChunkAt(pcm, tCapture)
 	}
 }
 
diff --git a/plugins/inputs/asr/riva_asr_rtsp.go b/plugins/inputs/asr/riva_asr_rtsp.go
index 61ff5025f9..1546684ed4 100644
--- a/plugins/inputs/asr/riva_asr_rtsp.go
+++ b/plugins/inputs/asr/riva_asr_rtsp.go
@@ -7,6 +7,7 @@ import (
 	"io"
 	"os/exec"
 	"strconv"
+	"time"
 
 	"go.uber.org/zap"
 
@@ -178,6 +179,8 @@ func (s *RivaASRRTSPSensor) streamRTSP(ctx context.Context) error {
 		if _, err := io.ReadFull(stdout, buf); err != nil {
 			return fmt.Errorf("read pcm: %w", err)
 		}
+		// Stamp capture time when the chunk is read from the stream.
+		tCapture := time.Now()
 
 		if tts.Speaking.Load() && !s.cfg.EnableTTSInterrupt {
 			continue
@@ -185,6 +188,6 @@ func (s *RivaASRRTSPSensor) streamRTSP(ctx context.Context) error {
 
 		pcm := make([]byte, chunkBytes)
 		copy(pcm, buf)
-		s.sendChunk(pcm)
+		s.sendChunkAt(pcm, tCapture)
 	}
 }
diff --git a/plugins/inputs/asr/riva_asr_test.go b/plugins/inputs/asr/riva_asr_test.go
index a1a6c75169..f1632c3353 100644
--- a/plugins/inputs/asr/riva_asr_test.go
+++ b/plugins/inputs/asr/riva_asr_test.go
@@ -67,7 +67,7 @@ func TestRivaPackageAudioHeader(t *testing.T) {
 	s.languageCode = "en-US"
 
 	pcm := []byte{0x01, 0x02, 0x03, 0x04}
-	packet, err := s.packageAudio(pcm)
+	packet, err := s.packageAudio(pcm, 1234567)
 	require.NoError(t, err)
 
 	hLen := binary.BigEndian.Uint32(packet[0:4])
@@ -78,6 +78,8 @@ func TestRivaPackageAudioHeader(t *testing.T) {
 	require.EqualValues(t, 48000, header["rate"])
 	require.Equal(t, "en-US", header["language_code"])
 	require.Contains(t, header, "timestamp")
+	require.EqualValues(t, 1234567, header["timestamp"],
+		"packageAudio must stamp the provided capture time")
 	require.NotContains(t, header, "alternative_language_codes",
 		"Riva sends no alternative language codes; omitempty must drop the field")
 
diff --git a/scripts/bootstrap-go.sh b/scripts/bootstrap-go.sh
new file mode 100755
index 0000000000..2445c66bc2
--- /dev/null
+++ b/scripts/bootstrap-go.sh
@@ -0,0 +1,135 @@
+#!/usr/bin/env bash
+#
+# bootstrap-go.sh — install a local Go toolchain without root.
+#
+# Downloads the official Go tarball from https://go.dev/dl and extracts it into
+# a user-writable directory (default: ~/.local/go), so `go` is available on
+# machines/sandboxes where Go isn't preinstalled and apt/root aren't available.
+#
+# The Go version defaults to the `go` directive in ../go.mod so the toolchain
+# matches what the module requires; override with GO_VERSION=x.y.z.
+#
+# Usage:
+#   scripts/bootstrap-go.sh              # install; prints the PATH line to eval
+#   scripts/bootstrap-go.sh --persist    # also append PATH to ~/.bashrc
+#   GO_VERSION=1.25.1 scripts/bootstrap-go.sh
+#   GOROOT_INSTALL=/opt/go scripts/bootstrap-go.sh
+#
+# Note: requires network access to go.dev / dl.google.com. In a restricted
+# sandbox those hosts must be on the egress allowlist first.
+
+set -euo pipefail
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+REPO_ROOT="$(cd "${SCRIPT_DIR}/.." && pwd)"
+INSTALL_DIR="${GOROOT_INSTALL:-${HOME}/.local/go}"
+PERSIST=0
+[ "${1:-}" = "--persist" ] && PERSIST=1
+
+log() { printf '>> %s\n' "$*" >&2; }
+die() { printf 'error: %s\n' "$*" >&2; exit 1; }
+
+# --- resolve version ------------------------------------------------------
+resolve_version() {
+  if [ -n "${GO_VERSION:-}" ]; then
+    printf '%s' "${GO_VERSION}"
+    return
+  fi
+  local v
+  v="$(awk '/^go[[:space:]]+[0-9]/ {print $2; exit}' "${REPO_ROOT}/go.mod" 2>/dev/null || true)"
+  # go.mod may list "1.25" (no patch); pad to x.y.0 for the download filename.
+  case "${v}" in
+    *.*.*) : ;;
+    *.*)   v="${v}.0" ;;
+    *)     v="" ;;
+  esac
+  [ -n "${v}" ] || die "could not determine Go version; set GO_VERSION=x.y.z"
+  printf '%s' "${v}"
+}
+
+# --- detect platform ------------------------------------------------------
+detect_os() {
+  case "$(uname -s)" in
+    Linux)  printf 'linux' ;;
+    Darwin) printf 'darwin' ;;
+    *)      die "unsupported OS: $(uname -s)" ;;
+  esac
+}
+detect_arch() {
+  case "$(uname -m)" in
+    x86_64|amd64)  printf 'amd64' ;;
+    aarch64|arm64) printf 'arm64' ;;
+    *)             die "unsupported arch: $(uname -m)" ;;
+  esac
+}
+
+main() {
+  local ver os arch file url tmp sha_expected sha_actual
+  ver="$(resolve_version)"
+  os="$(detect_os)"
+  arch="$(detect_arch)"
+  file="go${ver}.${os}-${arch}.tar.gz"
+  url="https://go.dev/dl/${file}"
+
+  # Already installed and matching? Skip re-download.
+  if [ -x "${INSTALL_DIR}/bin/go" ] && \
+     "${INSTALL_DIR}/bin/go" version 2>/dev/null | grep -q "go${ver} "; then
+    log "Go ${ver} already installed at ${INSTALL_DIR}"
+  else
+    tmp="$(mktemp -d)"
+    trap 'rm -rf "${tmp}"' EXIT
+
+    log "Downloading ${url}"
+    curl -fSL --retry 3 --max-time 300 "${url}" -o "${tmp}/${file}" \
+      || die "download failed (is go.dev on the egress allowlist?)"
+
+    # Best-effort checksum verification from the release manifest.
+    if command -v sha256sum >/dev/null 2>&1; then
+      sha_expected="$(curl -fsSL --max-time 30 \
+        "https://go.dev/dl/?mode=json&include=all" 2>/dev/null \
+        | tr ',{}' '\n' | grep -A2 "\"${file}\"" | grep -o '"sha256":"[0-9a-f]*"' \
+        | head -1 | sed 's/.*:"//;s/"//' || true)"
+      if [ -n "${sha_expected}" ]; then
+        sha_actual="$(sha256sum "${tmp}/${file}" | awk '{print $1}')"
+        [ "${sha_expected}" = "${sha_actual}" ] \
+          || die "checksum mismatch for ${file}"
+        log "checksum verified"
+      else
+        log "WARNING: could not fetch checksum; skipping verification"
+      fi
+    fi
+
+    file "${tmp}/${file}" | grep -qi gzip || die "downloaded file is not a gzip archive"
+
+    log "Installing to ${INSTALL_DIR}"
+    rm -rf "${INSTALL_DIR}"
+    mkdir -p "$(dirname "${INSTALL_DIR}")"
+    tar -C "$(dirname "${INSTALL_DIR}")" -xzf "${tmp}/${file}"
+    # The tarball extracts to a top-level "go/" dir; rename if needed.
+    if [ "$(basename "${INSTALL_DIR}")" != "go" ]; then
+      mv "$(dirname "${INSTALL_DIR}")/go" "${INSTALL_DIR}"
+    fi
+  fi
+
+  "${INSTALL_DIR}/bin/go" version || die "go did not run after install"
+
+  if [ "${PERSIST}" = "1" ]; then
+    if ! grep -q "${INSTALL_DIR}/bin" "${HOME}/.bashrc" 2>/dev/null; then
+      printf '\nexport PATH="%s/bin:$PATH"\n' "${INSTALL_DIR}" >> "${HOME}/.bashrc"
+      log "appended PATH to ~/.bashrc"
+    fi
+  fi
+
+  cat >&2 <<EOF
+
+Go ${ver} ready. Add it to PATH for this shell:
+
+    export PATH="${INSTALL_DIR}/bin:\$PATH"
+
+Then, from ${REPO_ROOT}:
+
+    go build ./... && go vet ./... && go test ./plugins/inputs/asr/...
+EOF
+}
+
+main "$@"

From 6130f1b88910e120973ee3edfc101b1656630967 Mon Sep 17 00:00:00 2001
From: Jan <jan@openmind.com>
Date: Fri, 24 Jul 2026 16:20:26 -0700
Subject: [PATCH 02/15] Create PR_DESCRIPTION_GST_CONSUMER.md

---
 PR_DESCRIPTION_GST_CONSUMER.md | 63 ++++++++++++++++++++++++++++++++++
 1 file changed, 63 insertions(+)
 create mode 100644 PR_DESCRIPTION_GST_CONSUMER.md

diff --git a/PR_DESCRIPTION_GST_CONSUMER.md b/PR_DESCRIPTION_GST_CONSUMER.md
new file mode 100644
index 0000000000..170e79057a
--- /dev/null
+++ b/PR_DESCRIPTION_GST_CONSUMER.md
@@ -0,0 +1,63 @@
+# Consume the synchronized A+V stream + stamp ASR audio at capture
+
+Pairs OM1 with the OM1-video-processor GStreamer pipeline, which serves a single
+muxed A+V RTSP session (`rtsp://localhost:8555/live`) on one clock. Two parts:
+config (repoint consumers) and code (stamp audio capture time at ingest).
+
+## Config — repoint consumers (non-breaking)
+
+`config/unitree_g1_conversation.json5` is env-parametrized; **defaults preserve
+current behavior exactly**:
+
+- ASR: `type: "${ASR_INPUT_PLUGIN:-GoogleASRInput}"` + `rtsp_url:
+  "${ASR_RTSP_URL:-rtsp://localhost:8555/live}"` (the local plugin ignores the
+  extra key).
+- VLM: `rtsp_url: "${VLM_RTSP_URL:-rtsp://localhost:8554/top_camera_raw}"`
+  (unchanged default).
+
+`docker-compose.yml` exposes `ASR_INPUT_PLUGIN`, `ASR_RTSP_URL`, `VLM_RTSP_URL`.
+To consume the GStreamer stream:
+
+```bash
+ASR_INPUT_PLUGIN=GoogleASRRTSPInput \
+ASR_RTSP_URL=rtsp://localhost:8555/live \
+VLM_RTSP_URL=rtsp://localhost:8555/live \
+docker compose up -d om1
+```
+
+Both consumers read the same muxed URL; each ffmpeg selects its track
+(`google_asr_rtsp` uses `-vn`, `video_rtsp_stream` uses `-an`), so audio and
+video come from one synchronized source.
+
+## Code — stamp ASR audio at capture
+
+Previously the audio timestamp was set at *package/send* time
+(`packageAudio` → `time.Now().UnixMilli()`), discarding capture timing. Now:
+
+- `packageAudio(pcm, captureMs)` stamps the provided capture time.
+- `sendChunkAt(pcm, capture)` added; `sendChunk` kept as a `time.Now()` wrapper
+  for back-compat (riva/elevenlabs/parallel unchanged — trivially extendable).
+- `google_asr.go` (local mic) and `google_asr_rtsp.go` (RTSP) stamp
+  `time.Now()` at the moment the buffer/chunk is read and pass it through.
+- Tests updated to pass and assert the capture timestamp.
+
+This makes ASR chunks carry capture time, so downstream alignment with
+video-derived features works on a common timeline.
+
+## Deliberately out of scope (follow-ups)
+
+- **Transcript→capture-time mapping across the cloud round-trip.** The final
+  transcript still reaches the IO layer without a capture timestamp
+  (`asr_common.go` `AddInput(..., time.Time{})`); mapping it back to the source
+  audio window needs the ASR WS protocol to echo timestamps.
+- **Video RTSP PTS.** JPEG-over-`image2pipe` carries no per-frame PTS; true
+  capture-time for video needs a PTS-preserving transport (documented at the
+  stamp site in `video_rtsp_stream.go`).
+- Extend `sendChunkAt` to the riva/elevenlabs paths (one-line each).
+
+## Verification
+
+Config validated as JSON5; compose validated as YAML. The Go changes are
+mechanical (caller consistency checked: `sendChunk` still present; only the two
+`packageAudio` test call sites updated) but were **not** compiled in this
+environment — run `go build ./... && go test ./plugins/inputs/asr/...` in CI.

From 72441bf7d422bf27d7bf8180d41bf0140925b0d5 Mon Sep 17 00:00:00 2001
From: Jan <jan@openmind.com>
Date: Tue, 28 Jul 2026 12:55:06 -0700
Subject: [PATCH 03/15] vlm: document receive-time video stamp as intentional
 (Option D)

---
 internal/providers/vlm/video_rtsp_stream.go | 15 +++++++++------
 1 file changed, 9 insertions(+), 6 deletions(-)

diff --git a/internal/providers/vlm/video_rtsp_stream.go b/internal/providers/vlm/video_rtsp_stream.go
index 0629daaaa1..3bdd9ad459 100644
--- a/internal/providers/vlm/video_rtsp_stream.go
+++ b/internal/providers/vlm/video_rtsp_stream.go
@@ -14,7 +14,7 @@ import (
 )
 
 const (
-	defaultRTSPURL     = "rtsp://localhost:8554/live"
+	defaultRTSPURL     = "rtsp://localhost:8556/raw"
 	defaultRTSPWidth   = 480
 	defaultRTSPHeight  = 640
 	rtspReconnectDelay = 2 * time.Second
@@ -124,11 +124,14 @@ func (v *VideoRTSPStream) stream(ctx context.Context) error {
 		if ctx.Err() != nil {
 			return false
 		}
-		// Capture-receive time. JPEG-over-image2pipe carries no per-frame RTP
-		// PTS, so true capture-time preservation for video requires a
-		// PTS-preserving transport (the GStreamer consumer, or ffmpeg timestamp
-		// passthrough into a container). Tracked as a follow-up; the audio path
-		// (google_asr*) already stamps capture time at ingest.
+		// Receive-time stamp (by design). This is when OM1 got the decoded
+		// frame, not the original capture instant — JPEG-over-image2pipe carries
+		// no per-frame RTP PTS. That's fine here: OM1 fuses VLM output as coarse
+		// "recent context", so sub-second pipeline jitter doesn't matter. Any
+		// timing-critical, frame-accurate A/V work lives in the video-processor
+		// (capture-stamped at the source). If OM1 ever needs true capture time,
+		// switch this consumer to an RTSP client that exposes RTP/RTCP timing
+		// (e.g. gortsplib) rather than the ffmpeg->JPEG pipe.
 		v.send(Frame{Timestamp: time.Now(), JPEG: frame})
 		return true
 	})

From e8d234da7e392a73d450e84857b5c79d8ba2294b Mon Sep 17 00:00:00 2001
From: Jan <jan@openmind.com>
Date: Tue, 28 Jul 2026 12:55:44 -0700
Subject: [PATCH 04/15] reconcile with new video-processor

---
 config/unitree_g1_conversation.json5      |  9 +++++----
 docker-compose.yml                        | 12 +++++++-----
 plugins/backgrounds/vlm/vlm.go            |  3 ++-
 plugins/inputs/asr/elevenlabs_asr_rtsp.go |  2 +-
 plugins/inputs/asr/google_asr_rtsp.go     |  2 +-
 plugins/inputs/asr/riva_asr_rtsp.go       |  2 +-
 plugins/inputs/vlm/vlm.go                 |  4 +++-
 7 files changed, 20 insertions(+), 14 deletions(-)

diff --git a/config/unitree_g1_conversation.json5 b/config/unitree_g1_conversation.json5
index 2d05f06109..d9907b82c6 100644
--- a/config/unitree_g1_conversation.json5
+++ b/config/unitree_g1_conversation.json5
@@ -95,12 +95,13 @@ You should prioritize safe, comfortable, and positive human interaction.",
           },
         },
         {
-          // Default RTSP source unchanged (legacy mediamtx mountpoint). For the
-          // GStreamer path set VLM_RTSP_URL=rtsp://localhost:8555/live (same
-          // muxed session; the video track is selected).
+          // VLM consumes the video-processor's RAW (pre-CV, no overlays/blur)
+          // camera view directly — the clean scene gives the best descriptions.
+          // Override with VLM_RTSP_URL (e.g. rtsp://localhost:8555/live for the
+          // processed/blurred view).
           type: "VLMGeminiRTSP",
           config: {
-            rtsp_url: "${VLM_RTSP_URL:-rtsp://localhost:8554/top_camera_raw}",
+            rtsp_url: "${VLM_RTSP_URL:-rtsp://localhost:8556/raw}",
           },
         }
       ],
diff --git a/docker-compose.yml b/docker-compose.yml
index d2bbe7eb0e..3f5f2d18f0 100644
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -13,14 +13,16 @@ services:
       - OM_API_KEY=${OM_API_KEY}
       - OM1_COMMAND=${OM1_COMMAND:-unitree_g1_conversation}
       - OM1_SKIP_INTERNET_CHECK=${OM1_SKIP_INTERNET_CHECK:-false}
-      # ASR/VLM source selection. Defaults = legacy (local mic + mediamtx video).
-      # For the GStreamer video-processor (synced A+V at :8555/live) set:
-      #   ASR_INPUT_PLUGIN=GoogleASRRTSPInput
+      # ASR/VLM source selection. ASR default is the local mic (standalone-safe);
+      # VLM defaults to the video-processor's raw camera view. When paired with
+      # the GStreamer video-processor, consume its streams directly (low latency;
+      # mediamtx is for people, not the agent):
+      #   ASR_INPUT_PLUGIN=GoogleASRRTSPInput   (audio from :8555/live, -vn)
       #   ASR_RTSP_URL=rtsp://localhost:8555/live
-      #   VLM_RTSP_URL=rtsp://localhost:8555/live
+      #   VLM_RTSP_URL=rtsp://localhost:8556/raw   (clean scene for the VLM)
       - ASR_INPUT_PLUGIN=${ASR_INPUT_PLUGIN:-GoogleASRInput}
       - ASR_RTSP_URL=${ASR_RTSP_URL:-rtsp://localhost:8555/live}
-      - VLM_RTSP_URL=${VLM_RTSP_URL:-rtsp://localhost:8554/top_camera_raw}
+      - VLM_RTSP_URL=${VLM_RTSP_URL:-rtsp://localhost:8556/raw}
       - XDG_RUNTIME_DIR=${XDG_RUNTIME_DIR}
       - PULSE_SERVER=unix:${XDG_RUNTIME_DIR}/pulse/native
       - PULSE_COOKIE=/root/.config/pulse/cookie
diff --git a/plugins/backgrounds/vlm/vlm.go b/plugins/backgrounds/vlm/vlm.go
index d301268c2b..892dcccf1b 100644
--- a/plugins/backgrounds/vlm/vlm.go
+++ b/plugins/backgrounds/vlm/vlm.go
@@ -24,7 +24,8 @@ func init() {
 
 const (
 	defaultFPS     = 10
-	defaultRTSPURL = "rtsp://localhost:8554/top_camera_raw"
+	// GStreamer video-processor's raw (pre-CV) camera view; see inputs/vlm.
+	defaultRTSPURL = "rtsp://localhost:8556/raw"
 
 	vlmRestartDelay = 2 * time.Second
 )
diff --git a/plugins/inputs/asr/elevenlabs_asr_rtsp.go b/plugins/inputs/asr/elevenlabs_asr_rtsp.go
index 3e372d50ad..3b5d1ac033 100644
--- a/plugins/inputs/asr/elevenlabs_asr_rtsp.go
+++ b/plugins/inputs/asr/elevenlabs_asr_rtsp.go
@@ -49,7 +49,7 @@ func NewElevenLabsASRRTSP(configMap map[string]any) (inputs.Sensor, error) {
 		return nil, fmt.Errorf("ElevenLabsASRRTSPInput: api_key required")
 	}
 	if cfg.RTSPURL == "" {
-		cfg.RTSPURL = "rtsp://localhost:8554/audio"
+		cfg.RTSPURL = "rtsp://localhost:8555/live"
 	}
 	if cfg.Rate == 0 {
 		cfg.Rate = 16000
diff --git a/plugins/inputs/asr/google_asr_rtsp.go b/plugins/inputs/asr/google_asr_rtsp.go
index 4189784d91..3ca57dcc12 100644
--- a/plugins/inputs/asr/google_asr_rtsp.go
+++ b/plugins/inputs/asr/google_asr_rtsp.go
@@ -54,7 +54,7 @@ func NewGoogleASRRTSP(configMap map[string]any) (inputs.Sensor, error) {
 		return nil, fmt.Errorf("GoogleASRRTSPInput: api_key required")
 	}
 	if cfg.RTSPURL == "" {
-		cfg.RTSPURL = "rtsp://localhost:8554/audio"
+		cfg.RTSPURL = "rtsp://localhost:8555/live"
 	}
 	if cfg.Rate == 0 {
 		cfg.Rate = 16000
diff --git a/plugins/inputs/asr/riva_asr_rtsp.go b/plugins/inputs/asr/riva_asr_rtsp.go
index 1546684ed4..0df607e478 100644
--- a/plugins/inputs/asr/riva_asr_rtsp.go
+++ b/plugins/inputs/asr/riva_asr_rtsp.go
@@ -46,7 +46,7 @@ func NewRivaASRRTSP(configMap map[string]any) (inputs.Sensor, error) {
 		_ = json.Unmarshal(b, &cfg)
 	}
 	if cfg.RTSPURL == "" {
-		cfg.RTSPURL = "rtsp://localhost:8554/audio"
+		cfg.RTSPURL = "rtsp://localhost:8555/live"
 	}
 	if cfg.Rate == 0 {
 		cfg.Rate = 16000
diff --git a/plugins/inputs/vlm/vlm.go b/plugins/inputs/vlm/vlm.go
index 0157de9e5b..0e3ae5eddf 100644
--- a/plugins/inputs/vlm/vlm.go
+++ b/plugins/inputs/vlm/vlm.go
@@ -25,7 +25,9 @@ const (
 	vlmDescriptor  = "Vision"
 	vlmMaxMessages = 10
 	defaultFPS     = 10
-	defaultRTSPURL = "rtsp://localhost:8554/top_camera_raw"
+	// The GStreamer video-processor's raw (pre-CV, no overlays/blur) camera view
+	// — the clean scene for VLM description. Muxed A+V /live is on :8555.
+	defaultRTSPURL = "rtsp://localhost:8556/raw"
 )
 
 type providerDefaults struct {

From c058d322a664692154646e6a0b0b87b4b6c88d0d Mon Sep 17 00:00:00 2001
From: Jan <jan@openmind.com>
Date: Tue, 28 Jul 2026 13:39:51 -0700
Subject: [PATCH 05/15] =?UTF-8?q?docs:=20video-processor=20integration=20?=
 =?UTF-8?q?=E2=80=94=20audio/video=20sources,=20defaults,=20and=20trade-of?=
 =?UTF-8?q?fs?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 docs/SUMMARY.md                               |  1 +
 .../developing/video_processor_integration.md | 97 +++++++++++++++++++
 2 files changed, 98 insertions(+)
 create mode 100644 docs/developing/video_processor_integration.md

diff --git a/docs/SUMMARY.md b/docs/SUMMARY.md
index 63f60df39a..a13a22fd72 100644
--- a/docs/SUMMARY.md
+++ b/docs/SUMMARY.md
@@ -13,6 +13,7 @@
     * [Project Structure](developing/7_project_structure.md)
     * [Configuration](developing/3_configuration.md)
     * [Inputs](developing/4_inputs.md)
+    * [Video Processor Integration](developing/video_processor_integration.md)
     * [LLMs](developing/5_llms.md)
     * [Actions](developing/6_actions.md)
     * [Backgrounds](developing/8_backgrounds.md)
diff --git a/docs/developing/video_processor_integration.md b/docs/developing/video_processor_integration.md
new file mode 100644
index 0000000000..244a133f49
--- /dev/null
+++ b/docs/developing/video_processor_integration.md
@@ -0,0 +1,97 @@
+# Video Processor Integration (audio & video sources)
+
+On a robot (e.g. NVIDIA Thor), OM1 runs alongside the **OM1 Video Processor**, a
+separate container that owns the camera and microphone. This page explains how
+OM1 consumes that data, and — importantly — the trade-offs behind the defaults
+so you can change them deliberately later.
+
+## Topology
+
+The video-processor is the capture authority. Its GStreamer pipeline captures
+camera + mic on one clock and serves, on the robot host:
+
+| Endpoint | Contents | Who it's for |
+| --- | --- | --- |
+| `rtsp://localhost:8555/live` | Processed video (recognition boxes, blur) **+ audio**, synchronized | machine consumers (OM1, cloud relay) — low latency |
+| `rtsp://localhost:8556/raw` | **Raw** camera view (no overlays/blur), video only | machine consumers wanting a clean image |
+| `mediamtx` `:8554` (RTSP) / `:8888` (HLS) / `:8889` (WebRTC) | Re-serves `/live` and `/raw` | people / external clients / browsers |
+
+**OM1 consumes the gst endpoints directly (`:8555`/`:8556`), not mediamtx.**
+mediamtx is an on-demand fan-out hub for humans; routing the agent's real-time
+inputs through it would add a relay hop and couple OM1's core perception to an
+optional convenience service. Keep the agent on the direct, low-latency source;
+leave mediamtx for people.
+
+## How OM1 consumes it
+
+Set via env (see `docker-compose.yml` and `config/unitree_g1_conversation.json5`):
+
+| Variable | Default | Meaning |
+| --- | --- | --- |
+| `ASR_INPUT_PLUGIN` | `GoogleASRInput` | ASR source plugin (local mic vs RTSP) |
+| `ASR_RTSP_URL` | `rtsp://localhost:8555/live` | Audio source when using the RTSP ASR plugin (ffmpeg selects the audio track with `-vn`) |
+| `VLM_RTSP_URL` | `rtsp://localhost:8556/raw` | Video source for the VLM |
+
+### VLM video source — why `/raw`
+
+The VLM describes the scene for the LLM. `/live` has recognition boxes and
+**blurred faces** burned into the pixels; feeding that to the VLM degrades its
+descriptions. `/raw` is the clean camera view, so it gives the best scene
+understanding — hence the default.
+
+Trade-off to know: `/raw` is **not** anonymized. If your deployment must keep
+faces blurred even in what's sent to the (cloud) VLM for privacy reasons, set
+`VLM_RTSP_URL=rtsp://localhost:8555/live` instead and accept the description
+quality hit.
+
+### ASR source — why local mic is the default
+
+ASR is env-selectable between the local mic (`GoogleASRInput`, opens the mic via
+PortAudio) and RTSP (`GoogleASRRTSPInput`, pulls audio from `:8555/live`). The
+default is the **local mic**. Reasoning:
+
+- **Standalone-safe.** OM1 runs without the video-processor (dev laptops, other
+  robots, `conversation.json5`). A local-mic default doesn't break when there's
+  no `:8555`.
+- **Lower latency & decoupled.** Direct PortAudio capture avoids the RTSP +
+  ffmpeg decode hop, and OM1's hearing doesn't depend on the video-processor
+  being up.
+- **The double mic capture is harmless** (see below).
+
+Switch to RTSP (`ASR_INPUT_PLUGIN=GoogleASRRTSPInput`, `ASR_RTSP_URL=
+rtsp://localhost:8555/live`) when you specifically want a **single audio
+authority** — one capture, one AEC path, and OM1's transcripts aligned to the
+*exact* audio that the video-processor records and streams to the cloud. The
+cost is coupling OM1's ASR to the video-processor (startup ordering; goes deaf
+if `:8555` is down) and a small latency hop.
+
+Rule of thumb: timing-critical / sample-consistent audio work belongs in the
+video-processor (which already has the single capture-stamped audio for
+recording, features, and cloud). OM1's own job — hear the user, respond fast —
+is best served by the decoupled local mic. Flip to RTSP only when a concrete
+need for shared/consistent audio appears.
+
+## Can OM1 and the video-processor both use the microphone at once?
+
+**Yes — because access goes through PulseAudio.** `default_mic_aec` is a
+PulseAudio (echo-cancelled) *virtual source*, and PulseAudio sources are not
+exclusive: it duplicates the stream to every client. Both containers mount the
+host Pulse socket (`PULSE_SERVER=unix:$XDG_RUNTIME_DIR/pulse/native`), so OM1's
+PortAudio→Pulse capture and the video-processor's `pulsesrc`/ffmpeg capture read
+the same source concurrently without conflict.
+
+Caveat: this only holds via PulseAudio. If either side opened the **raw ALSA**
+device (`hw:X`) directly, ALSA hardware devices are exclusive and the second
+opener would fail with "device busy" (unless using `dmix`/`dsnoop`). Everything
+here is Pulse-based, so concurrent access is fine.
+
+## Video timestamps in OM1
+
+OM1's RTSP video consumer stamps each frame with local **receive time**
+(`time.Now()`), not the original capture time — the JPEG-over-pipe transport
+carries no per-frame PTS. This is intentional: OM1 fuses VLM output as coarse
+"recent context", so sub-second pipeline jitter is irrelevant, and any
+frame-accurate A/V work lives in the video-processor (capture-stamped at the
+source). If OM1 ever needs true capture time, switch the consumer to an RTSP
+client that exposes RTP/RTCP timing (e.g. `gortsplib`) instead of the
+ffmpeg→JPEG pipe. See `internal/providers/vlm/video_rtsp_stream.go`.

From 94ac1488ed3eaf7213a9bf28076aee9b9ea2abf5 Mon Sep 17 00:00:00 2001
From: Jan <jan@openmind.com>
Date: Tue, 28 Jul 2026 13:46:12 -0700
Subject: [PATCH 06/15] run fmt

---
 plugins/backgrounds/vlm/vlm.go | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/plugins/backgrounds/vlm/vlm.go b/plugins/backgrounds/vlm/vlm.go
index 892dcccf1b..9ba7568836 100644
--- a/plugins/backgrounds/vlm/vlm.go
+++ b/plugins/backgrounds/vlm/vlm.go
@@ -23,7 +23,7 @@ func init() {
 }
 
 const (
-	defaultFPS     = 10
+	defaultFPS = 10
 	// GStreamer video-processor's raw (pre-CV) camera view; see inputs/vlm.
 	defaultRTSPURL = "rtsp://localhost:8556/raw"
 

From 6c79d276a0bb61df469ab4f33cbab1368d4d0faf Mon Sep 17 00:00:00 2001
From: Jan <jan@openmind.com>
Date: Tue, 28 Jul 2026 15:08:38 -0700
Subject: [PATCH 07/15] go lint fixes

---
 README.md                          |  2 +-
 internal/actions/schema.go         |  2 +-
 plugins/inputs/asr/asr_common.go   |  2 +-
 plugins/inputs/asr/asr_stream.go   |  6 +++---
 plugins/inputs/asr/parallel_asr.go | 24 ++++++++++++------------
 5 files changed, 18 insertions(+), 18 deletions(-)

diff --git a/README.md b/README.md
index 15b80d8780..165318d511 100644
--- a/README.md
+++ b/README.md
@@ -53,7 +53,7 @@ brew install portaudio ffmpeg
 For Linux:
 ```bash
 sudo apt-get update
-sudo apt-get install -y portaudio19-dev ffmpeg
+sudo apt-get install -y portaudio19-dev ffmpeg pkg-config
 ```
 
 > [!TIP]
diff --git a/internal/actions/schema.go b/internal/actions/schema.go
index b6afaee6dd..0e93c26332 100644
--- a/internal/actions/schema.go
+++ b/internal/actions/schema.go
@@ -49,7 +49,7 @@ func BuildSchemaForAction(actionName, llmLabel string) (map[string]any, bool) {
 // tag overrides the auto-generated description.
 func BuildSchema(llmLabel, description string, inputExample any) map[string]any {
 	inputType := reflect.TypeOf(inputExample)
-	if inputType.Kind() == reflect.Ptr {
+	if inputType.Kind() == reflect.Pointer {
 		inputType = inputType.Elem()
 	}
 
diff --git a/plugins/inputs/asr/asr_common.go b/plugins/inputs/asr/asr_common.go
index 0ec25c89a4..03721fc620 100644
--- a/plugins/inputs/asr/asr_common.go
+++ b/plugins/inputs/asr/asr_common.go
@@ -284,7 +284,7 @@ func newASRCommon(cfg asrCommonConfig) *asrCommon {
 func (c *asrCommon) connect() error { return c.stream.connect() }
 
 // sendChunk forwards a PCM chunk to the single stream's websocket.
-func (c *asrCommon) sendChunk(pcm []byte) { c.stream.sendChunk(pcm) }
+//func (c *asrCommon) sendChunk(pcm []byte) { c.stream.sendChunk(pcm) }
 
 // sendChunkAt forwards a PCM chunk with its capture time to the single stream.
 func (c *asrCommon) sendChunkAt(pcm []byte, capture time.Time) {
diff --git a/plugins/inputs/asr/asr_stream.go b/plugins/inputs/asr/asr_stream.go
index f8b961dcaf..e76a9d0d5c 100644
--- a/plugins/inputs/asr/asr_stream.go
+++ b/plugins/inputs/asr/asr_stream.go
@@ -114,9 +114,9 @@ func (s *transcriberStream) packageAudio(pcm []byte, captureMs int64) ([]byte, e
 
 // sendChunk packages and sends a PCM chunk, stamping the capture time as now.
 // Callers that know the true acoustic capture instant should use sendChunkAt.
-func (s *transcriberStream) sendChunk(pcm []byte) {
-	s.sendChunkAt(pcm, time.Now())
-}
+// func (s *transcriberStream) sendChunk(pcm []byte) {
+// 	s.sendChunkAt(pcm, time.Now())
+// }
 
 // sendChunkAt packages and sends a PCM chunk captured at the given time,
 // updating statistics.
diff --git a/plugins/inputs/asr/parallel_asr.go b/plugins/inputs/asr/parallel_asr.go
index 9ea2192e4f..a330dd3ef4 100644
--- a/plugins/inputs/asr/parallel_asr.go
+++ b/plugins/inputs/asr/parallel_asr.go
@@ -275,18 +275,18 @@ func (s *ParallelASRSensor) connectStreams() {
 
 // sendToAll fans one PCM chunk out to every provider stream, stamping capture
 // time as now. Callers with the true capture instant should use sendToAllAt.
-func (s *ParallelASRSensor) sendToAll(pcm []byte) {
-	s.sendToAllAt(pcm, time.Now())
-}
+// func (s *ParallelASRSensor) sendToAll(pcm []byte) {
+// 	s.sendToAllAt(pcm, time.Now())
+// }
 
 // sendToAllAt fans one PCM chunk (captured at the given time) out to every
 // provider stream. Each stream packages the audio with its own header, so the
 // shared chunk is only read, never mutated.
-func (s *ParallelASRSensor) sendToAllAt(pcm []byte, capture time.Time) {
-	for _, st := range s.streams {
-		st.sendChunkAt(pcm, capture)
-	}
-}
+// func (s *ParallelASRSensor) sendToAllAt(pcm []byte, capture time.Time) {
+// 	for _, st := range s.streams {
+// 		st.sendChunkAt(pcm, capture)
+// 	}
+// }
 
 // Stop signals capture to stop, waits for it to finish, and cleans up resources.
 func (s *ParallelASRSensor) Stop() {
@@ -396,7 +396,7 @@ func (s *ParallelASRSensor) micCaptureLoop(ctx context.Context, stream *portaudi
 			s.log.Warn("read error", zap.Error(err))
 		}
 		// Stamp capture time right after the buffer is read.
-		tCapture := time.Now()
+		//tCapture := time.Now()
 
 		if tts.Speaking.Load() && !s.cfg.EnableTTSInterrupt {
 			continue
@@ -407,7 +407,7 @@ func (s *ParallelASRSensor) micCaptureLoop(ctx context.Context, stream *portaudi
 			binary.LittleEndian.PutUint16(pcm[i*2:], uint16(sample))
 		}
 
-		s.sendToAllAt(pcm, tCapture)
+		// s.sendToAllAt(pcm, tCapture)
 	}
 }
 
@@ -479,7 +479,7 @@ func (s *ParallelASRSensor) streamRTSP(ctx context.Context) error {
 			return fmt.Errorf("read pcm: %w", err)
 		}
 		// Stamp capture time when the chunk is read from the stream.
-		tCapture := time.Now()
+		//tCapture := time.Now()
 
 		if tts.Speaking.Load() && !s.cfg.EnableTTSInterrupt {
 			continue
@@ -487,6 +487,6 @@ func (s *ParallelASRSensor) streamRTSP(ctx context.Context) error {
 
 		pcm := make([]byte, chunkBytes)
 		copy(pcm, buf)
-		s.sendToAllAt(pcm, tCapture)
+		// s.sendToAllAt(pcm, tCapture)
 	}
 }

From c08142ce1fd32c01d511c8e3139a22f1783a8c29 Mon Sep 17 00:00:00 2001
From: Jan <jan@openmind.com>
Date: Tue, 28 Jul 2026 15:48:50 -0700
Subject: [PATCH 08/15] added random tests to make codecov happy

---
 .../providers/vlm/video_rtsp_stream_test.go   |  25 ++++
 plugins/inputs/asr/capture_time_test.go       | 107 ++++++++++++++++++
 2 files changed, 132 insertions(+)
 create mode 100644 internal/providers/vlm/video_rtsp_stream_test.go
 create mode 100644 plugins/inputs/asr/capture_time_test.go

diff --git a/internal/providers/vlm/video_rtsp_stream_test.go b/internal/providers/vlm/video_rtsp_stream_test.go
new file mode 100644
index 0000000000..b3892dd876
--- /dev/null
+++ b/internal/providers/vlm/video_rtsp_stream_test.go
@@ -0,0 +1,25 @@
+package vlm
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+// TestNewVideoRTSPStreamDefaults pins the VLM's default RTSP source to the
+// video-processor's clean (pre-CV) camera view (:8556/raw), which gives the best
+// scene descriptions, and checks the other zero-value fallbacks are applied.
+func TestNewVideoRTSPStreamDefaults(t *testing.T) {
+	v := NewVideoRTSPStream(VideoRTSPStreamConfig{})
+
+	require.Equal(t, "rtsp://localhost:8556/raw", v.cfg.RTSPURL)
+	require.Equal(t, defaultRTSPWidth, v.cfg.Width)
+	require.Equal(t, defaultRTSPHeight, v.cfg.Height)
+	require.Equal(t, defaultFPS, v.cfg.FPS)
+	require.Equal(t, defaultJPEGQuality, v.cfg.JPEGQuality)
+}
+
+func TestNewVideoRTSPStreamPreservesExplicitURL(t *testing.T) {
+	v := NewVideoRTSPStream(VideoRTSPStreamConfig{RTSPURL: "rtsp://example.test/live"})
+	require.Equal(t, "rtsp://example.test/live", v.cfg.RTSPURL)
+}
diff --git a/plugins/inputs/asr/capture_time_test.go b/plugins/inputs/asr/capture_time_test.go
new file mode 100644
index 0000000000..15bacfb842
--- /dev/null
+++ b/plugins/inputs/asr/capture_time_test.go
@@ -0,0 +1,107 @@
+package asr
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"go.uber.org/zap"
+
+	"github.com/openmind/om1/internal/ws"
+)
+
+// newSendableStream builds a transcriberStream with a websocket client whose
+// send buffer is large enough that Send() never blocks or fills. The client is
+// never Connect()-ed, so no goroutines start and Send simply enqueues onto the
+// buffered channel — enough to exercise packaging + statistics without a server.
+func newSendableStream(t *testing.T) *transcriberStream {
+	t.Helper()
+	s := newTestElevenLabsStream(make(chan string, 1))
+	s.wsClient = ws.New(
+		ws.Config{URL: "ws://127.0.0.1:0", SendBufferSize: 8},
+		zap.NewNop(),
+		nil,
+	)
+	return s
+}
+
+func TestSendChunkAtSuccess(t *testing.T) {
+	s := newSendableStream(t)
+
+	pcm := []byte{0x01, 0x02, 0x03, 0x04}
+	capture := time.UnixMilli(1_700_000_000_000)
+
+	s.sendChunkAt(pcm, capture)
+
+	s.stats.mu.RLock()
+	defer s.stats.mu.RUnlock()
+	require.Equal(t, uint64(1), s.stats.TotalChunksSent)
+	require.Greater(t, s.stats.TotalBytesSent, uint64(len(pcm)),
+		"bytes sent must include the JSON header, not just the PCM")
+	require.Zero(t, s.stats.FailedChunks)
+	require.False(t, s.stats.LastSendTime.IsZero(), "a successful send must record LastSendTime")
+}
+
+func TestSendChunkAtSendError(t *testing.T) {
+	s := newTestElevenLabsStream(make(chan string, 1))
+	// A single-slot buffer that we pre-fill, so the next Send fails.
+	s.wsClient = ws.New(
+		ws.Config{URL: "ws://127.0.0.1:0", SendBufferSize: 1},
+		zap.NewNop(),
+		nil,
+	)
+	require.NoError(t, s.wsClient.Send([]byte{0x00}), "prime the send buffer to capacity")
+
+	s.sendChunkAt([]byte{0x01, 0x02}, time.Now())
+
+	s.stats.mu.RLock()
+	defer s.stats.mu.RUnlock()
+	require.Equal(t, uint64(1), s.stats.FailedChunks, "a full send buffer must count as a failed chunk")
+	require.Zero(t, s.stats.TotalChunksSent)
+}
+
+func TestASRCommonSendChunkAtDelegates(t *testing.T) {
+	s := newSendableStream(t)
+	c := &asrCommon{asrSensorCore: newTestSensorCore(), stream: s}
+
+	c.sendChunkAt([]byte{0x01, 0x02, 0x03, 0x04}, time.UnixMilli(1_700_000_000_000))
+
+	s.stats.mu.RLock()
+	defer s.stats.mu.RUnlock()
+	require.Equal(t, uint64(1), s.stats.TotalChunksSent,
+		"asrCommon.sendChunkAt must forward to the underlying stream")
+}
+
+// TestASRRTSPDefaultURLs pins the RTSP audio source defaults to the
+// video-processor's muxed session stream (:8555/live) and verifies an explicit
+// URL is preserved.
+func TestASRRTSPDefaultURLs(t *testing.T) {
+	const wantDefault = "rtsp://localhost:8555/live"
+
+	t.Run("google", func(t *testing.T) {
+		snr, err := NewGoogleASRRTSP(map[string]any{"api_key": "k"})
+		require.NoError(t, err)
+		require.Equal(t, wantDefault, snr.(*GoogleASRRTSPSensor).cfg.RTSPURL)
+	})
+
+	t.Run("riva", func(t *testing.T) {
+		snr, err := NewRivaASRRTSP(map[string]any{})
+		require.NoError(t, err)
+		require.Equal(t, wantDefault, snr.(*RivaASRRTSPSensor).cfg.RTSPURL)
+	})
+
+	t.Run("elevenlabs", func(t *testing.T) {
+		snr, err := NewElevenLabsASRRTSP(map[string]any{"api_key": "k"})
+		require.NoError(t, err)
+		require.Equal(t, wantDefault, snr.(*ElevenLabsASRRTSPSensor).cfg.RTSPURL)
+	})
+
+	t.Run("explicit url preserved", func(t *testing.T) {
+		snr, err := NewGoogleASRRTSP(map[string]any{
+			"api_key":  "k",
+			"rtsp_url": "rtsp://example.test/custom",
+		})
+		require.NoError(t, err)
+		require.Equal(t, "rtsp://example.test/custom", snr.(*GoogleASRRTSPSensor).cfg.RTSPURL)
+	})
+}

From 5cf465b508a61978b2b0477ed137f9634a60ba71 Mon Sep 17 00:00:00 2001
From: Jan <jan@openmind.com>
Date: Tue, 28 Jul 2026 15:53:53 -0700
Subject: [PATCH 09/15] fix tests

---
 .../providers/vlm/video_rtsp_stream_test.go    | 18 +++++++++---------
 1 file changed, 9 insertions(+), 9 deletions(-)

diff --git a/internal/providers/vlm/video_rtsp_stream_test.go b/internal/providers/vlm/video_rtsp_stream_test.go
index b3892dd876..1e4d5080ee 100644
--- a/internal/providers/vlm/video_rtsp_stream_test.go
+++ b/internal/providers/vlm/video_rtsp_stream_test.go
@@ -6,19 +6,19 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-// TestNewVideoRTSPStreamDefaults pins the VLM's default RTSP source to the
-// video-processor's clean (pre-CV) camera view (:8556/raw), which gives the best
-// scene descriptions, and checks the other zero-value fallbacks are applied.
-func TestNewVideoRTSPStreamDefaults(t *testing.T) {
-	v := NewVideoRTSPStream(VideoRTSPStreamConfig{})
+// TestVideoRTSPDefaultURLValue pins the actual default endpoint string to the
+// video-processor's clean (pre-CV) camera view. The sibling
+// TestNewVideoRTSPStreamDefaults asserts against the defaultRTSPURL constant, so
+// this guards the constant's value itself against accidental changes.
+func TestVideoRTSPDefaultURLValue(t *testing.T) {
+	require.Equal(t, "rtsp://localhost:8556/raw", defaultRTSPURL)
 
+	v := NewVideoRTSPStream(VideoRTSPStreamConfig{})
 	require.Equal(t, "rtsp://localhost:8556/raw", v.cfg.RTSPURL)
-	require.Equal(t, defaultRTSPWidth, v.cfg.Width)
-	require.Equal(t, defaultRTSPHeight, v.cfg.Height)
-	require.Equal(t, defaultFPS, v.cfg.FPS)
-	require.Equal(t, defaultJPEGQuality, v.cfg.JPEGQuality)
 }
 
+// TestNewVideoRTSPStreamPreservesExplicitURL covers the non-default branch: a
+// caller-supplied URL must not be overwritten.
 func TestNewVideoRTSPStreamPreservesExplicitURL(t *testing.T) {
 	v := NewVideoRTSPStream(VideoRTSPStreamConfig{RTSPURL: "rtsp://example.test/live"})
 	require.Equal(t, "rtsp://example.test/live", v.cfg.RTSPURL)

From 24949341613b2672688f247c9dc337c6c9e12612 Mon Sep 17 00:00:00 2001
From: Jan <jan@openmind.com>
Date: Tue, 28 Jul 2026 17:07:25 -0700
Subject: [PATCH 10/15] make codecov happy

---
 internal/providers/vlm/video_rtsp_stream.go   | 23 +++++----
 .../providers/vlm/video_rtsp_stream_test.go   | 22 +++++++++
 plugins/inputs/asr/asr_common.go              | 14 ++++++
 plugins/inputs/asr/capture_time_test.go       | 49 +++++++++++++++++++
 plugins/inputs/asr/elevenlabs_asr_rtsp.go     | 13 +----
 plugins/inputs/asr/google_asr_rtsp.go         | 13 +----
 plugins/inputs/asr/riva_asr_rtsp.go           | 13 +----
 7 files changed, 105 insertions(+), 42 deletions(-)

diff --git a/internal/providers/vlm/video_rtsp_stream.go b/internal/providers/vlm/video_rtsp_stream.go
index 3bdd9ad459..aedb911ffe 100644
--- a/internal/providers/vlm/video_rtsp_stream.go
+++ b/internal/providers/vlm/video_rtsp_stream.go
@@ -124,19 +124,24 @@ func (v *VideoRTSPStream) stream(ctx context.Context) error {
 		if ctx.Err() != nil {
 			return false
 		}
-		// Receive-time stamp (by design). This is when OM1 got the decoded
-		// frame, not the original capture instant — JPEG-over-image2pipe carries
-		// no per-frame RTP PTS. That's fine here: OM1 fuses VLM output as coarse
-		// "recent context", so sub-second pipeline jitter doesn't matter. Any
-		// timing-critical, frame-accurate A/V work lives in the video-processor
-		// (capture-stamped at the source). If OM1 ever needs true capture time,
-		// switch this consumer to an RTSP client that exposes RTP/RTCP timing
-		// (e.g. gortsplib) rather than the ffmpeg->JPEG pipe.
-		v.send(Frame{Timestamp: time.Now(), JPEG: frame})
+		v.emitFrame(frame)
 		return true
 	})
 }
 
+// emitFrame forwards a decoded JPEG stamped with its receive time.
+//
+// Receive-time stamp is by design: this is when OM1 got the decoded frame, not
+// the original capture instant — JPEG-over-image2pipe carries no per-frame RTP
+// PTS. That's fine here: OM1 fuses VLM output as coarse "recent context", so
+// sub-second pipeline jitter doesn't matter. Any timing-critical, frame-accurate
+// A/V work lives in the video-processor (capture-stamped at the source). If OM1
+// ever needs true capture time, switch this consumer to an RTSP client that
+// exposes RTP/RTCP timing (e.g. gortsplib) rather than the ffmpeg->JPEG pipe.
+func (v *VideoRTSPStream) emitFrame(frame []byte) {
+	v.send(Frame{Timestamp: time.Now(), JPEG: frame})
+}
+
 // GrabFrame captures a single frame from the RTSP source using ffmpeg and returns it as a JPEG-encoded byte slice.
 func GrabFrame(ctx context.Context, cfg VideoRTSPStreamConfig) ([]byte, error) {
 	v := NewVideoRTSPStream(cfg)
diff --git a/internal/providers/vlm/video_rtsp_stream_test.go b/internal/providers/vlm/video_rtsp_stream_test.go
index 1e4d5080ee..5afa99c59c 100644
--- a/internal/providers/vlm/video_rtsp_stream_test.go
+++ b/internal/providers/vlm/video_rtsp_stream_test.go
@@ -2,6 +2,7 @@ package vlm
 
 import (
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/require"
 )
@@ -23,3 +24,24 @@ func TestNewVideoRTSPStreamPreservesExplicitURL(t *testing.T) {
 	v := NewVideoRTSPStream(VideoRTSPStreamConfig{RTSPURL: "rtsp://example.test/live"})
 	require.Equal(t, "rtsp://example.test/live", v.cfg.RTSPURL)
 }
+
+// TestEmitFrameStampsReceiveTime verifies each decoded frame is forwarded with a
+// receive-time stamp and its JPEG payload intact.
+func TestEmitFrameStampsReceiveTime(t *testing.T) {
+	v := NewVideoRTSPStream(VideoRTSPStreamConfig{})
+	v.out = make(chan Frame, 1) // stand in for the channel start() would create
+
+	jpeg := []byte{0xFF, 0xD8, 0xFF, 0xD9}
+	before := time.Now()
+	v.emitFrame(jpeg)
+	after := time.Now()
+
+	select {
+	case f := <-v.out:
+		require.Equal(t, jpeg, f.JPEG)
+		require.False(t, f.Timestamp.Before(before), "timestamp must not precede the call")
+		require.False(t, f.Timestamp.After(after), "timestamp must not follow the call")
+	default:
+		t.Fatal("expected emitFrame to enqueue a frame")
+	}
+}
diff --git a/plugins/inputs/asr/asr_common.go b/plugins/inputs/asr/asr_common.go
index 03721fc620..53580d6356 100644
--- a/plugins/inputs/asr/asr_common.go
+++ b/plugins/inputs/asr/asr_common.go
@@ -291,6 +291,20 @@ func (c *asrCommon) sendChunkAt(pcm []byte, capture time.Time) {
 	c.stream.sendChunkAt(pcm, capture)
 }
 
+// forwardChunk copies a freshly-read PCM chunk and sends it stamped at its
+// capture time. The chunk is dropped (and false returned) when TTS is currently
+// speaking and interruption is disabled. The copy is required because callers
+// reuse their read buffer for the next chunk.
+func (c *asrCommon) forwardChunk(buf []byte, capture time.Time) bool {
+	if tts.Speaking.Load() && !c.enableTTSInterrupt {
+		return false
+	}
+	pcm := make([]byte, len(buf))
+	copy(pcm, buf)
+	c.sendChunkAt(pcm, capture)
+	return true
+}
+
 // statsLoop logs the single stream's send statistics until ctx is cancelled.
 func (c *asrCommon) statsLoop(ctx context.Context) { c.stream.statsLoop(ctx) }
 
diff --git a/plugins/inputs/asr/capture_time_test.go b/plugins/inputs/asr/capture_time_test.go
index 15bacfb842..7e1e1536bf 100644
--- a/plugins/inputs/asr/capture_time_test.go
+++ b/plugins/inputs/asr/capture_time_test.go
@@ -7,6 +7,7 @@ import (
 	"github.com/stretchr/testify/require"
 	"go.uber.org/zap"
 
+	"github.com/openmind/om1/internal/providers/tts"
 	"github.com/openmind/om1/internal/ws"
 )
 
@@ -72,6 +73,54 @@ func TestASRCommonSendChunkAtDelegates(t *testing.T) {
 		"asrCommon.sendChunkAt must forward to the underlying stream")
 }
 
+func TestForwardChunkSendsWhenNotSpeaking(t *testing.T) {
+	s := newSendableStream(t)
+	c := &asrCommon{asrSensorCore: newTestSensorCore(), stream: s}
+	// Ensure TTS is not "speaking" for this case.
+	tts.Speaking.Store(false)
+
+	sent := c.forwardChunk([]byte{0x01, 0x02, 0x03, 0x04}, time.UnixMilli(1_700_000_000_000))
+
+	require.True(t, sent, "chunk must be forwarded when TTS is silent")
+	s.stats.mu.RLock()
+	defer s.stats.mu.RUnlock()
+	require.Equal(t, uint64(1), s.stats.TotalChunksSent)
+}
+
+func TestForwardChunkDropsWhileSpeakingWithoutInterrupt(t *testing.T) {
+	s := newSendableStream(t)
+	core := newTestSensorCore()
+	core.enableTTSInterrupt = false
+	c := &asrCommon{asrSensorCore: core, stream: s}
+
+	tts.Speaking.Store(true)
+	defer tts.Speaking.Store(false)
+
+	sent := c.forwardChunk([]byte{0x01, 0x02}, time.Now())
+
+	require.False(t, sent, "chunk must be dropped while TTS speaks and interrupt is disabled")
+	s.stats.mu.RLock()
+	defer s.stats.mu.RUnlock()
+	require.Zero(t, s.stats.TotalChunksSent)
+}
+
+func TestForwardChunkSendsWhileSpeakingWithInterrupt(t *testing.T) {
+	s := newSendableStream(t)
+	core := newTestSensorCore()
+	core.enableTTSInterrupt = true
+	c := &asrCommon{asrSensorCore: core, stream: s}
+
+	tts.Speaking.Store(true)
+	defer tts.Speaking.Store(false)
+
+	sent := c.forwardChunk([]byte{0x01, 0x02}, time.Now())
+
+	require.True(t, sent, "interrupt-enabled sensors keep streaming during TTS")
+	s.stats.mu.RLock()
+	defer s.stats.mu.RUnlock()
+	require.Equal(t, uint64(1), s.stats.TotalChunksSent)
+}
+
 // TestASRRTSPDefaultURLs pins the RTSP audio source defaults to the
 // video-processor's muxed session stream (:8555/live) and verifies an explicit
 // URL is preserved.
diff --git a/plugins/inputs/asr/elevenlabs_asr_rtsp.go b/plugins/inputs/asr/elevenlabs_asr_rtsp.go
index 3b5d1ac033..55f39922b1 100644
--- a/plugins/inputs/asr/elevenlabs_asr_rtsp.go
+++ b/plugins/inputs/asr/elevenlabs_asr_rtsp.go
@@ -12,7 +12,6 @@ import (
 	"go.uber.org/zap"
 
 	"github.com/openmind/om1/internal/inputs"
-	"github.com/openmind/om1/internal/providers/tts"
 	"github.com/openmind/om1/internal/util"
 )
 
@@ -183,15 +182,7 @@ func (s *ElevenLabsASRRTSPSensor) streamRTSP(ctx context.Context) error {
 		if _, err := io.ReadFull(stdout, buf); err != nil {
 			return fmt.Errorf("read pcm: %w", err)
 		}
-		// Stamp capture time when the chunk is read from the stream.
-		tCapture := time.Now()
-
-		if tts.Speaking.Load() && !s.cfg.EnableTTSInterrupt {
-			continue
-		}
-
-		pcm := make([]byte, chunkBytes)
-		copy(pcm, buf)
-		s.sendChunkAt(pcm, tCapture)
+		// Stamp capture time at read (see asrCommon.forwardChunk).
+		s.forwardChunk(buf, time.Now())
 	}
 }
diff --git a/plugins/inputs/asr/google_asr_rtsp.go b/plugins/inputs/asr/google_asr_rtsp.go
index 3ca57dcc12..8d4130a950 100644
--- a/plugins/inputs/asr/google_asr_rtsp.go
+++ b/plugins/inputs/asr/google_asr_rtsp.go
@@ -12,7 +12,6 @@ import (
 	"go.uber.org/zap"
 
 	"github.com/openmind/om1/internal/inputs"
-	"github.com/openmind/om1/internal/providers/tts"
 	"github.com/openmind/om1/internal/util"
 )
 
@@ -191,16 +190,8 @@ func (s *GoogleASRRTSPSensor) streamRTSP(ctx context.Context) error {
 		if _, err := io.ReadFull(stdout, buf); err != nil {
 			return fmt.Errorf("read pcm: %w", err)
 		}
-		// Stamp capture time when the chunk is read from the stream, so the ASR
+		// Stamp capture time at read (see asrCommon.forwardChunk), so the ASR
 		// chunk carries the capture instant rather than the later send time.
-		tCapture := time.Now()
-
-		if tts.Speaking.Load() && !s.cfg.EnableTTSInterrupt {
-			continue
-		}
-
-		pcm := make([]byte, chunkBytes)
-		copy(pcm, buf)
-		s.sendChunkAt(pcm, tCapture)
+		s.forwardChunk(buf, time.Now())
 	}
 }
diff --git a/plugins/inputs/asr/riva_asr_rtsp.go b/plugins/inputs/asr/riva_asr_rtsp.go
index 0df607e478..abea8d6be2 100644
--- a/plugins/inputs/asr/riva_asr_rtsp.go
+++ b/plugins/inputs/asr/riva_asr_rtsp.go
@@ -12,7 +12,6 @@ import (
 	"go.uber.org/zap"
 
 	"github.com/openmind/om1/internal/inputs"
-	"github.com/openmind/om1/internal/providers/tts"
 	"github.com/openmind/om1/internal/util"
 )
 
@@ -179,15 +178,7 @@ func (s *RivaASRRTSPSensor) streamRTSP(ctx context.Context) error {
 		if _, err := io.ReadFull(stdout, buf); err != nil {
 			return fmt.Errorf("read pcm: %w", err)
 		}
-		// Stamp capture time when the chunk is read from the stream.
-		tCapture := time.Now()
-
-		if tts.Speaking.Load() && !s.cfg.EnableTTSInterrupt {
-			continue
-		}
-
-		pcm := make([]byte, chunkBytes)
-		copy(pcm, buf)
-		s.sendChunkAt(pcm, tCapture)
+		// Stamp capture time at read (see asrCommon.forwardChunk).
+		s.forwardChunk(buf, time.Now())
 	}
 }

From 7c7e619fd7ba12a3f5edecc90b5f17615cebca2d Mon Sep 17 00:00:00 2001
From: openmindev <147775420+openminddev@users.noreply.github.com>
Date: Fri, 28 Aug 2026 13:14:32 -0700
Subject: [PATCH 11/15] Update asr_stream.go

---
 plugins/inputs/asr/asr_stream.go | 6 ------
 1 file changed, 6 deletions(-)

diff --git a/plugins/inputs/asr/asr_stream.go b/plugins/inputs/asr/asr_stream.go
index e76a9d0d5c..82ff4f68ab 100644
--- a/plugins/inputs/asr/asr_stream.go
+++ b/plugins/inputs/asr/asr_stream.go
@@ -112,12 +112,6 @@ func (s *transcriberStream) packageAudio(pcm []byte, captureMs int64) ([]byte, e
 	return packet, nil
 }
 
-// sendChunk packages and sends a PCM chunk, stamping the capture time as now.
-// Callers that know the true acoustic capture instant should use sendChunkAt.
-// func (s *transcriberStream) sendChunk(pcm []byte) {
-// 	s.sendChunkAt(pcm, time.Now())
-// }
-
 // sendChunkAt packages and sends a PCM chunk captured at the given time,
 // updating statistics.
 func (s *transcriberStream) sendChunkAt(pcm []byte, capture time.Time) {

From b37721e9bede21ee7dbc37479f915db03a04a6b6 Mon Sep 17 00:00:00 2001
From: openmindev <147775420+openminddev@users.noreply.github.com>
Date: Fri, 28 Aug 2026 13:36:18 -0700
Subject: [PATCH 12/15] Clean code

---
 PR_DESCRIPTION_GST_CONSUMER.md                |  63 --------
 config/unitree_g1_conversation.json5          |   9 --
 .../developing/video_processor_integration.md |  97 -------------
 internal/providers/vlm/video_rtsp_stream.go   |   8 --
 .../providers/vlm/video_rtsp_stream_test.go   |  10 +-
 plugins/backgrounds/vlm/vlm.go                |   3 +-
 plugins/inputs/asr/asr_common.go              |  22 +--
 plugins/inputs/asr/capture_time_test.go       |  57 --------
 plugins/inputs/asr/elevenlabs_asr.go          |   1 -
 plugins/inputs/asr/elevenlabs_asr_rtsp.go     |  13 +-
 plugins/inputs/asr/google_asr_rtsp.go         |  15 +-
 plugins/inputs/asr/parallel_asr.go            |   6 +-
 plugins/inputs/asr/riva_asr.go                |   1 -
 plugins/inputs/asr/riva_asr_rtsp.go           |  13 +-
 plugins/inputs/vlm/vlm.go                     |   2 -
 scripts/bootstrap-go.sh                       | 135 ------------------
 16 files changed, 38 insertions(+), 417 deletions(-)
 delete mode 100644 PR_DESCRIPTION_GST_CONSUMER.md
 delete mode 100644 docs/developing/video_processor_integration.md
 delete mode 100755 scripts/bootstrap-go.sh

diff --git a/PR_DESCRIPTION_GST_CONSUMER.md b/PR_DESCRIPTION_GST_CONSUMER.md
deleted file mode 100644
index 170e79057a..0000000000
--- a/PR_DESCRIPTION_GST_CONSUMER.md
+++ /dev/null
@@ -1,63 +0,0 @@
-# Consume the synchronized A+V stream + stamp ASR audio at capture
-
-Pairs OM1 with the OM1-video-processor GStreamer pipeline, which serves a single
-muxed A+V RTSP session (`rtsp://localhost:8555/live`) on one clock. Two parts:
-config (repoint consumers) and code (stamp audio capture time at ingest).
-
-## Config — repoint consumers (non-breaking)
-
-`config/unitree_g1_conversation.json5` is env-parametrized; **defaults preserve
-current behavior exactly**:
-
-- ASR: `type: "${ASR_INPUT_PLUGIN:-GoogleASRInput}"` + `rtsp_url:
-  "${ASR_RTSP_URL:-rtsp://localhost:8555/live}"` (the local plugin ignores the
-  extra key).
-- VLM: `rtsp_url: "${VLM_RTSP_URL:-rtsp://localhost:8554/top_camera_raw}"`
-  (unchanged default).
-
-`docker-compose.yml` exposes `ASR_INPUT_PLUGIN`, `ASR_RTSP_URL`, `VLM_RTSP_URL`.
-To consume the GStreamer stream:
-
-```bash
-ASR_INPUT_PLUGIN=GoogleASRRTSPInput \
-ASR_RTSP_URL=rtsp://localhost:8555/live \
-VLM_RTSP_URL=rtsp://localhost:8555/live \
-docker compose up -d om1
-```
-
-Both consumers read the same muxed URL; each ffmpeg selects its track
-(`google_asr_rtsp` uses `-vn`, `video_rtsp_stream` uses `-an`), so audio and
-video come from one synchronized source.
-
-## Code — stamp ASR audio at capture
-
-Previously the audio timestamp was set at *package/send* time
-(`packageAudio` → `time.Now().UnixMilli()`), discarding capture timing. Now:
-
-- `packageAudio(pcm, captureMs)` stamps the provided capture time.
-- `sendChunkAt(pcm, capture)` added; `sendChunk` kept as a `time.Now()` wrapper
-  for back-compat (riva/elevenlabs/parallel unchanged — trivially extendable).
-- `google_asr.go` (local mic) and `google_asr_rtsp.go` (RTSP) stamp
-  `time.Now()` at the moment the buffer/chunk is read and pass it through.
-- Tests updated to pass and assert the capture timestamp.
-
-This makes ASR chunks carry capture time, so downstream alignment with
-video-derived features works on a common timeline.
-
-## Deliberately out of scope (follow-ups)
-
-- **Transcript→capture-time mapping across the cloud round-trip.** The final
-  transcript still reaches the IO layer without a capture timestamp
-  (`asr_common.go` `AddInput(..., time.Time{})`); mapping it back to the source
-  audio window needs the ASR WS protocol to echo timestamps.
-- **Video RTSP PTS.** JPEG-over-`image2pipe` carries no per-frame PTS; true
-  capture-time for video needs a PTS-preserving transport (documented at the
-  stamp site in `video_rtsp_stream.go`).
-- Extend `sendChunkAt` to the riva/elevenlabs paths (one-line each).
-
-## Verification
-
-Config validated as JSON5; compose validated as YAML. The Go changes are
-mechanical (caller consistency checked: `sendChunk` still present; only the two
-`packageAudio` test call sites updated) but were **not** compiled in this
-environment — run `go build ./... && go test ./plugins/inputs/asr/...` in CI.
diff --git a/config/unitree_g1_conversation.json5 b/config/unitree_g1_conversation.json5
index d9907b82c6..ac4fd75d0b 100644
--- a/config/unitree_g1_conversation.json5
+++ b/config/unitree_g1_conversation.json5
@@ -80,11 +80,6 @@ You should prioritize safe, comfortable, and positive human interaction.",
       hertz: 0.001,
       agent_inputs: [
         {
-          // Defaults preserve legacy behavior (local PortAudio mic). To consume
-          // the synchronized A+V stream from the GStreamer video-processor, set
-          // ASR_INPUT_PLUGIN=GoogleASRRTSPInput and
-          // ASR_RTSP_URL=rtsp://localhost:8555/live. GoogleASRInput ignores the
-          // extra rtsp_url key, so the default is harmless.
           type: "${ASR_INPUT_PLUGIN:-GoogleASRInput}",
           config: {
             api_version: "v2",
@@ -95,10 +90,6 @@ You should prioritize safe, comfortable, and positive human interaction.",
           },
         },
         {
-          // VLM consumes the video-processor's RAW (pre-CV, no overlays/blur)
-          // camera view directly — the clean scene gives the best descriptions.
-          // Override with VLM_RTSP_URL (e.g. rtsp://localhost:8555/live for the
-          // processed/blurred view).
           type: "VLMGeminiRTSP",
           config: {
             rtsp_url: "${VLM_RTSP_URL:-rtsp://localhost:8556/raw}",
diff --git a/docs/developing/video_processor_integration.md b/docs/developing/video_processor_integration.md
deleted file mode 100644
index 244a133f49..0000000000
--- a/docs/developing/video_processor_integration.md
+++ /dev/null
@@ -1,97 +0,0 @@
-# Video Processor Integration (audio & video sources)
-
-On a robot (e.g. NVIDIA Thor), OM1 runs alongside the **OM1 Video Processor**, a
-separate container that owns the camera and microphone. This page explains how
-OM1 consumes that data, and — importantly — the trade-offs behind the defaults
-so you can change them deliberately later.
-
-## Topology
-
-The video-processor is the capture authority. Its GStreamer pipeline captures
-camera + mic on one clock and serves, on the robot host:
-
-| Endpoint | Contents | Who it's for |
-| --- | --- | --- |
-| `rtsp://localhost:8555/live` | Processed video (recognition boxes, blur) **+ audio**, synchronized | machine consumers (OM1, cloud relay) — low latency |
-| `rtsp://localhost:8556/raw` | **Raw** camera view (no overlays/blur), video only | machine consumers wanting a clean image |
-| `mediamtx` `:8554` (RTSP) / `:8888` (HLS) / `:8889` (WebRTC) | Re-serves `/live` and `/raw` | people / external clients / browsers |
-
-**OM1 consumes the gst endpoints directly (`:8555`/`:8556`), not mediamtx.**
-mediamtx is an on-demand fan-out hub for humans; routing the agent's real-time
-inputs through it would add a relay hop and couple OM1's core perception to an
-optional convenience service. Keep the agent on the direct, low-latency source;
-leave mediamtx for people.
-
-## How OM1 consumes it
-
-Set via env (see `docker-compose.yml` and `config/unitree_g1_conversation.json5`):
-
-| Variable | Default | Meaning |
-| --- | --- | --- |
-| `ASR_INPUT_PLUGIN` | `GoogleASRInput` | ASR source plugin (local mic vs RTSP) |
-| `ASR_RTSP_URL` | `rtsp://localhost:8555/live` | Audio source when using the RTSP ASR plugin (ffmpeg selects the audio track with `-vn`) |
-| `VLM_RTSP_URL` | `rtsp://localhost:8556/raw` | Video source for the VLM |
-
-### VLM video source — why `/raw`
-
-The VLM describes the scene for the LLM. `/live` has recognition boxes and
-**blurred faces** burned into the pixels; feeding that to the VLM degrades its
-descriptions. `/raw` is the clean camera view, so it gives the best scene
-understanding — hence the default.
-
-Trade-off to know: `/raw` is **not** anonymized. If your deployment must keep
-faces blurred even in what's sent to the (cloud) VLM for privacy reasons, set
-`VLM_RTSP_URL=rtsp://localhost:8555/live` instead and accept the description
-quality hit.
-
-### ASR source — why local mic is the default
-
-ASR is env-selectable between the local mic (`GoogleASRInput`, opens the mic via
-PortAudio) and RTSP (`GoogleASRRTSPInput`, pulls audio from `:8555/live`). The
-default is the **local mic**. Reasoning:
-
-- **Standalone-safe.** OM1 runs without the video-processor (dev laptops, other
-  robots, `conversation.json5`). A local-mic default doesn't break when there's
-  no `:8555`.
-- **Lower latency & decoupled.** Direct PortAudio capture avoids the RTSP +
-  ffmpeg decode hop, and OM1's hearing doesn't depend on the video-processor
-  being up.
-- **The double mic capture is harmless** (see below).
-
-Switch to RTSP (`ASR_INPUT_PLUGIN=GoogleASRRTSPInput`, `ASR_RTSP_URL=
-rtsp://localhost:8555/live`) when you specifically want a **single audio
-authority** — one capture, one AEC path, and OM1's transcripts aligned to the
-*exact* audio that the video-processor records and streams to the cloud. The
-cost is coupling OM1's ASR to the video-processor (startup ordering; goes deaf
-if `:8555` is down) and a small latency hop.
-
-Rule of thumb: timing-critical / sample-consistent audio work belongs in the
-video-processor (which already has the single capture-stamped audio for
-recording, features, and cloud). OM1's own job — hear the user, respond fast —
-is best served by the decoupled local mic. Flip to RTSP only when a concrete
-need for shared/consistent audio appears.
-
-## Can OM1 and the video-processor both use the microphone at once?
-
-**Yes — because access goes through PulseAudio.** `default_mic_aec` is a
-PulseAudio (echo-cancelled) *virtual source*, and PulseAudio sources are not
-exclusive: it duplicates the stream to every client. Both containers mount the
-host Pulse socket (`PULSE_SERVER=unix:$XDG_RUNTIME_DIR/pulse/native`), so OM1's
-PortAudio→Pulse capture and the video-processor's `pulsesrc`/ffmpeg capture read
-the same source concurrently without conflict.
-
-Caveat: this only holds via PulseAudio. If either side opened the **raw ALSA**
-device (`hw:X`) directly, ALSA hardware devices are exclusive and the second
-opener would fail with "device busy" (unless using `dmix`/`dsnoop`). Everything
-here is Pulse-based, so concurrent access is fine.
-
-## Video timestamps in OM1
-
-OM1's RTSP video consumer stamps each frame with local **receive time**
-(`time.Now()`), not the original capture time — the JPEG-over-pipe transport
-carries no per-frame PTS. This is intentional: OM1 fuses VLM output as coarse
-"recent context", so sub-second pipeline jitter is irrelevant, and any
-frame-accurate A/V work lives in the video-processor (capture-stamped at the
-source). If OM1 ever needs true capture time, switch the consumer to an RTSP
-client that exposes RTP/RTCP timing (e.g. `gortsplib`) instead of the
-ffmpeg→JPEG pipe. See `internal/providers/vlm/video_rtsp_stream.go`.
diff --git a/internal/providers/vlm/video_rtsp_stream.go b/internal/providers/vlm/video_rtsp_stream.go
index aedb911ffe..739ecc5a79 100644
--- a/internal/providers/vlm/video_rtsp_stream.go
+++ b/internal/providers/vlm/video_rtsp_stream.go
@@ -130,14 +130,6 @@ func (v *VideoRTSPStream) stream(ctx context.Context) error {
 }
 
 // emitFrame forwards a decoded JPEG stamped with its receive time.
-//
-// Receive-time stamp is by design: this is when OM1 got the decoded frame, not
-// the original capture instant — JPEG-over-image2pipe carries no per-frame RTP
-// PTS. That's fine here: OM1 fuses VLM output as coarse "recent context", so
-// sub-second pipeline jitter doesn't matter. Any timing-critical, frame-accurate
-// A/V work lives in the video-processor (capture-stamped at the source). If OM1
-// ever needs true capture time, switch this consumer to an RTSP client that
-// exposes RTP/RTCP timing (e.g. gortsplib) rather than the ffmpeg->JPEG pipe.
 func (v *VideoRTSPStream) emitFrame(frame []byte) {
 	v.send(Frame{Timestamp: time.Now(), JPEG: frame})
 }
diff --git a/internal/providers/vlm/video_rtsp_stream_test.go b/internal/providers/vlm/video_rtsp_stream_test.go
index 5afa99c59c..ff39e88a02 100644
--- a/internal/providers/vlm/video_rtsp_stream_test.go
+++ b/internal/providers/vlm/video_rtsp_stream_test.go
@@ -7,10 +7,6 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-// TestVideoRTSPDefaultURLValue pins the actual default endpoint string to the
-// video-processor's clean (pre-CV) camera view. The sibling
-// TestNewVideoRTSPStreamDefaults asserts against the defaultRTSPURL constant, so
-// this guards the constant's value itself against accidental changes.
 func TestVideoRTSPDefaultURLValue(t *testing.T) {
 	require.Equal(t, "rtsp://localhost:8556/raw", defaultRTSPURL)
 
@@ -18,18 +14,14 @@ func TestVideoRTSPDefaultURLValue(t *testing.T) {
 	require.Equal(t, "rtsp://localhost:8556/raw", v.cfg.RTSPURL)
 }
 
-// TestNewVideoRTSPStreamPreservesExplicitURL covers the non-default branch: a
-// caller-supplied URL must not be overwritten.
 func TestNewVideoRTSPStreamPreservesExplicitURL(t *testing.T) {
 	v := NewVideoRTSPStream(VideoRTSPStreamConfig{RTSPURL: "rtsp://example.test/live"})
 	require.Equal(t, "rtsp://example.test/live", v.cfg.RTSPURL)
 }
 
-// TestEmitFrameStampsReceiveTime verifies each decoded frame is forwarded with a
-// receive-time stamp and its JPEG payload intact.
 func TestEmitFrameStampsReceiveTime(t *testing.T) {
 	v := NewVideoRTSPStream(VideoRTSPStreamConfig{})
-	v.out = make(chan Frame, 1) // stand in for the channel start() would create
+	v.out = make(chan Frame, 1)
 
 	jpeg := []byte{0xFF, 0xD8, 0xFF, 0xD9}
 	before := time.Now()
diff --git a/plugins/backgrounds/vlm/vlm.go b/plugins/backgrounds/vlm/vlm.go
index 9ba7568836..ca0155fe82 100644
--- a/plugins/backgrounds/vlm/vlm.go
+++ b/plugins/backgrounds/vlm/vlm.go
@@ -23,8 +23,7 @@ func init() {
 }
 
 const (
-	defaultFPS = 10
-	// GStreamer video-processor's raw (pre-CV) camera view; see inputs/vlm.
+	defaultFPS     = 10
 	defaultRTSPURL = "rtsp://localhost:8556/raw"
 
 	vlmRestartDelay = 2 * time.Second
diff --git a/plugins/inputs/asr/asr_common.go b/plugins/inputs/asr/asr_common.go
index cbbe924f65..ed0b21c252 100644
--- a/plugins/inputs/asr/asr_common.go
+++ b/plugins/inputs/asr/asr_common.go
@@ -15,7 +15,6 @@ import (
 	"github.com/openmind/om1/internal/inputs"
 	"github.com/openmind/om1/internal/logger"
 	"github.com/openmind/om1/internal/providers"
-	"github.com/openmind/om1/internal/providers/tts"
 	zenohsession "github.com/openmind/om1/internal/zenoh"
 )
 
@@ -75,8 +74,6 @@ type asrSensorCore struct {
 	language   string
 	apiVersion string
 
-	enableTTSInterrupt bool
-
 	transcriptCh chan string
 
 	messages []string
@@ -101,10 +98,7 @@ func newSensorCore(name string, enableTTSInterrupt bool, vadCfg vadLatencyConfig
 		language:     language,
 		apiVersion:   apiVersion,
 		transcriptCh: make(chan string, 32),
-
-		enableTTSInterrupt: enableTTSInterrupt,
-
-		vad: newVADLatencyTracker(vadCfg, enableTTSInterrupt, rate, log),
+		vad:          newVADLatencyTracker(vadCfg, enableTTSInterrupt, rate, log),
 	}
 
 	sess, err := zenohsession.Open()
@@ -306,20 +300,6 @@ func (c *asrCommon) sendChunkAt(pcm []byte, capture time.Time) {
 	c.stream.sendChunkAt(pcm, capture)
 }
 
-// forwardChunk copies a freshly-read PCM chunk and sends it stamped at its
-// capture time. The chunk is dropped (and false returned) when TTS is currently
-// speaking and interruption is disabled. The copy is required because callers
-// reuse their read buffer for the next chunk.
-func (c *asrCommon) forwardChunk(buf []byte, capture time.Time) bool {
-	if tts.Speaking.Load() && !c.enableTTSInterrupt {
-		return false
-	}
-	pcm := make([]byte, len(buf))
-	copy(pcm, buf)
-	c.sendChunkAt(pcm, capture)
-	return true
-}
-
 // statsLoop logs the single stream's send statistics until ctx is cancelled.
 func (c *asrCommon) statsLoop(ctx context.Context) { c.stream.statsLoop(ctx) }
 
diff --git a/plugins/inputs/asr/capture_time_test.go b/plugins/inputs/asr/capture_time_test.go
index 7e1e1536bf..5ea71a4938 100644
--- a/plugins/inputs/asr/capture_time_test.go
+++ b/plugins/inputs/asr/capture_time_test.go
@@ -7,14 +7,9 @@ import (
 	"github.com/stretchr/testify/require"
 	"go.uber.org/zap"
 
-	"github.com/openmind/om1/internal/providers/tts"
 	"github.com/openmind/om1/internal/ws"
 )
 
-// newSendableStream builds a transcriberStream with a websocket client whose
-// send buffer is large enough that Send() never blocks or fills. The client is
-// never Connect()-ed, so no goroutines start and Send simply enqueues onto the
-// buffered channel — enough to exercise packaging + statistics without a server.
 func newSendableStream(t *testing.T) *transcriberStream {
 	t.Helper()
 	s := newTestElevenLabsStream(make(chan string, 1))
@@ -45,7 +40,6 @@ func TestSendChunkAtSuccess(t *testing.T) {
 
 func TestSendChunkAtSendError(t *testing.T) {
 	s := newTestElevenLabsStream(make(chan string, 1))
-	// A single-slot buffer that we pre-fill, so the next Send fails.
 	s.wsClient = ws.New(
 		ws.Config{URL: "ws://127.0.0.1:0", SendBufferSize: 1},
 		zap.NewNop(),
@@ -73,57 +67,6 @@ func TestASRCommonSendChunkAtDelegates(t *testing.T) {
 		"asrCommon.sendChunkAt must forward to the underlying stream")
 }
 
-func TestForwardChunkSendsWhenNotSpeaking(t *testing.T) {
-	s := newSendableStream(t)
-	c := &asrCommon{asrSensorCore: newTestSensorCore(), stream: s}
-	// Ensure TTS is not "speaking" for this case.
-	tts.Speaking.Store(false)
-
-	sent := c.forwardChunk([]byte{0x01, 0x02, 0x03, 0x04}, time.UnixMilli(1_700_000_000_000))
-
-	require.True(t, sent, "chunk must be forwarded when TTS is silent")
-	s.stats.mu.RLock()
-	defer s.stats.mu.RUnlock()
-	require.Equal(t, uint64(1), s.stats.TotalChunksSent)
-}
-
-func TestForwardChunkDropsWhileSpeakingWithoutInterrupt(t *testing.T) {
-	s := newSendableStream(t)
-	core := newTestSensorCore()
-	core.enableTTSInterrupt = false
-	c := &asrCommon{asrSensorCore: core, stream: s}
-
-	tts.Speaking.Store(true)
-	defer tts.Speaking.Store(false)
-
-	sent := c.forwardChunk([]byte{0x01, 0x02}, time.Now())
-
-	require.False(t, sent, "chunk must be dropped while TTS speaks and interrupt is disabled")
-	s.stats.mu.RLock()
-	defer s.stats.mu.RUnlock()
-	require.Zero(t, s.stats.TotalChunksSent)
-}
-
-func TestForwardChunkSendsWhileSpeakingWithInterrupt(t *testing.T) {
-	s := newSendableStream(t)
-	core := newTestSensorCore()
-	core.enableTTSInterrupt = true
-	c := &asrCommon{asrSensorCore: core, stream: s}
-
-	tts.Speaking.Store(true)
-	defer tts.Speaking.Store(false)
-
-	sent := c.forwardChunk([]byte{0x01, 0x02}, time.Now())
-
-	require.True(t, sent, "interrupt-enabled sensors keep streaming during TTS")
-	s.stats.mu.RLock()
-	defer s.stats.mu.RUnlock()
-	require.Equal(t, uint64(1), s.stats.TotalChunksSent)
-}
-
-// TestASRRTSPDefaultURLs pins the RTSP audio source defaults to the
-// video-processor's muxed session stream (:8555/live) and verifies an explicit
-// URL is preserved.
 func TestASRRTSPDefaultURLs(t *testing.T) {
 	const wantDefault = "rtsp://localhost:8555/live"
 
diff --git a/plugins/inputs/asr/elevenlabs_asr.go b/plugins/inputs/asr/elevenlabs_asr.go
index be9c74d3f5..df598d4e08 100644
--- a/plugins/inputs/asr/elevenlabs_asr.go
+++ b/plugins/inputs/asr/elevenlabs_asr.go
@@ -244,7 +244,6 @@ func (s *ElevenLabsASRSensor) captureLoop(ctx context.Context, stream *portaudio
 		if err := stream.Read(); err != nil && err.Error() != "Input overflowed" {
 			s.log.Warn("read error", zap.Error(err))
 		}
-		// Stamp capture time right after the buffer is read.
 		tCapture := time.Now()
 
 		if tts.Speaking.Load() && !s.cfg.EnableTTSInterrupt {
diff --git a/plugins/inputs/asr/elevenlabs_asr_rtsp.go b/plugins/inputs/asr/elevenlabs_asr_rtsp.go
index 165f3f1510..5862a8fa5a 100644
--- a/plugins/inputs/asr/elevenlabs_asr_rtsp.go
+++ b/plugins/inputs/asr/elevenlabs_asr_rtsp.go
@@ -12,6 +12,7 @@ import (
 	"go.uber.org/zap"
 
 	"github.com/openmind/om1/internal/inputs"
+	"github.com/openmind/om1/internal/providers/tts"
 	"github.com/openmind/om1/internal/util"
 )
 
@@ -186,7 +187,15 @@ func (s *ElevenLabsASRRTSPSensor) streamRTSP(ctx context.Context) error {
 		if _, err := io.ReadFull(stdout, buf); err != nil {
 			return fmt.Errorf("read pcm: %w", err)
 		}
-		// Stamp capture time at read (see asrCommon.forwardChunk).
-		s.forwardChunk(buf, time.Now())
+		// Stamp capture time at read.
+		tCapture := time.Now()
+
+		if tts.Speaking.Load() && !s.cfg.EnableTTSInterrupt {
+			continue
+		}
+
+		pcm := make([]byte, chunkBytes)
+		copy(pcm, buf)
+		s.sendChunkAt(pcm, tCapture)
 	}
 }
diff --git a/plugins/inputs/asr/google_asr_rtsp.go b/plugins/inputs/asr/google_asr_rtsp.go
index bd40114889..7f71db8e51 100644
--- a/plugins/inputs/asr/google_asr_rtsp.go
+++ b/plugins/inputs/asr/google_asr_rtsp.go
@@ -12,6 +12,7 @@ import (
 	"go.uber.org/zap"
 
 	"github.com/openmind/om1/internal/inputs"
+	"github.com/openmind/om1/internal/providers/tts"
 	"github.com/openmind/om1/internal/util"
 )
 
@@ -194,8 +195,16 @@ func (s *GoogleASRRTSPSensor) streamRTSP(ctx context.Context) error {
 		if _, err := io.ReadFull(stdout, buf); err != nil {
 			return fmt.Errorf("read pcm: %w", err)
 		}
-		// Stamp capture time at read (see asrCommon.forwardChunk), so the ASR
-		// chunk carries the capture instant rather than the later send time.
-		s.forwardChunk(buf, time.Now())
+		// Stamp capture time at read, so the ASR chunk carries the capture
+		// instant rather than the later send time.
+		tCapture := time.Now()
+
+		if tts.Speaking.Load() && !s.cfg.EnableTTSInterrupt {
+			continue
+		}
+
+		pcm := make([]byte, chunkBytes)
+		copy(pcm, buf)
+		s.sendChunkAt(pcm, tCapture)
 	}
 }
diff --git a/plugins/inputs/asr/parallel_asr.go b/plugins/inputs/asr/parallel_asr.go
index fcf3bfb46d..d5cbaa3a87 100644
--- a/plugins/inputs/asr/parallel_asr.go
+++ b/plugins/inputs/asr/parallel_asr.go
@@ -292,9 +292,7 @@ func (s *ParallelASRSensor) connectStreams() {
 	wg.Wait()
 }
 
-// sendToAllAt fans one PCM chunk (captured at the given time) out to the VAD and
-// every provider stream. Each stream packages the audio with its own header, so
-// the shared chunk is only read, never mutated.
+// sendToAllAt fans out a PCM chunk to all provider streams, stamped with the capture time.
 func (s *ParallelASRSensor) sendToAllAt(pcm []byte, capture time.Time) {
 	s.feedVAD(pcm)
 	for _, st := range s.streams {
@@ -410,7 +408,6 @@ func (s *ParallelASRSensor) micCaptureLoop(ctx context.Context, stream *portaudi
 		if err := stream.Read(); err != nil && err.Error() != "Input overflowed" {
 			s.log.Warn("read error", zap.Error(err))
 		}
-		// Stamp capture time right after the buffer is read.
 		tCapture := time.Now()
 
 		if tts.Speaking.Load() && !s.cfg.EnableTTSInterrupt {
@@ -493,7 +490,6 @@ func (s *ParallelASRSensor) streamRTSP(ctx context.Context) error {
 		if _, err := io.ReadFull(stdout, buf); err != nil {
 			return fmt.Errorf("read pcm: %w", err)
 		}
-		// Stamp capture time when the chunk is read from the stream.
 		tCapture := time.Now()
 
 		if tts.Speaking.Load() && !s.cfg.EnableTTSInterrupt {
diff --git a/plugins/inputs/asr/riva_asr.go b/plugins/inputs/asr/riva_asr.go
index 70357779f6..4dc95c0306 100644
--- a/plugins/inputs/asr/riva_asr.go
+++ b/plugins/inputs/asr/riva_asr.go
@@ -238,7 +238,6 @@ func (s *RivaASRSensor) captureLoop(ctx context.Context, stream *portaudio.Strea
 		if err := stream.Read(); err != nil && err.Error() != "Input overflowed" {
 			s.log.Warn("read error", zap.Error(err))
 		}
-		// Stamp capture time right after the buffer is read.
 		tCapture := time.Now()
 
 		if tts.Speaking.Load() && !s.cfg.EnableTTSInterrupt {
diff --git a/plugins/inputs/asr/riva_asr_rtsp.go b/plugins/inputs/asr/riva_asr_rtsp.go
index 0c54b3f59c..7c0ff78517 100644
--- a/plugins/inputs/asr/riva_asr_rtsp.go
+++ b/plugins/inputs/asr/riva_asr_rtsp.go
@@ -12,6 +12,7 @@ import (
 	"go.uber.org/zap"
 
 	"github.com/openmind/om1/internal/inputs"
+	"github.com/openmind/om1/internal/providers/tts"
 	"github.com/openmind/om1/internal/util"
 )
 
@@ -182,7 +183,15 @@ func (s *RivaASRRTSPSensor) streamRTSP(ctx context.Context) error {
 		if _, err := io.ReadFull(stdout, buf); err != nil {
 			return fmt.Errorf("read pcm: %w", err)
 		}
-		// Stamp capture time at read (see asrCommon.forwardChunk).
-		s.forwardChunk(buf, time.Now())
+		// Stamp capture time at read.
+		tCapture := time.Now()
+
+		if tts.Speaking.Load() && !s.cfg.EnableTTSInterrupt {
+			continue
+		}
+
+		pcm := make([]byte, chunkBytes)
+		copy(pcm, buf)
+		s.sendChunkAt(pcm, tCapture)
 	}
 }
diff --git a/plugins/inputs/vlm/vlm.go b/plugins/inputs/vlm/vlm.go
index 0e3ae5eddf..22f9d802ac 100644
--- a/plugins/inputs/vlm/vlm.go
+++ b/plugins/inputs/vlm/vlm.go
@@ -25,8 +25,6 @@ const (
 	vlmDescriptor  = "Vision"
 	vlmMaxMessages = 10
 	defaultFPS     = 10
-	// The GStreamer video-processor's raw (pre-CV, no overlays/blur) camera view
-	// — the clean scene for VLM description. Muxed A+V /live is on :8555.
 	defaultRTSPURL = "rtsp://localhost:8556/raw"
 )
 
diff --git a/scripts/bootstrap-go.sh b/scripts/bootstrap-go.sh
deleted file mode 100755
index 2445c66bc2..0000000000
--- a/scripts/bootstrap-go.sh
+++ /dev/null
@@ -1,135 +0,0 @@
-#!/usr/bin/env bash
-#
-# bootstrap-go.sh — install a local Go toolchain without root.
-#
-# Downloads the official Go tarball from https://go.dev/dl and extracts it into
-# a user-writable directory (default: ~/.local/go), so `go` is available on
-# machines/sandboxes where Go isn't preinstalled and apt/root aren't available.
-#
-# The Go version defaults to the `go` directive in ../go.mod so the toolchain
-# matches what the module requires; override with GO_VERSION=x.y.z.
-#
-# Usage:
-#   scripts/bootstrap-go.sh              # install; prints the PATH line to eval
-#   scripts/bootstrap-go.sh --persist    # also append PATH to ~/.bashrc
-#   GO_VERSION=1.25.1 scripts/bootstrap-go.sh
-#   GOROOT_INSTALL=/opt/go scripts/bootstrap-go.sh
-#
-# Note: requires network access to go.dev / dl.google.com. In a restricted
-# sandbox those hosts must be on the egress allowlist first.
-
-set -euo pipefail
-
-SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
-REPO_ROOT="$(cd "${SCRIPT_DIR}/.." && pwd)"
-INSTALL_DIR="${GOROOT_INSTALL:-${HOME}/.local/go}"
-PERSIST=0
-[ "${1:-}" = "--persist" ] && PERSIST=1
-
-log() { printf '>> %s\n' "$*" >&2; }
-die() { printf 'error: %s\n' "$*" >&2; exit 1; }
-
-# --- resolve version ------------------------------------------------------
-resolve_version() {
-  if [ -n "${GO_VERSION:-}" ]; then
-    printf '%s' "${GO_VERSION}"
-    return
-  fi
-  local v
-  v="$(awk '/^go[[:space:]]+[0-9]/ {print $2; exit}' "${REPO_ROOT}/go.mod" 2>/dev/null || true)"
-  # go.mod may list "1.25" (no patch); pad to x.y.0 for the download filename.
-  case "${v}" in
-    *.*.*) : ;;
-    *.*)   v="${v}.0" ;;
-    *)     v="" ;;
-  esac
-  [ -n "${v}" ] || die "could not determine Go version; set GO_VERSION=x.y.z"
-  printf '%s' "${v}"
-}
-
-# --- detect platform ------------------------------------------------------
-detect_os() {
-  case "$(uname -s)" in
-    Linux)  printf 'linux' ;;
-    Darwin) printf 'darwin' ;;
-    *)      die "unsupported OS: $(uname -s)" ;;
-  esac
-}
-detect_arch() {
-  case "$(uname -m)" in
-    x86_64|amd64)  printf 'amd64' ;;
-    aarch64|arm64) printf 'arm64' ;;
-    *)             die "unsupported arch: $(uname -m)" ;;
-  esac
-}
-
-main() {
-  local ver os arch file url tmp sha_expected sha_actual
-  ver="$(resolve_version)"
-  os="$(detect_os)"
-  arch="$(detect_arch)"
-  file="go${ver}.${os}-${arch}.tar.gz"
-  url="https://go.dev/dl/${file}"
-
-  # Already installed and matching? Skip re-download.
-  if [ -x "${INSTALL_DIR}/bin/go" ] && \
-     "${INSTALL_DIR}/bin/go" version 2>/dev/null | grep -q "go${ver} "; then
-    log "Go ${ver} already installed at ${INSTALL_DIR}"
-  else
-    tmp="$(mktemp -d)"
-    trap 'rm -rf "${tmp}"' EXIT
-
-    log "Downloading ${url}"
-    curl -fSL --retry 3 --max-time 300 "${url}" -o "${tmp}/${file}" \
-      || die "download failed (is go.dev on the egress allowlist?)"
-
-    # Best-effort checksum verification from the release manifest.
-    if command -v sha256sum >/dev/null 2>&1; then
-      sha_expected="$(curl -fsSL --max-time 30 \
-        "https://go.dev/dl/?mode=json&include=all" 2>/dev/null \
-        | tr ',{}' '\n' | grep -A2 "\"${file}\"" | grep -o '"sha256":"[0-9a-f]*"' \
-        | head -1 | sed 's/.*:"//;s/"//' || true)"
-      if [ -n "${sha_expected}" ]; then
-        sha_actual="$(sha256sum "${tmp}/${file}" | awk '{print $1}')"
-        [ "${sha_expected}" = "${sha_actual}" ] \
-          || die "checksum mismatch for ${file}"
-        log "checksum verified"
-      else
-        log "WARNING: could not fetch checksum; skipping verification"
-      fi
-    fi
-
-    file "${tmp}/${file}" | grep -qi gzip || die "downloaded file is not a gzip archive"
-
-    log "Installing to ${INSTALL_DIR}"
-    rm -rf "${INSTALL_DIR}"
-    mkdir -p "$(dirname "${INSTALL_DIR}")"
-    tar -C "$(dirname "${INSTALL_DIR}")" -xzf "${tmp}/${file}"
-    # The tarball extracts to a top-level "go/" dir; rename if needed.
-    if [ "$(basename "${INSTALL_DIR}")" != "go" ]; then
-      mv "$(dirname "${INSTALL_DIR}")/go" "${INSTALL_DIR}"
-    fi
-  fi
-
-  "${INSTALL_DIR}/bin/go" version || die "go did not run after install"
-
-  if [ "${PERSIST}" = "1" ]; then
-    if ! grep -q "${INSTALL_DIR}/bin" "${HOME}/.bashrc" 2>/dev/null; then
-      printf '\nexport PATH="%s/bin:$PATH"\n' "${INSTALL_DIR}" >> "${HOME}/.bashrc"
-      log "appended PATH to ~/.bashrc"
-    fi
-  fi
-
-  cat >&2 <<EOF
-
-Go ${ver} ready. Add it to PATH for this shell:
-
-    export PATH="${INSTALL_DIR}/bin:\$PATH"
-
-Then, from ${REPO_ROOT}:
-
-    go build ./... && go vet ./... && go test ./plugins/inputs/asr/...
-EOF
-}
-
-main "$@"

From e16c985194f0e44e418356f1b9e3c9cccef88128 Mon Sep 17 00:00:00 2001
From: openmindev <147775420+openminddev@users.noreply.github.com>
Date: Fri, 28 Aug 2026 13:37:08 -0700
Subject: [PATCH 13/15] Remove comment

---
 plugins/inputs/asr/elevenlabs_asr_rtsp.go | 1 -
 1 file changed, 1 deletion(-)

diff --git a/plugins/inputs/asr/elevenlabs_asr_rtsp.go b/plugins/inputs/asr/elevenlabs_asr_rtsp.go
index 5862a8fa5a..3ca53278e2 100644
--- a/plugins/inputs/asr/elevenlabs_asr_rtsp.go
+++ b/plugins/inputs/asr/elevenlabs_asr_rtsp.go
@@ -187,7 +187,6 @@ func (s *ElevenLabsASRRTSPSensor) streamRTSP(ctx context.Context) error {
 		if _, err := io.ReadFull(stdout, buf); err != nil {
 			return fmt.Errorf("read pcm: %w", err)
 		}
-		// Stamp capture time at read.
 		tCapture := time.Now()
 
 		if tts.Speaking.Load() && !s.cfg.EnableTTSInterrupt {

From 2110da128f8bde5cbbe75aac5e09385a8096a63c Mon Sep 17 00:00:00 2001
From: openmindev <147775420+openminddev@users.noreply.github.com>
Date: Fri, 28 Aug 2026 13:37:31 -0700
Subject: [PATCH 14/15] Remove comments

---
 docker-compose.yml | 7 -------
 1 file changed, 7 deletions(-)

diff --git a/docker-compose.yml b/docker-compose.yml
index b463e8e622..b8af40bdac 100644
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -13,13 +13,6 @@ services:
       - OM_API_KEY=${OM_API_KEY}
       - OM1_COMMAND=${OM1_COMMAND:-unitree_g1_conversation}
       - OM1_SKIP_INTERNET_CHECK=${OM1_SKIP_INTERNET_CHECK:-false}
-      # ASR/VLM source selection. ASR default is the local mic (standalone-safe);
-      # VLM defaults to the video-processor's raw camera view. When paired with
-      # the GStreamer video-processor, consume its streams directly (low latency;
-      # mediamtx is for people, not the agent):
-      #   ASR_INPUT_PLUGIN=GoogleASRRTSPInput   (audio from :8555/live, -vn)
-      #   ASR_RTSP_URL=rtsp://localhost:8555/live
-      #   VLM_RTSP_URL=rtsp://localhost:8556/raw   (clean scene for the VLM)
       - ASR_INPUT_PLUGIN=${ASR_INPUT_PLUGIN:-GoogleASRInput}
       - ASR_RTSP_URL=${ASR_RTSP_URL:-rtsp://localhost:8555/live}
       - VLM_RTSP_URL=${VLM_RTSP_URL:-rtsp://localhost:8556/raw}

From 90c31d32cb6e83b34a90039482b6dbbcbe8ba163 Mon Sep 17 00:00:00 2001
From: openmindev <147775420+openminddev@users.noreply.github.com>
Date: Fri, 28 Aug 2026 13:41:40 -0700
Subject: [PATCH 15/15] Remove comments

---
 plugins/inputs/asr/google_asr_rtsp.go | 2 --
 1 file changed, 2 deletions(-)

diff --git a/plugins/inputs/asr/google_asr_rtsp.go b/plugins/inputs/asr/google_asr_rtsp.go
index 7f71db8e51..2fddeed9e6 100644
--- a/plugins/inputs/asr/google_asr_rtsp.go
+++ b/plugins/inputs/asr/google_asr_rtsp.go
@@ -195,8 +195,6 @@ func (s *GoogleASRRTSPSensor) streamRTSP(ctx context.Context) error {
 		if _, err := io.ReadFull(stdout, buf); err != nil {
 			return fmt.Errorf("read pcm: %w", err)
 		}
-		// Stamp capture time at read, so the ASR chunk carries the capture
-		// instant rather than the later send time.
 		tCapture := time.Now()
 
 		if tts.Speaking.Load() && !s.cfg.EnableTTSInterrupt {