Merge pull request #308 from kerberos-io/feat/recording-fps-metadata

feat: propagate recording FPS with uploads

machinery/go.mod

@@ -24,7 +24,7 @@ require (
 	github.com/gorilla/websocket v1.5.3
 	github.com/kellydunn/golang-geo v0.7.0
 	github.com/kerberos-io/joy4 v1.0.64
-	github.com/kerberos-io/onvif v1.2.1
+	github.com/kerberos-io/onvif v1.2.2
 	github.com/minio/minio-go/v6 v6.0.57
 	github.com/nfnt/resize v0.0.0-20180221191011-83c6a9932646
 	github.com/op/go-logging v0.0.0-20160315200505-970db520ece7
@@ -32,6 +32,7 @@ require (
 	github.com/pion/rtp v1.8.19
 	github.com/pion/webrtc/v4 v4.1.2
 	github.com/sirupsen/logrus v1.9.3
+	github.com/stretchr/testify v1.10.0
 	github.com/swaggo/files v1.0.1
 	github.com/swaggo/gin-swagger v1.6.0
 	github.com/swaggo/swag v1.16.4
@@ -58,6 +59,7 @@ require (
 	github.com/clbanning/mxj v1.8.4 // indirect
 	github.com/clbanning/mxj/v2 v2.7.0 // indirect
 	github.com/cloudwego/base64x v0.1.5 // indirect
+	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/elastic/go-windows v1.0.2 // indirect
 	github.com/elgs/gostrgen v0.0.0-20161222160715-9d61ae07eeae // indirect
 	github.com/erikstmartin/go-testdb v0.0.0-20160219214506-8d10e4a1bae5 // indirect
@@ -108,6 +110,7 @@ require (
 	github.com/pion/stun/v3 v3.0.0 // indirect
 	github.com/pion/transport/v3 v3.0.7 // indirect
 	github.com/pion/turn/v4 v4.0.0 // indirect
+	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/prometheus/procfs v0.15.1 // indirect
 	github.com/twitchyliquid64/golang-asm v0.15.1 // indirect
 	github.com/ugorji/go/codec v1.2.12 // indirect

machinery/go.sum

@@ -774,8 +774,8 @@ github.com/kellydunn/golang-geo v0.7.0 h1:A5j0/BvNgGwY6Yb6inXQxzYwlPHc6WVZR+Mrar
 github.com/kellydunn/golang-geo v0.7.0/go.mod h1:YYlQPJ+DPEzrHx8kT3oPHC/NjyvCCXE+IuKGKdrjrcU=
 github.com/kerberos-io/joy4 v1.0.64 h1:gTUSotHSOhp9mNqEecgq88tQHvpj7TjmrvPUsPm0idg=
 github.com/kerberos-io/joy4 v1.0.64/go.mod h1:nZp4AjvKvTOXRrmDyAIOw+Da+JA5OcSo/JundGfOlFU=
-github.com/kerberos-io/onvif v1.2.1 h1:+vxyHPylt0ufK8gv7FL+KzhJUeComMGrTmP5KxT2YEc=
-github.com/kerberos-io/onvif v1.2.1/go.mod h1:XSgEQXmEDjUQTbdXvsaRJt6Az8YPGj7L+j5iXKEGijU=
+github.com/kerberos-io/onvif v1.2.2 h1:QnxITps7xvAVD2abWRsa3+p9QexjKESQQldUFMx/mYA=
+github.com/kerberos-io/onvif v1.2.2/go.mod h1:XSgEQXmEDjUQTbdXvsaRJt6Az8YPGj7L+j5iXKEGijU=
 github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8=
 github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
 github.com/klauspost/asmfmt v1.3.2/go.mod h1:AG8TuvYojzulgDAMCnYn50l/5QV3Bs/tp6j0HLHbNSE=

machinery/src/capture/cleanup_test.go

@@ -28,7 +28,8 @@ func writeRecording(t *testing.T, recordingsDir, name string, ageMinutes int) {
 // marking it as still queued for upload.
 func markPending(t *testing.T, cloudDir, name string) {
 	t.Helper()
-	if err := os.WriteFile(filepath.Join(cloudDir, name), nil, 0o644); err != nil {
+	markerName := models.RecordingUploadMetadataFileName(name)
+	if err := os.WriteFile(filepath.Join(cloudDir, markerName), nil, 0o644); err != nil {
 		t.Fatalf("write marker %s: %v", name, err)
 	}
 }
@@ -71,6 +72,24 @@ func TestPickRecordingToCleanup_PrefersUploaded(t *testing.T) {
 	}
 }
 
+func TestPickRecordingToCleanup_RecognizesLegacyMarkerName(t *testing.T) {
+	recordingsDir, cloudDir := newCleanupDirs(t)
+
+	writeRecording(t, recordingsDir, "legacy_pending.mp4", 30)
+	if err := os.WriteFile(filepath.Join(cloudDir, "legacy_pending.mp4"), nil, 0o644); err != nil {
+		t.Fatalf("write legacy marker: %v", err)
+	}
+	writeRecording(t, recordingsDir, "uploaded.mp4", 10)
+
+	name, pending, err := pickRecordingToCleanup(recordingsDir, cloudDir)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if pending || name != "uploaded.mp4" {
+		t.Fatalf("cleanup picked name=%q pending=%v, want uploaded.mp4 pending=false", name, pending)
+	}
+}
+
 // Among several already-uploaded recordings, the oldest uploaded one is chosen.
 func TestPickRecordingToCleanup_OldestUploadedFirst(t *testing.T) {
 	recordingsDir, cloudDir := newCleanupDirs(t)

machinery/src/capture/main.go

@@ -4,8 +4,11 @@ package capture
 import (
 	"context"
 	"encoding/base64"
+	"encoding/json"
 	"image"
+	"math"
 	"os"
+	"path/filepath"
 	"strconv"
 	"time"
 
@@ -50,6 +53,52 @@ func publishRecordingState(mqttClient mqtt.Client, hubKey string, configuration
 	}
 }
 
+func recordingUploadMetadata(name, deviceKey string, timestamp int64, mp4Video *video.MP4) models.RecordingUploadMetadata {
+	metadata := models.RecordingUploadMetadata{
+		FileName:  filepath.Base(name),
+		DeviceKey: deviceKey,
+		Timestamp: timestamp,
+		Duration:  mp4Video.VideoTotalDuration,
+	}
+	value := mp4Video.AverageFPS()
+	if value > 0 && value <= 240 && !math.IsInf(value, 0) && !math.IsNaN(value) {
+		metadata.FPS = int(math.Floor(value))
+	}
+	return metadata
+}
+
+// queueRecordingForUpload creates the marker consumed by the upload worker and
+// stores metadata captured from the finalized recording.
+func queueRecordingForUpload(configDirectory string, metadata models.RecordingUploadMetadata) {
+	payload, err := json.Marshal(metadata)
+	if err != nil {
+		log.Log.Error("capture.main.queueRecordingForUpload(): " + err.Error())
+		return
+	}
+
+	// Publish the marker with a same-filesystem rename. Writing directly to the
+	// watched directory would briefly expose an empty file to the upload poller.
+	marker, err := os.CreateTemp(filepath.Join(configDirectory, "data"), ".upload-marker-*")
+	if err == nil {
+		_, err = marker.Write(payload)
+	}
+	if err == nil {
+		err = marker.Chmod(0644)
+	}
+	if marker != nil {
+		if closeErr := marker.Close(); err == nil {
+			err = closeErr
+		}
+		defer os.Remove(marker.Name())
+	}
+	if err == nil {
+		err = os.Rename(marker.Name(), filepath.Join(configDirectory, "data", "cloud", models.RecordingUploadMetadataFileName(metadata.FileName)))
+	}
+	if err != nil {
+		log.Log.Error("capture.main.queueRecordingForUpload(): " + err.Error())
+	}
+}
+
 const (
 	// manualRecordingHeartbeatTimeout is how long the agent keeps a manual
 	// (live-view / remote) recording alive after the LAST viewer heartbeat. The
@@ -152,8 +201,10 @@ func CleanupRecordingDirectory(configDirectory string, configuration *models.Con
 		// now-dangling upload marker so the upload loop doesn't keep trying to
 		// upload a file that no longer exists.
 		log.Log.Warning("HandleRecordStream: removed oldest recording as part of cleanup, but it was STILL PENDING UPLOAD (disk full of un-uploaded recordings) - " + recordingsDirectory + "/" + name)
-		if err := os.Remove(cloudDirectory + "/" + name); err != nil && !os.IsNotExist(err) {
-			log.Log.Info("HandleRecordStream: could not remove dangling upload marker " + name + ", " + err.Error())
+		for _, markerName := range uploadMarkerNames(name) {
+			if err := os.Remove(filepath.Join(cloudDirectory, markerName)); err != nil && !os.IsNotExist(err) {
+				log.Log.Info("HandleRecordStream: could not remove dangling upload marker " + markerName + ", " + err.Error())
+			}
 		}
 	} else {
 		log.Log.Info("HandleRecordStream: removed oldest file as part of cleanup - " + recordingsDirectory + "/" + name)
@@ -246,9 +297,9 @@ func pickRecordingToCleanup(recordingsDirectory, cloudDirectory string) (string,
 			oldestAnyTime = modTime
 		}
 
-		// A recording is still pending upload if a marker with the same name
-		// exists in the cloud directory. Skip those when picking a safe candidate.
-		if _, statErr := os.Stat(cloudDirectory + "/" + entry.Name()); statErr == nil {
+		// A recording is still pending upload if either its current .metadata
+		// marker or a marker created by an older agent exists.
+		if recordingPendingUpload(cloudDirectory, entry.Name()) {
 			continue
 		}
 
@@ -267,6 +318,19 @@ func pickRecordingToCleanup(recordingsDirectory, cloudDirectory string) (string,
 	return "", false, os.ErrNotExist
 }
 
+func uploadMarkerNames(recordingName string) []string {
+	return []string{models.RecordingUploadMetadataFileName(recordingName), filepath.Base(recordingName)}
+}
+
+func recordingPendingUpload(cloudDirectory, recordingName string) bool {
+	for _, markerName := range uploadMarkerNames(recordingName) {
+		if _, err := os.Stat(filepath.Join(cloudDirectory, markerName)); err == nil {
+			return true
+		}
+	}
+	return false
+}
+
 func HandleRecordStream(queue *packets.Queue, configDirectory string, configuration *models.Configuration, communication *models.Communication, rtspClient RTSPClient, mqttClient mqtt.Client) {
 
 	config := configuration.Config
@@ -438,9 +502,7 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
 						}
 					}
 
-					// Create a symbol link.
-					fc, _ := os.Create(configDirectory + "/data/cloud/" + name)
-					fc.Close()
+					queueRecordingForUpload(configDirectory, recordingUploadMetadata(name, config.Key, startRecording, mp4Video))
 
 					recordingStatus = "idle"
 
@@ -597,9 +659,7 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
 						}
 					}
 
-					// Create a symbol link.
-					fc, _ := os.Create(configDirectory + "/data/cloud/" + name)
-					fc.Close()
+					queueRecordingForUpload(configDirectory, recordingUploadMetadata(name, config.Key, startRecording, mp4Video))
 
 					recordingStatus = "idle"
 
@@ -869,9 +929,7 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
 					}
 				}
 
-				// Create a symbol linc.
-				fc, _ := os.Create(configDirectory + "/data/cloud/" + name)
-				fc.Close()
+				queueRecordingForUpload(configDirectory, recordingUploadMetadata(name, config.Key, displayTime, mp4Video))
 
 				// Clean up the recording directory if necessary.
 				CleanupRecordingDirectory(configDirectory, configuration)

machinery/src/capture/main_test.go

@@ -0,0 +1,60 @@
+package capture
+
+import (
+	"encoding/json"
+	"math"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/kerberos-io/agent/machinery/src/models"
+	"github.com/kerberos-io/agent/machinery/src/video"
+)
+
+func TestQueueRecordingForUploadStoresFinalizedMetadata(t *testing.T) {
+	configDirectory := t.TempDir()
+	if err := os.MkdirAll(filepath.Join(configDirectory, "data", "cloud"), 0o755); err != nil {
+		t.Fatalf("mkdir cloud queue: %v", err)
+	}
+
+	mp4Video := &video.MP4{VideoTotalDuration: 20452, SampleCount: 613}
+	metadata := recordingUploadMetadata("recording.mp4", "device-key", 1785934709414, mp4Video)
+	queueRecordingForUpload(configDirectory, metadata)
+
+	got, err := os.ReadFile(filepath.Join(configDirectory, "data", "cloud", "recording.metadata"))
+	if err != nil {
+		t.Fatalf("read upload marker: %v", err)
+	}
+	var stored models.RecordingUploadMetadata
+	if err := json.Unmarshal(got, &stored); err != nil {
+		t.Fatalf("decode upload marker: %v", err)
+	}
+	if stored.FileName != "recording.mp4" || stored.DeviceKey != "device-key" || stored.Timestamp != 1785934709414 || stored.Duration != 20452 || stored.FPS != 29 {
+		t.Fatalf("upload marker = %+v", stored)
+	}
+}
+
+func TestQueueRecordingForUploadKeepsUnknownFPSCompatible(t *testing.T) {
+	for _, fps := range []float64{0, 0.99, -1, math.NaN(), math.Inf(1), 241} {
+		t.Run("invalid FPS", func(t *testing.T) {
+			configDirectory := t.TempDir()
+			if err := os.MkdirAll(filepath.Join(configDirectory, "data", "cloud"), 0o755); err != nil {
+				t.Fatalf("mkdir cloud queue: %v", err)
+			}
+
+			metadata := models.RecordingUploadMetadata{FileName: "recording.mp4"}
+			if fps >= 1 && fps <= 240 && !math.IsNaN(fps) && !math.IsInf(fps, 0) {
+				metadata.FPS = int(math.Floor(fps))
+			}
+			queueRecordingForUpload(configDirectory, metadata)
+
+			got, err := os.ReadFile(filepath.Join(configDirectory, "data", "cloud", "recording.metadata"))
+			if err != nil {
+				t.Fatalf("read upload marker: %v", err)
+			}
+			if string(got) != `{"filename":"recording.mp4","device_key":"","timestamp":0,"duration":0}` {
+				t.Fatalf("upload marker = %q, want metadata without FPS", got)
+			}
+		})
+	}
+}

machinery/src/cloud/cloud.go

@@ -78,7 +78,8 @@ func HandleUpload(configDirectory string, configuration *models.Configuration, c
 					default:
 					}
 
-					fileName := f.Name()
+					markerFileName := f.Name()
+					fileName := models.RecordingFileNameFromUploadMarker(markerFileName)
 					uploaded := false
 					configured := false
 					err = nil
@@ -113,7 +114,7 @@ func HandleUpload(configDirectory string, configuration *models.Configuration, c
 					// Check if the file is uploaded, if so, remove it.
 					if uploaded {
 						delay = 500 * time.Millisecond // reset
-						err := os.Remove(watchDirectory + fileName)
+						err := os.Remove(watchDirectory + markerFileName)
 						if err != nil {
 							log.Log.Error("HandleUpload: " + err.Error())
 						}
@@ -127,7 +128,7 @@ func HandleUpload(configDirectory string, configuration *models.Configuration, c
 							}
 						}
 					} else if !configured {
-						err := os.Remove(watchDirectory + fileName)
+						err := os.Remove(watchDirectory + markerFileName)
 						if err != nil {
 							log.Log.Error("HandleUpload: " + err.Error())
 						}
@@ -231,14 +232,20 @@ func rawJSONOrEmptyArray(b []byte) json.RawMessage {
 func HandleHeartBeat(configuration *models.Configuration, communication *models.Communication, uptimeStart time.Time) {
 	log.Log.Debug("cloud.HandleHeartBeat(): started")
 
+	// Bound every heartbeat POST so a stalled connection (e.g. a saturated uplink
+	// or an unresponsive Hub/Vault) fails fast on this cycle instead of blocking
+	// the whole heartbeat loop indefinitely. A hung POST would otherwise stop all
+	// further heartbeats, and Hub marks a camera offline once its last heartbeat is
+	// older than 180s even while capture is healthy.
+	const heartbeatHTTPTimeout = 30 * time.Second
 	var client *http.Client
 	if os.Getenv("AGENT_TLS_INSECURE") == "true" {
 		tr := &http.Transport{
 			TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
 		}
-		client = &http.Client{Transport: tr}
+		client = &http.Client{Transport: tr, Timeout: heartbeatHTTPTimeout}
 	} else {
-		client = &http.Client{}
+		client = &http.Client{Timeout: heartbeatHTTPTimeout}
 	}
 
 	kerberosAgentVersion := utils.VERSION

machinery/src/cloud/kerberos_hub.go

@@ -85,6 +85,7 @@ func UploadKerberosHub(configuration *models.Configuration, fileName string) (bo
 	req.Header.Set("X-Kerberos-Hub-PublicKey", config.HubKey)
 	req.Header.Set("X-Kerberos-Hub-PrivateKey", config.HubPrivateKey)
 	req.Header.Set("X-Kerberos-Hub-Region", config.S3.Region)
+	setQueuedRecordingMetadataHeaders(req.Header, fileName)
 
 	var client *http.Client
 	if os.Getenv("AGENT_TLS_INSECURE") == "true" {
@@ -128,6 +129,7 @@ func UploadKerberosHub(configuration *models.Configuration, fileName string) (bo
 	req.Header.Set("X-Kerberos-Hub-PublicKey", config.HubKey)
 	req.Header.Set("X-Kerberos-Hub-PrivateKey", config.HubPrivateKey)
 	req.Header.Set("X-Kerberos-Hub-Region", config.S3.Region)
+	setQueuedRecordingMetadataHeaders(req.Header, fileName)
 	resp, err = client.Do(req)
 	if resp != nil {
 		defer resp.Body.Close()

machinery/src/cloud/kerberos_vault.go

@@ -165,6 +165,7 @@ func uploadVaultLegacy(vault models.KStorage, publicKey, deviceKey, fileName, la
 	}
 	req.Header.Set("Content-Type", "video/mp4")
 	setVaultHeaders(req.Header, vault, publicKey, deviceKey, fileName)
+	setQueuedRecordingMetadataHeaders(req.Header, fileName)
 
 	client := newVaultHTTPClient(0)
 	resp, err := client.Do(req)

machinery/src/cloud/recording_metadata.go

@@ -0,0 +1,88 @@
+package cloud
+
+import (
+	"encoding/json"
+	"math"
+	"net/http"
+	"os"
+	"path/filepath"
+	"strconv"
+	"strings"
+
+	"github.com/kerberos-io/agent/machinery/src/models"
+)
+
+const recordingFPSHeader = "X-Kerberos-Storage-Fps"
+const recordingDurationHeader = "X-Kerberos-Storage-Duration"
+const recordingTimestampHeader = "X-Kerberos-Storage-Timestamp"
+
+// queuedRecordingFPS reads the FPS snapshot written into the upload marker
+// when the recording was finalized. Historical empty markers intentionally
+// return no value so receivers can retain their existing MP4-derived fallback.
+func queuedRecordingFPS(fileName string) string {
+	value, ok := readRecordingUploadMetadata(fileName)
+	if !ok {
+		return ""
+	}
+
+	marker := strings.TrimSpace(string(value))
+	if strings.HasPrefix(marker, "{") {
+		metadata, ok := decodeRecordingUploadMetadata(value)
+		if !ok || metadata.FPS <= 0 || metadata.FPS > 240 {
+			return ""
+		}
+		return strconv.Itoa(metadata.FPS)
+	}
+
+	// Compatibility with markers created before upload metadata used JSON.
+	fps := marker
+	parsed, err := strconv.ParseFloat(fps, 64)
+	if err != nil || parsed <= 0 || parsed > 240 || math.IsInf(parsed, 0) || math.IsNaN(parsed) {
+		return ""
+	}
+	return fps
+}
+
+func queuedRecordingMetadata(fileName string) (models.RecordingUploadMetadata, bool) {
+	value, ok := readRecordingUploadMetadata(fileName)
+	if !ok || !strings.HasPrefix(strings.TrimSpace(string(value)), "{") {
+		return models.RecordingUploadMetadata{}, false
+	}
+	return decodeRecordingUploadMetadata(value)
+}
+
+func decodeRecordingUploadMetadata(value []byte) (models.RecordingUploadMetadata, bool) {
+	var metadata models.RecordingUploadMetadata
+	if err := json.Unmarshal(value, &metadata); err != nil {
+		return models.RecordingUploadMetadata{}, false
+	}
+	return metadata, true
+}
+
+func readRecordingUploadMetadata(fileName string) ([]byte, bool) {
+	markerNames := []string{
+		models.RecordingUploadMetadataFileName(fileName),
+		filepath.Base(fileName),
+	}
+	for _, markerName := range markerNames {
+		value, err := os.ReadFile(filepath.Join("data", "cloud", markerName))
+		if err == nil {
+			return value, true
+		}
+	}
+	return nil, false
+}
+
+func setQueuedRecordingMetadataHeaders(header http.Header, fileName string) {
+	if fps := queuedRecordingFPS(fileName); fps != "" {
+		header.Set(recordingFPSHeader, fps)
+	}
+	if metadata, ok := queuedRecordingMetadata(fileName); ok {
+		if metadata.Duration > 0 {
+			header.Set(recordingDurationHeader, strconv.FormatUint(metadata.Duration, 10))
+		}
+		if metadata.Timestamp > 0 {
+			header.Set(recordingTimestampHeader, strconv.FormatInt(metadata.Timestamp, 10))
+		}
+	}
+}

machinery/src/cloud/tus_client.go

@@ -163,6 +163,15 @@ func runTusUpload(baseURL, metadata, fileName, label, slot string, setHeaders tu
 	// never consume the retry budget (matching the legacy single-POST behaviour).
 	lastStatus := 0
 
+	// highWaterOffset is the furthest server-acknowledged offset observed across
+	// all attempts (via HEAD or PATCH). It lets the retry budget be refreshed only
+	// on GENUINE net forward progress. Without it, a server that keeps resetting the
+	// offset — e.g. a persistent 409 ERR_MISMATCHED_OFFSET where HEAD reports 0 again
+	// while the first chunk still "succeeds" — would refresh the budget every attempt
+	// and loop forever, wedging the upload worker on one recording and saturating the
+	// uplink.
+	highWaterOffset := int64(0)
+
 	for attempt := 0; attempt < maxAttempts; attempt++ {
 		// (1) Ensure we have an active upload URL, creating one if needed.
 		if uploadURL == "" {
@@ -196,6 +205,15 @@ func runTusUpload(baseURL, metadata, fileName, label, slot string, setHeaders tu
 			continue
 		}
 
+		// The furthest offset any previous attempt reached. If this attempt pushes
+		// past it (via HEAD showing server-side progress or a successful PATCH) we made
+		// genuine net progress and may refresh the retry budget; if not, a repeated
+		// failure at the same spot must count against maxAttempts.
+		startHighWater := highWaterOffset
+		if offset > highWaterOffset {
+			highWaterOffset = offset
+		}
+
 		// (3) All bytes are present but the upload was not finalized (e.g. the
 		// completion hook failed). A completed tus upload cannot be re-finalized
 		// with another PATCH, so delete it and re-upload to force a clean finalize.
@@ -216,7 +234,6 @@ func runTusUpload(baseURL, metadata, fileName, label, slot string, setHeaders tu
 		// checkpointing the offset after each one so an interruption resumes from the
 		// last completed chunk instead of re-uploading everything.
 		chunkSize := tusChunkSize()
-		progressed := false
 		patchFailed := false
 		var lastBody string
 		loggedProgressBucket := tusProgressBucket(offset, size)
@@ -244,10 +261,10 @@ func runTusUpload(baseURL, metadata, fileName, label, slot string, setHeaders tu
 				patchFailed = true
 				break
 			}
-			if newOffset > offset {
-				progressed = true
-			}
 			offset = newOffset
+			if offset > highWaterOffset {
+				highWaterOffset = offset
+			}
 			lastBody = respBody
 			logTusUploadProgress(label, offset, size, &loggedProgressBucket)
 			if offset < size {
@@ -256,10 +273,13 @@ func runTusUpload(baseURL, metadata, fileName, label, slot string, setHeaders tu
 			}
 		}
 		if patchFailed {
-			if progressed {
-				// Forward progress refreshes the retry budget: maxAttempts bounds the
-				// number of consecutive failures, not the number of chunks needed for
-				// a large recording.
+			if highWaterOffset > startHighWater {
+				// Genuine net progress (we advanced past the furthest point any previous
+				// attempt reached) refreshes the retry budget: maxAttempts bounds the
+				// number of consecutive *non-progressing* failures, not the number of
+				// chunks needed for a large recording. A server that keeps rejecting the
+				// same offset (no net progress, e.g. a persistent ERR_MISMATCHED_OFFSET)
+				// therefore gives up after maxAttempts instead of retrying forever.
 				attempt = -1
 			}
 			continue
@@ -285,14 +305,17 @@ func runTusUpload(baseURL, metadata, fileName, label, slot string, setHeaders tu
 // is additionally carried in the tus Upload-Metadata.
 func uploadVaultResumable(vault models.KStorage, publicKey, deviceKey, fileName, label, slot string) (bool, bool, bool, string, error) {
 	baseURL := strings.TrimRight(vault.URI, "/") + tusUploadPath
-	metadata := encodeTusMetadata(map[string]string{
+	metadataValues := map[string]string{
 		"filename":  fileName,
 		"device":    deviceKey,
 		"directory": vault.Directory,
 		"provider":  vault.Provider,
 		"capture":   "IPCamera",
 		"cloudkey":  publicKey,
-	})
+		"fps":       queuedRecordingFPS(fileName),
+	}
+	addRecordingTusMetadata(metadataValues, fileName)
+	metadata := encodeTusMetadata(metadataValues)
 	setHeaders := func(h http.Header, fn string) {
 		setVaultTusHeaders(h, vault, publicKey, deviceKey, fn)
 	}
@@ -306,17 +329,33 @@ func uploadVaultResumable(vault models.KStorage, publicKey, deviceKey, fileName,
 // intentionally omitted from the metadata here.
 func uploadHubResumable(config *models.Config, fileName, label, slot string) (bool, bool, bool, string, error) {
 	baseURL := strings.TrimRight(config.HubURI, "/") + tusUploadPath
-	metadata := encodeTusMetadata(map[string]string{
+	metadataValues := map[string]string{
 		"filename": fileName,
 		"device":   config.Key,
 		"capture":  "IPCamera",
-	})
+		"fps":      queuedRecordingFPS(fileName),
+	}
+	addRecordingTusMetadata(metadataValues, fileName)
+	metadata := encodeTusMetadata(metadataValues)
 	setHeaders := func(h http.Header, fn string) {
 		setHubTusHeaders(h, config, fn)
 	}
 	return runTusUpload(baseURL, metadata, fileName, label, slot, setHeaders)
 }
 
+func addRecordingTusMetadata(values map[string]string, fileName string) {
+	metadata, ok := queuedRecordingMetadata(fileName)
+	if !ok {
+		return
+	}
+	if metadata.Duration > 0 {
+		values["duration"] = strconv.FormatUint(metadata.Duration, 10)
+	}
+	if metadata.Timestamp > 0 {
+		values["timestamp"] = strconv.FormatInt(metadata.Timestamp, 10)
+	}
+}
+
 // tusCreate performs the tus "creation" request (POST). On success it returns
 // the resolved upload URL the agent should use for subsequent HEAD/PATCH calls.
 func tusCreate(client *http.Client, baseURL string, size int64, metadata string, setHeaders tusHeaderFunc, fileName string) (string, int, error) {

machinery/src/cloud/tus_client_test.go

@@ -49,6 +49,14 @@ type fakeTus struct {
 	// after storing the bytes, simulating a failed completion hook.
 	failFinalize int
 
+	// loseProgress simulates a vault that never durably retains the in-progress
+	// upload: every PATCH is acknowledged (the response advertises the advanced
+	// offset) but the stored offset is immediately reset to 0. HEAD therefore
+	// keeps reporting 0 and the next chunk — sent at the advanced offset — is
+	// rejected with 409, reproducing the cross-replica ERR_MISMATCHED_OFFSET
+	// loop that previously wedged the agent's upload worker forever.
+	loseProgress bool
+
 	// requests records the headers of every received request (in order) so
 	// tests can assert which auth/routing headers the client sent per method.
 	requests []recordedRequest
@@ -155,6 +163,31 @@ func (s *fakeTus) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 			w.WriteHeader(http.StatusNotFound)
 			return
 		}
+		if s.loseProgress {
+			reqOffset, _ := strconv.ParseInt(r.Header.Get("Upload-Offset"), 10, 64)
+			s.mu.Lock()
+			cur := u.offset
+			if reqOffset != cur {
+				// The offset the client resumes from no longer matches what this
+				// "replica" retained, so reject like a vault returning
+				// ERR_MISMATCHED_OFFSET.
+				s.mu.Unlock()
+				w.Header().Set("Upload-Offset", strconv.FormatInt(cur, 10))
+				w.WriteHeader(http.StatusConflict)
+				return
+			}
+			n, _ := io.Copy(io.Discard, r.Body)
+			s.lastPatchBytes = n
+			s.patchSizes = append(s.patchSizes, n)
+			// Advertise progress to the client, then immediately forget it so the
+			// next chunk (sent at the advanced offset) mismatches again.
+			reported := cur + n
+			u.offset = 0
+			s.mu.Unlock()
+			w.Header().Set("Upload-Offset", strconv.FormatInt(reported, 10))
+			w.WriteHeader(http.StatusNoContent)
+			return
+		}
 		n, _ := io.Copy(io.Discard, r.Body)
 		s.mu.Lock()
 		u.offset += n
@@ -210,6 +243,17 @@ func withRecording(t *testing.T, fileName string, payload []byte) {
 	}
 }
 
+func withQueuedRecordingFPS(t *testing.T, fileName, fps string) {
+	t.Helper()
+	if err := os.MkdirAll("data/cloud", 0o755); err != nil {
+		t.Fatalf("mkdir cloud queue: %v", err)
+	}
+	markerName := models.RecordingUploadMetadataFileName(fileName)
+	if err := os.WriteFile(filepath.Join("data/cloud", markerName), []byte(fps), 0o644); err != nil {
+		t.Fatalf("write cloud queue marker: %v", err)
+	}
+}
+
 func testVault(uri string) models.KStorage {
 	return models.KStorage{
 		URI:             uri,
@@ -228,6 +272,7 @@ func TestUploadVaultResumable_HappyPath(t *testing.T) {
 	fileName := "1564859471_6-474162_oprit_577-283-727-375_1153_27.mp4"
 	payload := bytes.Repeat([]byte("x"), 4096)
 	withRecording(t, fileName, payload)
+	withQueuedRecordingFPS(t, fileName, `{"filename":"recording.mp4","device_key":"device-key","timestamp":1785934709414,"duration":20452,"fps":29}`)
 
 	uploaded, responded, supported, _, err := uploadVaultResumable(testVault(ts.URL), "pk", "dev", fileName, "test", "primary")
 	if err != nil {
@@ -242,6 +287,102 @@ func TestUploadVaultResumable_HappyPath(t *testing.T) {
 	if _, err := os.Stat(tusSidecarPath(fileName, "primary")); !os.IsNotExist(err) {
 		t.Fatalf("expected sidecar to be removed after success, stat err = %v", err)
 	}
+	posts := srv.requestsForMethod(http.MethodPost)
+	metadata := decodeTusMetadata(posts[0].header.Get("Upload-Metadata"))
+	if got := metadata["fps"]; got != "29" {
+		t.Fatalf("POST metadata fps = %q, want %q", got, "29")
+	}
+	if got := metadata["duration"]; got != "20452" {
+		t.Fatalf("POST metadata duration = %q, want %q", got, "20452")
+	}
+	if got := metadata["timestamp"]; got != "1785934709414" {
+		t.Fatalf("POST metadata timestamp = %q, want %q", got, "1785934709414")
+	}
+}
+
+func TestQueuedRecordingFPSValidation(t *testing.T) {
+	for _, test := range []struct {
+		name string
+		fps  string
+		want string
+	}{
+		{name: "json", fps: `{"fps":29}`, want: "29"},
+		{name: "json with future field", fps: `{"fps":29,"codec":"h264"}`, want: "29"},
+		{name: "json without fps", fps: `{}`},
+		{name: "json invalid fps", fps: `{"fps":241}`},
+		{name: "legacy fractional", fps: "29.97", want: "29.97"},
+		{name: "legacy trimmed", fps: " 25 \n", want: "25"},
+		{name: "empty"},
+		{name: "invalid", fps: "invalid"},
+		{name: "zero", fps: "0"},
+		{name: "negative", fps: "-1"},
+		{name: "nan", fps: "NaN"},
+		{name: "infinite", fps: "+Inf"},
+		{name: "unreasonable", fps: "241"},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			fileName := "recording.mp4"
+			withRecording(t, fileName, []byte("recording"))
+			withQueuedRecordingFPS(t, fileName, test.fps)
+
+			if got := queuedRecordingFPS(fileName); got != test.want {
+				t.Fatalf("queuedRecordingFPS() = %q, want %q", got, test.want)
+			}
+
+			header := make(http.Header)
+			setQueuedRecordingMetadataHeaders(header, fileName)
+			if got := header.Get(recordingFPSHeader); got != test.want {
+				t.Fatalf("legacy FPS header = %q, want %q", got, test.want)
+			}
+		})
+	}
+}
+
+func TestQueuedRecordingFPSAllowsMissingHistoricalMarker(t *testing.T) {
+	fileName := "recording.mp4"
+	withRecording(t, fileName, []byte("recording"))
+
+	if got := queuedRecordingFPS(fileName); got != "" {
+		t.Fatalf("queuedRecordingFPS() = %q, want empty for missing marker", got)
+	}
+	header := make(http.Header)
+	setQueuedRecordingMetadataHeaders(header, fileName)
+	if got := header.Get(recordingFPSHeader); got != "" {
+		t.Fatalf("legacy FPS header = %q, want empty for missing marker", got)
+	}
+}
+
+func TestQueuedRecordingMetadataHeaders(t *testing.T) {
+	fileName := "recording.mp4"
+	withRecording(t, fileName, []byte("recording"))
+	withQueuedRecordingFPS(t, fileName, `{"filename":"recording.mp4","device_key":"device-key","timestamp":1785934709414,"duration":20452,"fps":25}`)
+
+	header := make(http.Header)
+	setQueuedRecordingMetadataHeaders(header, fileName)
+	if got := header.Get(recordingFPSHeader); got != "25" {
+		t.Fatalf("FPS header = %q", got)
+	}
+	if got := header.Get(recordingDurationHeader); got != "20452" {
+		t.Fatalf("duration header = %q", got)
+	}
+	if got := header.Get(recordingTimestampHeader); got != "1785934709414" {
+		t.Fatalf("timestamp header = %q", got)
+	}
+}
+
+func TestQueuedRecordingFPSAllowsLegacyMarkerFileName(t *testing.T) {
+	fileName := "recording.mp4"
+	withRecording(t, fileName, []byte("recording"))
+	if err := os.MkdirAll("data/cloud", 0o755); err != nil {
+		t.Fatalf("mkdir cloud queue: %v", err)
+	}
+	if err := os.WriteFile(filepath.Join("data/cloud", fileName), []byte("25"), 0o644); err != nil {
+		t.Fatalf("write legacy cloud queue marker: %v", err)
+	}
+
+	if got := queuedRecordingFPS(fileName); got != "25" {
+		t.Fatalf("queuedRecordingFPS() = %q, want legacy marker FPS", got)
+	}
 }
 
 func TestUploadVaultResumable_Chunked(t *testing.T) {
@@ -393,6 +534,61 @@ func TestUploadVaultResumable_NetworkErrorKeepsRetryBudget(t *testing.T) {
 	}
 }
 
+func TestUploadVaultResumable_MismatchedOffsetGivesUp(t *testing.T) {
+	// A vault that never durably retains the in-progress upload (offset resets to
+	// 0 between chunks) makes every resume "progress" by one chunk and then fail
+	// the next chunk with 409. Before the high-water gating fix this refreshed the
+	// retry budget every attempt and looped forever, wedging the upload worker and
+	// saturating the uplink (which starved heartbeats and reported the camera
+	// offline). The loop must now be bounded: give up after a fixed number of
+	// non-progressing attempts and report responded=true so the caller re-queues.
+	srv := newFakeTus()
+	srv.loseProgress = true
+	ts := httptest.NewServer(srv)
+	defer ts.Close()
+
+	// Keep the between-attempt back-off tiny so the test stays fast.
+	oldDelay := tusBackoffBaseDelay
+	tusBackoffBaseDelay = time.Millisecond
+	defer func() { tusBackoffBaseDelay = oldDelay }()
+
+	// Force multiple chunks so there is always a second chunk to be rejected.
+	t.Setenv("AGENT_TUS_CHUNK_SIZE_BYTES", "4096")
+
+	fileName := "1564859471_6-474162_oprit_577-283-727-375_1153_27.mp4"
+	withRecording(t, fileName, bytes.Repeat([]byte("m"), 12288))
+
+	done := make(chan struct{})
+	var uploaded, responded bool
+	var upErr error
+	go func() {
+		uploaded, responded, _, _, upErr = uploadVaultResumable(testVault(ts.URL), "pk", "dev", fileName, "test", "primary")
+		close(done)
+	}()
+
+	select {
+	case <-done:
+	case <-time.After(30 * time.Second):
+		t.Fatal("resumable upload did not terminate: the retry loop is unbounded on a persistent mismatched offset")
+	}
+
+	if uploaded {
+		t.Fatal("expected uploaded=false when the vault never retains the offset")
+	}
+	if !responded {
+		t.Fatal("expected responded=true (the vault answered) so the caller re-queues the recording")
+	}
+	if upErr == nil {
+		t.Fatal("expected an error when the upload cannot complete")
+	}
+
+	// The bounded retry budget must cap the number of PATCH requests. Two PATCHes
+	// per attempt across a handful of attempts stays comfortably below this.
+	if count, _ := srv.patchCounts(); count > 50 {
+		t.Fatalf("expected a bounded number of PATCH requests, got %d (retry loop not bounded)", count)
+	}
+}
+
 func TestUploadVaultResumable_FinalizeRetry(t *testing.T) {
 	srv := newFakeTus()
 	srv.failFinalize = 1
@@ -491,6 +687,7 @@ func TestUploadHubResumable_HappyPath(t *testing.T) {
 	fileName := "1564859471_6-474162_oprit_577-283-727-375_1153_27.mp4"
 	payload := bytes.Repeat([]byte("h"), 4096)
 	withRecording(t, fileName, payload)
+	withQueuedRecordingFPS(t, fileName, "29.97")
 
 	uploaded, _, supported, _, err := uploadHubResumable(testHubConfig(ts.URL), fileName, "test", "hub")
 	if err != nil {
@@ -561,6 +758,9 @@ func TestUploadHubResumable_HappyPath(t *testing.T) {
 	if meta["capture"] != "IPCamera" {
 		t.Errorf("hub metadata capture = %q, want %q", meta["capture"], "IPCamera")
 	}
+	if meta["fps"] != "29.97" {
+		t.Errorf("hub metadata fps = %q, want %q", meta["fps"], "29.97")
+	}
 }
 
 func TestUploadHubResumable_Unsupported(t *testing.T) {

machinery/src/components/kerberos.go

@@ -341,6 +341,11 @@ func RunAgent(configDirectory string, configuration *models.Configuration, commu
 	communication.HandleONVIF = make(chan models.OnvifAction, 10)
 	go onvif.HandleONVIFActions(configuration, communication)
 
+	// Handle ONVIF event stream — opt-in via Capture.ONVIFMotion="true".
+	// Stops when the agent's shared context is cancelled. The function
+	// is a no-op if ONVIFMotion is not enabled.
+	go onvif.HandleONVIFEventStream(*communication.Context, configuration, communication)
+
 	communication.HandleAudio = make(chan models.AudioDataPartial, 10)
 	if rtspBackChannelClient.HasBackChannel {
 		communication.HasBackChannel = true

machinery/src/computervision/main.go

@@ -24,17 +24,12 @@ func ProcessMotion(motionCursor *packets.QueueCursor, configuration *models.Conf
 	var motionRectangle models.MotionRectangle
 	var motionRectangles []models.MotionRectangle
 
-	// Resolve the motion sensitivity (pixel-change threshold):
-	//   nil (unset) -> default 150
-	//   0           -> motion detection DISABLED (temporary off switch from the UI)
-	//   > 0         -> trigger when the number of changed pixels exceeds it
+	// Resolve the motion sensitivity (pixel-change threshold). Nil, zero, and
+	// negative values use the historical default so older configurations keep
+	// recording after an upgrade.
 	pixelThreshold := 150
-	motionDisabled := false
-	if config.Capture.PixelChangeThreshold != nil {
+	if config.Capture.PixelChangeThreshold != nil && *config.Capture.PixelChangeThreshold > 0 {
 		pixelThreshold = *config.Capture.PixelChangeThreshold
-		if pixelThreshold <= 0 {
-			motionDisabled = true
-		}
 	}
 	// In motion mode we always run detection. In CONTINUOUS mode recording is
 	// 24/7 so motion detection is normally skipped, BUT if a motion region is
@@ -45,11 +40,7 @@ func ProcessMotion(motionCursor *packets.QueueCursor, configuration *models.Conf
 	continuousMode := config.Capture.Continuous == "true"
 	hasMotionRegion := config.Region != nil && len(config.Region.Polygon) > 0
 
-	if motionDisabled {
-
-		log.Log.Info("computervision.main.ProcessMotion(): motion detection disabled (pixelChangeThreshold set to 0), skipping.")
-
-	} else if continuousMode && !hasMotionRegion {
+	if continuousMode && !hasMotionRegion {
 
 		log.Log.Info("computervision.main.ProcessMotion(): continuous recording enabled and no motion region configured, so no motion detection required.")
 
@@ -231,12 +222,13 @@ func ProcessMotion(motionCursor *packets.QueueCursor, configuration *models.Conf
 													"mainWidth":  configuration.Config.Capture.IPCamera.Width,
 													"mainHeight": configuration.Config.Capture.IPCamera.Height,
 													"regions":    motionRectangles,
-													"polygon":    regionPolygons,												// Motion sensitivity = the pixel-change threshold that must
-												// be exceeded before motion triggers. The live view renders
-												// a reference square of sqrt(threshold) px (in this MOTION
-												// frame's pixel space) so the user can visually gauge how
-												// large a moving object must be before it is detected.
-												"pixelChangeThreshold": pixelThreshold,												},
+													"polygon":    regionPolygons, // Motion sensitivity = the pixel-change threshold that must
+													// be exceeded before motion triggers. The live view renders
+													// a reference square of sqrt(threshold) px (in this MOTION
+													// frame's pixel space) so the user can visually gauge how
+													// large a moving object must be before it is detected.
+													"pixelChangeThreshold": pixelThreshold,
+												},
 											},
 										}
 										payload, err := models.PackageMQTTMessage(configuration, message)

machinery/src/config/main.go

@@ -377,6 +377,9 @@ func applyAgentEnvVars(configuration *models.Configuration, prefix string, apply
 			case "AGENT_CAPTURE_MOTION":
 				configuration.Config.Capture.Motion = value
 				break
+			case "AGENT_CAPTURE_ONVIF_MOTION":
+				configuration.Config.Capture.ONVIFMotion = value
+				break
 			case "AGENT_CAPTURE_SNAPSHOTS":
 				configuration.Config.Capture.Snapshots = value
 				break
@@ -648,6 +651,16 @@ func applyAgentEnvVars(configuration *models.Configuration, prefix string, apply
 		}
 	}
 
+	// Motion sensitivity historically used 0 to mean "use the default". Preserve
+	// that behaviour for configurations created before this field became a
+	// pointer, and also recover invalid negative values. Only apply this to the
+	// effective configuration so missing values can still be inherited between
+	// the separate global and custom layers.
+	if applyDefaults && (configuration.Config.Capture.PixelChangeThreshold == nil || *configuration.Config.Capture.PixelChangeThreshold <= 0) {
+		defaultPixelChangeThreshold := 150
+		configuration.Config.Capture.PixelChangeThreshold = &defaultPixelChangeThreshold
+	}
+
 	// Signing is a new feature, so if empty we set default values. Only applied
 	// for the effective configuration (applyDefaults), not for the separate
 	// global/custom views.

machinery/src/config/main_test.go

@@ -0,0 +1,40 @@
+package config
+
+import (
+	"testing"
+
+	"github.com/kerberos-io/agent/machinery/src/models"
+)
+
+func TestApplyAgentEnvVarsPixelChangeThresholdDefault(t *testing.T) {
+	tests := []struct {
+		name      string
+		threshold *int
+		want      int
+	}{
+		{name: "missing", want: 150},
+		{name: "legacy zero", threshold: intPointer(0), want: 150},
+		{name: "negative", threshold: intPointer(-1), want: 150},
+		{name: "positive", threshold: intPointer(275), want: 275},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			configuration := &models.Configuration{}
+			configuration.Config.Capture.PixelChangeThreshold = test.threshold
+
+			applyAgentEnvVars(configuration, "TEST_", true)
+
+			if configuration.Config.Capture.PixelChangeThreshold == nil {
+				t.Fatal("PixelChangeThreshold is nil after applying defaults")
+			}
+			if got := *configuration.Config.Capture.PixelChangeThreshold; got != test.want {
+				t.Fatalf("PixelChangeThreshold = %d, want %d", got, test.want)
+			}
+		})
+	}
+}
+
+func intPointer(value int) *int {
+	return &value
+}

machinery/src/models/config.go

@@ -75,6 +75,14 @@ type Capture struct {
 	Fragmented            string      `json:"fragmented,omitempty" bson:"fragmented,omitempty"`
 	FragmentedDuration    int64       `json:"fragmentedduration,omitempty" bson:"fragmentedduration,omitempty"`
 	PixelChangeThreshold  *int        `json:"pixelChangeThreshold,omitempty"`
+	// ONVIFMotion routes the camera's ONVIF motion events into the
+	// agent's motion-triggered recording pipeline. When "true" the
+	// agent opens an event/stream against the configured ONVIF
+	// endpoint and forwards Motion+Active events to HandleMotion.
+	// Requires Capture.IPCamera.ONVIFXAddr / ONVIFUsername /
+	// ONVIFPassword to be set. Default empty (disabled) keeps the
+	// existing pixel-diff motion detection as the only source.
+	ONVIFMotion string `json:"onvif_motion,omitempty" bson:"onvif_motion,omitempty"`
 }
 
 // IPCamera configuration, such as the RTSP url of the IPCamera and the FPS.

machinery/src/models/recording_upload_metadata.go

@@ -0,0 +1,37 @@
+package models
+
+import (
+	"path/filepath"
+	"strings"
+)
+
+const RecordingUploadMetadataExtension = ".metadata"
+
+// RecordingUploadMetadata is persisted in the upload queue marker associated
+// with a recording. New optional fields can be added without changing the queue
+// mechanism or breaking older agents.
+type RecordingUploadMetadata struct {
+	FileName  string `json:"filename"`
+	DeviceKey string `json:"device_key"`
+	Timestamp int64  `json:"timestamp"` // Unix milliseconds.
+	Duration  uint64 `json:"duration"`  // Milliseconds.
+	FPS       int    `json:"fps,omitempty"`
+}
+
+// RecordingUploadMetadataFileName returns the queue marker name associated
+// with a recording, replacing the recording extension with .metadata.
+func RecordingUploadMetadataFileName(recordingFileName string) string {
+	name := filepath.Base(recordingFileName)
+	extension := filepath.Ext(name)
+	return strings.TrimSuffix(name, extension) + RecordingUploadMetadataExtension
+}
+
+// RecordingFileNameFromUploadMarker resolves a queue entry to its recording.
+// Markers created by older agents used the recording filename directly.
+func RecordingFileNameFromUploadMarker(markerFileName string) string {
+	name := filepath.Base(markerFileName)
+	if strings.HasSuffix(name, RecordingUploadMetadataExtension) {
+		return strings.TrimSuffix(name, RecordingUploadMetadataExtension) + ".mp4"
+	}
+	return name
+}

machinery/src/models/recording_upload_metadata_test.go

@@ -0,0 +1,15 @@
+package models
+
+import "testing"
+
+func TestRecordingUploadMetadataFileNames(t *testing.T) {
+	if got := RecordingUploadMetadataFileName("141245_x_x_.mp4"); got != "141245_x_x_.metadata" {
+		t.Fatalf("RecordingUploadMetadataFileName() = %q", got)
+	}
+	if got := RecordingFileNameFromUploadMarker("141245_x_x_.metadata"); got != "141245_x_x_.mp4" {
+		t.Fatalf("RecordingFileNameFromUploadMarker() = %q", got)
+	}
+	if got := RecordingFileNameFromUploadMarker("legacy.mp4"); got != "legacy.mp4" {
+		t.Fatalf("legacy RecordingFileNameFromUploadMarker() = %q", got)
+	}
+}

machinery/src/onvif/events.go

@@ -0,0 +1,225 @@
+package onvif
+
+import (
+	"context"
+	"errors"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/kerberos-io/agent/machinery/src/log"
+	"github.com/kerberos-io/agent/machinery/src/models"
+	"github.com/kerberos-io/onvif/event/stream"
+)
+
+// The library handles in-stream reconnect; these guards cover the
+// initial-connect path the library cannot see.
+const (
+	initialBackoff = time.Second
+	maxBackoff     = 5 * time.Minute
+)
+
+// HandleONVIFEventStream opens an event/stream against the configured
+// ONVIF camera and routes Motion events into communication.HandleMotion.
+//
+// Behind the Capture.ONVIFMotion flag; the goroutine returns
+// immediately when not enabled. The flag is read once at start, so
+// toggling at runtime requires an agent restart. On transient
+// construction failure (camera not yet ready at boot, brief network
+// blip, credential reload) the goroutine retries with exponential
+// backoff. Exits when ctx is cancelled.
+func HandleONVIFEventStream(ctx context.Context, configuration *models.Configuration, communication *models.Communication) {
+	log.Log.Debug("onvif.HandleONVIFEventStream(): started")
+	defer log.Log.Debug("onvif.HandleONVIFEventStream(): finished")
+
+	if !isONVIFMotionEnabled(configuration.Config.Capture.ONVIFMotion) {
+		return
+	}
+	if configuration.Config.Capture.IPCamera.ONVIFXAddr == "" {
+		log.Log.Warning("onvif.HandleONVIFEventStream(): ONVIFMotion enabled but ONVIFXAddr is empty; nothing to do")
+		return
+	}
+
+	backoff := initialBackoff
+	for {
+		if ctx.Err() != nil {
+			return
+		}
+		recoverable := runStreamOnce(ctx, configuration, communication)
+		if !recoverable {
+			return
+		}
+		if !sleepCtx(ctx, backoff) {
+			return
+		}
+		backoff *= 2
+		if backoff > maxBackoff {
+			backoff = maxBackoff
+		}
+	}
+}
+
+// runStreamOnce returns true when the caller should retry construction
+// (transient failure), false on clean ctx-driven shutdown.
+func runStreamOnce(ctx context.Context, configuration *models.Configuration, communication *models.Communication) (retry bool) {
+	camera := configuration.Config.Capture.IPCamera
+
+	device, _, err := ConnectToOnvifDevice(&camera)
+	if err != nil {
+		log.Log.Error("onvif.HandleONVIFEventStream(): connect: " + err.Error())
+		return true
+	}
+
+	deviceID := resolveDeviceID(configuration.Name, camera.ONVIFXAddr)
+	s, err := stream.NewStream(ctx, device, stream.Options{DeviceID: deviceID})
+	if err != nil {
+		log.Log.Error("onvif.HandleONVIFEventStream(): open stream: " + err.Error())
+		return true
+	}
+	defer func() {
+		if err := s.Close(); err != nil {
+			log.Log.Debug("onvif.HandleONVIFEventStream(): close: " + err.Error())
+		}
+	}()
+
+	log.Log.Info("onvif.HandleONVIFEventStream(): consuming events for " + deviceID)
+
+	// recovering = the first successful event after an error streak
+	// logs a recovery line so on-call operators see the clear-of-
+	// condition for the ERROR they were paged on.
+	var recovering bool
+	for {
+		select {
+		case <-ctx.Done():
+			return false
+		case ev, ok := <-s.Events():
+			if !ok {
+				return false
+			}
+			if recovering {
+				log.Log.Info("onvif.HandleONVIFEventStream(): event stream recovered for " + deviceID)
+				recovering = false
+			}
+			dispatchEvent(ctx, ev, configuration, communication)
+		case e, ok := <-s.Errors():
+			if !ok {
+				return false
+			}
+			recovering = true
+			logStreamError(e)
+		}
+	}
+}
+
+// dispatchEvent routes motion-active events to HandleMotion.
+//
+// The ctx pre-check + ctx-in-select guards a shutdown race: the agent
+// closes HandleMotion shortly after cancelling ctx, and a stale event
+// reaching the send would otherwise panic on a closed channel.
+func dispatchEvent(ctx context.Context, ev stream.Event, configuration *models.Configuration, communication *models.Communication) {
+	topic := sanitiseTopic(ev.Topic)
+	if ev.Kind != stream.KindMotion {
+		log.Log.Debug("onvif.dispatchEvent(): non-motion event " + ev.Kind.String() + " topic=" + topic)
+		return
+	}
+	if ev.State != stream.StateActive {
+		return
+	}
+	if !isTransition(ev.Operation) {
+		log.Log.Debug("onvif.dispatchEvent(): " + ev.Operation.String() + " is not a transition, not a trigger: topic=" + topic)
+		return
+	}
+	if configuration.Config.Capture.Recording == "false" {
+		return
+	}
+	if ctx.Err() != nil {
+		return
+	}
+
+	dataToPass := models.MotionDataPartial{
+		Timestamp:       time.Now().Unix(),
+		NumberOfChanges: 0, // ONVIF does not quantify motion area.
+	}
+	select {
+	case <-ctx.Done():
+	case communication.HandleMotion <- dataToPass:
+		// Logged on the send, not before it: this line records that a
+		// recording started, so a dropped event must not leave one.
+		log.Log.Debug("onvif.dispatchEvent(): recording trigger " + ev.Kind.String() + " topic=" + topic)
+	default:
+		log.Log.Debug("onvif.dispatchEvent(): HandleMotion full, dropping ONVIF motion event")
+	}
+}
+
+// isTransition reports whether an operation represents a state change.
+// A camera replays every property's current state as Initialized on
+// each new subscription and announces removals as Deleted; neither is
+// motion starting. Absent (Unknown) counts — PropertyOperation is
+// optional per WS-Notification and many non-property events omit it.
+func isTransition(op stream.PropertyOperation) bool {
+	return op == stream.PropertyChanged || op == stream.PropertyUnknown
+}
+
+// maxLoggedTopic bounds a topic in the log; the wire imposes no limit,
+// and the reject path logs every event received.
+const maxLoggedTopic = 256
+
+// sanitiseTopic makes a camera-controlled topic safe to concatenate
+// into a log line. logrus's coloured text formatter writes the message
+// unquoted, so a raw newline would let a camera forge entries in the
+// log being used to diagnose it.
+func sanitiseTopic(topic string) string {
+	if len(topic) > maxLoggedTopic {
+		topic = topic[:maxLoggedTopic] + "…(truncated)"
+	}
+	quoted := strconv.Quote(topic)
+	return quoted[1 : len(quoted)-1]
+}
+
+// logStreamError logs at a level matching severity: recreate is loud
+// because it usually means the camera is offline; pull and renew are
+// debug because the library recovers from them automatically.
+func logStreamError(e error) {
+	var recreate stream.ErrRecreateFailed
+	var pull stream.ErrPullFailed
+	var renew stream.ErrRenewFailed
+	switch {
+	case errors.As(e, &recreate):
+		log.Log.Error("onvif.HandleONVIFEventStream(): subscription recreate failed (camera may be offline): " + recreate.Err.Error())
+	case errors.As(e, &renew):
+		log.Log.Debug("onvif.HandleONVIFEventStream(): renew failed (will recover via pull/recreate): " + renew.Err.Error())
+	case errors.As(e, &pull):
+		log.Log.Debug("onvif.HandleONVIFEventStream(): pull failed (will retry): " + pull.Err.Error())
+	default:
+		log.Log.Info("onvif.HandleONVIFEventStream(): stream error: " + e.Error())
+	}
+}
+
+func isONVIFMotionEnabled(v string) bool {
+	return strings.EqualFold(strings.TrimSpace(v), "true")
+}
+
+// resolveDeviceID falls back from operator-supplied name to ONVIF
+// endpoint to a constant placeholder so log lines always have
+// something to grep.
+func resolveDeviceID(configName, xaddr string) string {
+	if n := strings.TrimSpace(configName); n != "" {
+		return n
+	}
+	if x := strings.TrimSpace(xaddr); x != "" {
+		return x
+	}
+	return "unknown"
+}
+
+// sleepCtx returns false if ctx was cancelled, true if d elapsed.
+func sleepCtx(ctx context.Context, d time.Duration) bool {
+	t := time.NewTimer(d)
+	defer t.Stop()
+	select {
+	case <-ctx.Done():
+		return false
+	case <-t.C:
+		return true
+	}
+}

machinery/src/onvif/events_test.go

@@ -0,0 +1,375 @@
+package onvif
+
+import (
+	"bytes"
+	"context"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/kerberos-io/agent/machinery/src/models"
+	"github.com/kerberos-io/onvif/event/stream"
+	"github.com/sirupsen/logrus"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func makeConfig(recording, onvifMotion, name string) *models.Configuration {
+	return &models.Configuration{
+		Name: name,
+		Config: models.Config{
+			Capture: models.Capture{
+				Recording:   recording,
+				ONVIFMotion: onvifMotion,
+			},
+		},
+	}
+}
+
+func makeCommunication(buffer int) *models.Communication {
+	return &models.Communication{
+		HandleMotion: make(chan models.MotionDataPartial, buffer),
+	}
+}
+
+// --- dispatchEvent ---------------------------------------------------
+
+func TestDispatchEvent_MotionActive_SendsToHandleMotion(t *testing.T) {
+	cfg := makeConfig("true", "true", "cam-1")
+	comm := makeCommunication(1)
+	ev := stream.Event{Kind: stream.KindMotion, State: stream.StateActive}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	dispatchEvent(ctx, ev, cfg, comm)
+
+	select {
+	case m := <-comm.HandleMotion:
+		assert.NotZero(t, m.Timestamp)
+	case <-time.After(time.Second):
+		t.Fatal("expected motion data on HandleMotion")
+	}
+}
+
+func TestDispatchEvent_MotionInactive_DoesNotSend(t *testing.T) {
+	cfg := makeConfig("true", "true", "cam-1")
+	comm := makeCommunication(1)
+	ev := stream.Event{Kind: stream.KindMotion, State: stream.StateInactive}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	dispatchEvent(ctx, ev, cfg, comm)
+
+	select {
+	case <-comm.HandleMotion:
+		t.Fatal("inactive motion must not reach HandleMotion (motion-stop is a follow-up)")
+	case <-time.After(100 * time.Millisecond):
+	}
+}
+
+func TestDispatchEvent_NonMotionKindIgnored(t *testing.T) {
+	cfg := makeConfig("true", "true", "cam-1")
+	comm := makeCommunication(1)
+	ev := stream.Event{Kind: stream.KindDigitalInput, State: stream.StateActive}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	dispatchEvent(ctx, ev, cfg, comm)
+
+	select {
+	case <-comm.HandleMotion:
+		t.Fatal("non-motion kinds must not reach HandleMotion")
+	case <-time.After(100 * time.Millisecond):
+	}
+}
+
+// captureDebugLog redirects logrus to a buffer at debug level for the
+// duration of a test and returns what was written. It mutates package
+// globals, so callers must not run in parallel.
+func captureDebugLog(t *testing.T) *bytes.Buffer {
+	t.Helper()
+	var buf bytes.Buffer
+	prevOut, prevLevel := logrus.StandardLogger().Out, logrus.GetLevel()
+	logrus.SetOutput(&buf)
+	logrus.SetLevel(logrus.DebugLevel)
+	t.Cleanup(func() {
+		logrus.SetOutput(prevOut)
+		logrus.SetLevel(prevLevel)
+	})
+	return &buf
+}
+
+// TestDispatchEvent_LogsTheTriggeringTopic — a dispatched event is what
+// actually starts a recording, so its topic is the one an operator needs
+// when a camera records for the wrong reason (or the right reason and
+// nobody can prove which). Rejected events were already logged; without
+// this the triggering topic is only knowable by elimination.
+func TestDispatchEvent_LogsTheTriggeringTopic(t *testing.T) {
+	buf := captureDebugLog(t)
+
+	cfg := makeConfig("true", "true", "cam-1")
+	comm := makeCommunication(1)
+	ev := stream.Event{
+		Kind:  stream.KindMotion,
+		State: stream.StateActive,
+		Topic: "tns1:RuleEngine/tnsaxis:VMD3/vmd3_video_1",
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	dispatchEvent(ctx, ev, cfg, comm)
+
+	assert.Contains(t, buf.String(), "tns1:RuleEngine/tnsaxis:VMD3/vmd3_video_1",
+		"the dispatched event's topic must appear in the log")
+	assert.Contains(t, buf.String(), "Motion",
+		"the dispatched event's Kind must appear in the log")
+}
+
+// TestDispatchEvent_PropertyOperation — a camera replays the current
+// state of every property topic as Initialized whenever a pull-point
+// subscription is created. If that counts as a trigger, every
+// reconnect restarts a recording for any motion property that happens
+// to be active, and a flapping subscription manufactures motion out of
+// nothing. Only reject Initialized specifically: PropertyOperation is
+// optional per WS-Notification and absent on many non-property events,
+// which decode reports as PropertyUnknown.
+func TestDispatchEvent_PropertyOperation(t *testing.T) {
+	tests := []struct {
+		name     string
+		op       stream.PropertyOperation
+		wantSend bool
+	}{
+		{"changed is a real transition", stream.PropertyChanged, true},
+		{"absent attribute still counts", stream.PropertyUnknown, true},
+		{"initialized is a subscription state replay", stream.PropertyInitialized, false},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			cfg := makeConfig("true", "true", "cam-1")
+			comm := makeCommunication(1)
+			ev := stream.Event{
+				Kind:      stream.KindMotion,
+				State:     stream.StateActive,
+				Operation: tt.op,
+				Topic:     "tns1:RuleEngine/tnsaxis:VMD3/vmd3_video_1",
+			}
+
+			ctx, cancel := context.WithCancel(context.Background())
+			defer cancel()
+			dispatchEvent(ctx, ev, cfg, comm)
+
+			select {
+			case <-comm.HandleMotion:
+				if !tt.wantSend {
+					t.Fatalf("%v must not trigger a recording", tt.op)
+				}
+			case <-time.After(100 * time.Millisecond):
+				if tt.wantSend {
+					t.Fatalf("%v must trigger a recording", tt.op)
+				}
+			}
+		})
+	}
+}
+
+func TestDispatchEvent_RecordingDisabled_DoesNotSend(t *testing.T) {
+	cfg := makeConfig("false", "true", "cam-1")
+	comm := makeCommunication(1)
+	ev := stream.Event{Kind: stream.KindMotion, State: stream.StateActive}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	dispatchEvent(ctx, ev, cfg, comm)
+
+	select {
+	case <-comm.HandleMotion:
+		t.Fatal("Recording=false must gate the send (matches computervision behaviour)")
+	case <-time.After(100 * time.Millisecond):
+	}
+}
+
+func TestDispatchEvent_HandleMotionFull_DropsRatherThanBlocks(t *testing.T) {
+	cfg := makeConfig("true", "true", "cam-1")
+	// Pre-fill the buffer so the next send would block.
+	comm := &models.Communication{HandleMotion: make(chan models.MotionDataPartial, 1)}
+	comm.HandleMotion <- models.MotionDataPartial{}
+	ev := stream.Event{Kind: stream.KindMotion, State: stream.StateActive}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	done := make(chan struct{})
+	go func() {
+		dispatchEvent(ctx, ev, cfg, comm)
+		close(done)
+	}()
+
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		t.Fatal("dispatchEvent must drop when HandleMotion is full, not block")
+	}
+}
+
+func TestDispatchEvent_CtxCancelledAndHandleMotionClosed_DoesNotPanic(t *testing.T) {
+	// Regression for the shutdown race: between cancel() and
+	// close(HandleMotion) the agent leaves a 3s window. If dispatchEvent
+	// runs in that window AFTER the channel is closed, a non-protected
+	// send would panic. The ctx pre-check must short-circuit before the
+	// send is attempted.
+	cfg := makeConfig("true", "true", "cam-1")
+	comm := &models.Communication{HandleMotion: make(chan models.MotionDataPartial, 1)}
+	close(comm.HandleMotion)
+	ev := stream.Event{Kind: stream.KindMotion, State: stream.StateActive}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel() // already cancelled, matching the shutdown sequence
+
+	assert.NotPanics(t, func() {
+		dispatchEvent(ctx, ev, cfg, comm)
+	})
+}
+
+// --- isONVIFMotionEnabled --------------------------------------------
+
+func TestIsONVIFMotionEnabled_CaseAndWhitespace(t *testing.T) {
+	tests := []struct {
+		in   string
+		want bool
+	}{
+		{"true", true},
+		{"True", true},
+		{"TRUE", true},
+		{" true", true},
+		{"true ", true},
+		{"  true  ", true},
+		{"false", false},
+		{"False", false},
+		{"", false},
+		{"yes", false},
+		{"1", false},
+	}
+	for _, tc := range tests {
+		t.Run(tc.in, func(t *testing.T) {
+			assert.Equal(t, tc.want, isONVIFMotionEnabled(tc.in))
+		})
+	}
+}
+
+// --- resolveDeviceID -------------------------------------------------
+
+func TestResolveDeviceID_FallbackChain(t *testing.T) {
+	tests := []struct {
+		name    string
+		cfgName string
+		xaddr   string
+		want    string
+	}{
+		{"name_set", "front-door", "192.168.1.10", "front-door"},
+		{"name_empty_xaddr_set", "", "192.168.1.10", "192.168.1.10"},
+		{"name_whitespace_only_xaddr_set", "  ", "192.168.1.10", "192.168.1.10"},
+		{"both_empty", "", "", "unknown"},
+		{"name_with_trailing_whitespace", "cam-2  ", "192.168.1.10", "cam-2"},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			assert.Equal(t, tc.want, resolveDeviceID(tc.cfgName, tc.xaddr))
+		})
+	}
+}
+
+// TestDispatchEvent_OnlyRealTransitionsTrigger — a camera replays every
+// property topic's state on each new subscription (Initialized) and
+// announces removals (Deleted). Neither is a motion transition, and a
+// flapping pull-point would otherwise manufacture recordings out of
+// replayed state. PropertyOperation is optional per WS-Notification, so
+// absent (Unknown) still counts — many non-property events omit it.
+func TestDispatchEvent_OnlyRealTransitionsTrigger(t *testing.T) {
+	tests := []struct {
+		op       stream.PropertyOperation
+		wantSend bool
+	}{
+		{stream.PropertyChanged, true},
+		{stream.PropertyUnknown, true},
+		{stream.PropertyInitialized, false},
+		{stream.PropertyDeleted, false},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.op.String(), func(t *testing.T) {
+			cfg := makeConfig("true", "true", "cam-1")
+			comm := makeCommunication(1)
+			ev := stream.Event{Kind: stream.KindMotion, State: stream.StateActive, Operation: tt.op}
+
+			ctx, cancel := context.WithCancel(context.Background())
+			defer cancel()
+			dispatchEvent(ctx, ev, cfg, comm)
+
+			if tt.wantSend {
+				require.Len(t, comm.HandleMotion, 1, "%v must trigger a recording", tt.op)
+				return
+			}
+			require.Empty(t, comm.HandleMotion, "%v must not trigger a recording", tt.op)
+		})
+	}
+}
+
+// TestSanitiseTopic — ev.Topic is camera-controlled and reaches the log
+// unmodified. logrus's coloured text formatter (the default) writes the
+// message without quoting, so an embedded newline forges whole log
+// lines: a compromised camera can fabricate ERROR entries or spoof
+// another device's id, in the logs an operator is reading to diagnose
+// that very camera. Length is also unbounded on the wire, and the
+// reject path logs every event, so an oversized topic is a cheap way to
+// evict a container's whole retained history.
+func TestSanitiseTopic(t *testing.T) {
+	tests := []struct {
+		name string
+		in   string
+		want string
+	}{
+		{"ordinary topic passes through", "tns1:RuleEngine/tnsaxis:VMD3/vmd3_video_1", "tns1:RuleEngine/tnsaxis:VMD3/vmd3_video_1"},
+		{"newline cannot forge a line", "a\nERRO[fake] boom", `a\nERRO[fake] boom`},
+		{"carriage return", "a\rb", `a\rb`},
+		{"tab", "a\tb", `a\tb`},
+		{"NUL", "a\x00b", `a\x00b`},
+		{"empty", "", ""},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := sanitiseTopic(tt.in)
+			assert.Equal(t, tt.want, got)
+			assert.NotContains(t, got, "\n", "no raw newline may survive")
+			assert.NotContains(t, got, "\r", "no raw carriage return may survive")
+		})
+	}
+}
+
+func TestSanitiseTopic_Truncates(t *testing.T) {
+	got := sanitiseTopic(strings.Repeat("x", maxLoggedTopic*2))
+	assert.LessOrEqual(t, len(got), maxLoggedTopic+len("…(truncated)"))
+	assert.Contains(t, got, "truncated")
+}
+
+// TestDispatchEvent_LogsTriggerOnlyWhenSent — the trigger line is the
+// record that a recording started. Logging it before the send means a
+// dropped event (full channel, or shutdown) leaves a line claiming a
+// recording that never began.
+func TestDispatchEvent_LogsTriggerOnlyWhenSent(t *testing.T) {
+	buf := captureDebugLog(t)
+
+	cfg := makeConfig("true", "true", "cam-1")
+	comm := &models.Communication{HandleMotion: make(chan models.MotionDataPartial, 1)}
+	comm.HandleMotion <- models.MotionDataPartial{} // full
+	ev := stream.Event{Kind: stream.KindMotion, State: stream.StateActive, Topic: "tns1:VideoSource/MotionAlarm"}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	dispatchEvent(ctx, ev, cfg, comm)
+
+	assert.NotContains(t, buf.String(), "recording trigger",
+		"a dropped event must not be logged as a trigger")
+	assert.Contains(t, buf.String(), "dropping", "the drop itself must still be logged")
+}

machinery/src/video/mp4.go

@@ -283,6 +283,8 @@ func (mp4 *MP4) flushPendingVideoSample(nextPTS uint64) bool {
 	err := mp4.MultiTrackFragment.AddFullSampleToTrack(*mp4.VideoFullSample, uint32(mp4.VideoTrack))
 	if err != nil {
 		log.Log.Error("mp4.flushPendingVideoSample(): error adding sample: " + err.Error())
+	} else {
+		mp4.SampleCount++
 	}
 	if isKF {
 		mp4.TotalKeyframesWritten++
@@ -296,6 +298,15 @@ func (mp4 *MP4) flushPendingVideoSample(nextPTS uint64) bool {
 	return true
 }
 
+// AverageFPS returns the average frame rate of the video samples actually
+// committed to this recording.
+func (mp4 *MP4) AverageFPS() float64 {
+	if mp4.SampleCount == 0 || mp4.VideoTotalDuration == 0 {
+		return 0
+	}
+	return float64(mp4.SampleCount) * 1000 / float64(mp4.VideoTotalDuration)
+}
+
 // AddSampleToTrack appends a sample to the given track.
 //
 // For video, pts is the decode timestamp (DTS, in milliseconds) and

machinery/src/video/mp4_duration_test.go

@@ -2,6 +2,7 @@ package video
 
 import (
 	"fmt"
+	"math"
 	"os"
 	"testing"
 
@@ -173,4 +174,10 @@ func TestMP4Duration(t *testing.T) {
 		t.Errorf("MISMATCH: mdhd.Duration should be 0 for fragmented MP4, got %d",
 			parsedFile.Moov.Traks[0].Mdia.Mdhd.Duration)
 	}
+	if mp4Video.SampleCount != sampleCount {
+		t.Errorf("SampleCount = %d, finalized MP4 contains %d video samples", mp4Video.SampleCount, sampleCount)
+	}
+	if fps := mp4Video.AverageFPS(); math.Abs(fps-25) > 0.001 {
+		t.Errorf("AverageFPS() = %.3f, want 25", fps)
+	}
 }