Merge pull request #313 from kerberos-io/fix/moq-recovery-strategy

fix/moq-recovery-strategy

.devcontainer/Dockerfile

@@ -1,4 +1,4 @@
-FROM mcr.microsoft.com/devcontainers/go:1.24-bookworm
+FROM mcr.microsoft.com/devcontainers/go:1.24-trixie
 
 # Install node environment
 RUN apt-get update && \

.devcontainer/devcontainer.json

@@ -1,7 +1,7 @@
 // For format details, see https://aka.ms/devcontainer.json. For config options, see the
 // README at: https://github.com/devcontainers/templates/tree/main/src/python
 {
-	"name": "go:1.24-bookworm",
+	"name": "go:1.24-trixie",
 	"runArgs": [
 		"--name=agent",
 		"--network=host"
@@ -20,5 +20,5 @@
 		3000,
 		8080
 	],
-	"postCreateCommand": "cd ui && yarn install && yarn build && cd ../machinery && go mod download"
+	"postCreateCommand": "cd ui && yarn install && yarn build && cd ../machinery && go mod download && bash ./verify-moq-devcontainer.sh"
 }

.github/workflows/pr-build.yml

@@ -16,10 +16,8 @@ jobs:
         include:
           - architecture: amd64
             runner: ubuntu-24.04
-            dockerfile: Dockerfile
           - architecture: arm64
             runner: ubuntu-24.04-arm
-            dockerfile: Dockerfile.arm64
     steps:
       - name: Checkout
         uses: actions/checkout@v3
@@ -34,7 +32,7 @@ jobs:
           length: 7
       - name: Run Build
         run: |
-          docker build -t ${{ matrix.architecture }} -f ${{ matrix.dockerfile }} .
+          docker build -t ${{ matrix.architecture }} .
           CID=$(docker create ${{matrix.architecture}})
           docker cp ${CID}:/home/agent ./output-${{matrix.architecture}}
           docker rm ${CID}

.github/workflows/release-bump.yml

@@ -108,7 +108,7 @@ jobs:
           length: 7
       - name: Run Build
         run: |
-          docker build --provenance=false --build-arg VERSION=${{ needs.bump-release.outputs.tag }} -t ${{matrix.architecture}} -f Dockerfile.arm64 .
+          docker build --provenance=false --build-arg VERSION=${{ needs.bump-release.outputs.tag }} -t ${{matrix.architecture}} .
           CID=$(docker create ${{matrix.architecture}})
           docker cp ${CID}:/home/agent ./output-${{matrix.architecture}}
           docker rm ${CID}

.github/workflows/release-create.yml

@@ -71,7 +71,7 @@ jobs:
           length: 7
       - name: Run Build
         run: |
-          docker build --provenance=false --build-arg VERSION=${{github.event.inputs.tag || github.ref_name}} -t ${{matrix.architecture}} -f Dockerfile.arm64 .
+          docker build --provenance=false --build-arg VERSION=${{github.event.inputs.tag || github.ref_name}} -t ${{matrix.architecture}} .
           CID=$(docker create ${{matrix.architecture}})
           docker cp ${CID}:/home/agent ./output-${{matrix.architecture}}
           docker rm ${CID}

.vscode/tasks.json

@@ -0,0 +1,14 @@
+{
+  "version": "2.0.0",
+  "tasks": [
+    {
+      "label": "agent: moq verify",
+      "type": "shell",
+      "command": "bash ./verify-moq-devcontainer.sh",
+      "options": {
+        "cwd": "${workspaceFolder}/machinery"
+      },
+      "problemMatcher": []
+    }
+  ]
+}

Dockerfile

@@ -1,13 +1,15 @@
 
-ARG BASE_IMAGE_VERSION=amd64-ddbe40e
+ARG GO_IMAGE=golang:1.24-trixie
+ARG RUNTIME_IMAGE=debian:trixie-slim
 ARG VERSION=0.0.0
-FROM kerberos/base:${BASE_IMAGE_VERSION} AS build-machinery
+FROM ${GO_IMAGE} AS build-machinery
 LABEL AUTHOR=uug.ai
 
 # Re-declare VERSION inside this stage so the value passed via
 # `--build-arg VERSION=...` (e.g. the release tag) is available below.
 # ARGs declared before the first FROM are not visible inside build stages.
 ARG VERSION
+ARG TARGETARCH
 
 ENV GOROOT=/usr/local/go
 ENV GOPATH=/go
@@ -17,9 +19,10 @@ ENV GOSUMDB=off
 ##########################################
 # Installing some additional dependencies.
 
-RUN apt-get upgrade -y && apt-get update && apt-get install -y --fix-missing --no-install-recommends \
+RUN apt-get update && apt-get install -y --fix-missing --no-install-recommends \
 	git build-essential cmake pkg-config unzip libgtk2.0-dev \
-	curl ca-certificates libcurl4-openssl-dev libssl-dev libjpeg62-turbo-dev && \
+	curl ca-certificates libavcodec-dev libavutil-dev libcurl4-openssl-dev \
+	libssl-dev libjpeg62-turbo-dev libswscale-dev && \
 	rm -rf /var/lib/apt/lists/*
 
 ##############################################################################
@@ -43,7 +46,9 @@ RUN cd /go/src/github.com/kerberos-io/agent/machinery && \
 	if [ -z "${VERSION}" ] || [ "${VERSION}" = "0.0.0" ]; then \
 		VERSION=$(cd /go/src/github.com/kerberos-io/agent && git describe --tags --always 2>/dev/null || echo "0.0.0"); \
 	fi && \
-	go build -tags timetzdata,netgo,osusergo --ldflags "-s -w -X github.com/kerberos-io/agent/machinery/src/utils.VERSION=${VERSION} -extldflags '-static -latomic'" main.go && \
+	BUILD_TAGS=timetzdata,netgo,osusergo && \
+	case "${TARGETARCH:-$(go env GOARCH)}" in amd64|arm64) BUILD_TAGS="moq,${BUILD_TAGS}" ;; esac && \
+	go build -tags "${BUILD_TAGS}" --ldflags "-s -w -X github.com/kerberos-io/agent/machinery/src/utils.VERSION=${VERSION}" main.go && \
 	mkdir -p /agent && \
 	mv main /agent && \
 	mv version /agent && \
@@ -89,12 +94,16 @@ RUN mkdir -p ./agent && cp -r /go/src/github.com/kerberos-io/agent/machinery/www
 ############################################
 # Publish main binary to GitHub release
 
-FROM alpine:latest
+FROM ${RUNTIME_IMAGE}
 
 ############################
 # Protect by non-root user.
 
-RUN addgroup -S kerberosio && adduser -S agent -G kerberosio && addgroup agent video
+RUN apt-get update && apt-get install -y --no-install-recommends \
+	ca-certificates curl ffmpeg libatomic1 libcap2-bin libstdc++6 && \
+	rm -rf /var/lib/apt/lists/* && \
+	groupadd --system kerberosio && \
+	useradd --system --gid kerberosio --groups video --create-home agent
 
 #################################
 # Copy files from previous images
@@ -102,8 +111,6 @@ RUN addgroup -S kerberosio && adduser -S agent -G kerberosio && addgroup agent v
 COPY --chown=0:0 --from=build-machinery /dist /
 COPY --chown=0:0 --from=build-ui /dist /
 
-RUN apk update && apk add ca-certificates curl ffmpeg libstdc++ libc6-compat --no-cache && rm -rf /var/cache/apk/*
-
 ##################
 # Try running agent
 
@@ -123,7 +130,7 @@ RUN chown -R agent:kerberosio /home/agent/www
 
 ###########################
 # Grant the necessary root capabilities to the process trying to bind to the privileged port
-RUN apk add libcap && setcap 'cap_net_bind_service=+ep' /home/agent/main
+RUN setcap 'cap_net_bind_service=+ep' /home/agent/main
 
 ###################
 # Run non-root user

Dockerfile.arm64

@@ -1,147 +0,0 @@
-
-ARG BASE_IMAGE_VERSION=arm64-ddbe40e
-ARG VERSION=0.0.0
-FROM kerberos/base:${BASE_IMAGE_VERSION} AS build-machinery
-LABEL AUTHOR=uug.ai
-
-# Re-declare VERSION inside this stage so the value passed via
-# `--build-arg VERSION=...` (e.g. the release tag) is available below.
-# ARGs declared before the first FROM are not visible inside build stages.
-ARG VERSION
-
-ENV GOROOT=/usr/local/go
-ENV GOPATH=/go
-ENV PATH=$GOPATH/bin:$GOROOT/bin:/usr/local/lib:$PATH
-ENV GOSUMDB=off
-
-##########################################
-# Installing some additional dependencies.
-
-RUN apt-get upgrade -y && apt-get update && apt-get install -y --fix-missing --no-install-recommends \
-	git build-essential cmake pkg-config unzip libgtk2.0-dev \
-	curl ca-certificates libcurl4-openssl-dev libssl-dev libjpeg62-turbo-dev && \
-	rm -rf /var/lib/apt/lists/*
-
-##############################################################################
-# Copy all the relevant source code in the Docker image, so we can build this.
-
-RUN mkdir -p /go/src/github.com/kerberos-io/agent
-COPY machinery /go/src/github.com/kerberos-io/agent/machinery
-RUN rm -rf /go/src/github.com/kerberos-io/agent/machinery/.env
-
-##################################################################
-# Get the latest commit hash, so we know which version we're running
-COPY .git /go/src/github.com/kerberos-io/agent/.git
-RUN cd /go/src/github.com/kerberos-io/agent/.git && git log --format="%H" -n 1 | head -c7 > /go/src/github.com/kerberos-io/agent/machinery/version
-RUN cat /go/src/github.com/kerberos-io/agent/machinery/version
-
-##################
-# Build Machinery
-
-RUN cd /go/src/github.com/kerberos-io/agent/machinery && \
-	go mod download && \
-	if [ -z "${VERSION}" ] || [ "${VERSION}" = "0.0.0" ]; then \
-		VERSION=$(cd /go/src/github.com/kerberos-io/agent && git describe --tags --always 2>/dev/null || echo "0.0.0"); \
-	fi && \
-	go build -tags timetzdata,netgo,osusergo --ldflags "-s -w -X github.com/kerberos-io/agent/machinery/src/utils.VERSION=${VERSION} -extldflags '-static -latomic'" main.go && \
-	mkdir -p /agent && \
-	mv main /agent && \
-	mv version /agent && \
-	mv data /agent && \
-	mkdir -p /agent/data/cloud && \
-	mkdir -p /agent/data/snapshots && \
-	mkdir -p /agent/data/log && \
-	mkdir -p /agent/data/recordings && \
-	mkdir -p /agent/data/capture-test && \
-	mkdir -p /agent/data/config
-
-####################################
-# Let's create a /dist folder containing just the files necessary for runtime.
-# Later, it will be copied as the / (root) of the output image.
-
-WORKDIR /dist
-RUN cp -r /agent ./
-
-####################################################################################
-# This will collect dependent libraries so they're later copied to the final image.
-
-RUN /dist/agent/main version
-
-FROM node:22-alpine AS build-ui
-
-RUN apk update && apk upgrade --available && sync
-
-########################
-# Build Web (React app)
-
-RUN mkdir -p /go/src/github.com/kerberos-io/agent/machinery/www
-COPY ui /go/src/github.com/kerberos-io/agent/ui
-RUN cd /go/src/github.com/kerberos-io/agent/ui && rm -rf yarn.lock && yarn config set network-timeout 300000 && \
-	yarn && yarn build
-
-####################################
-# Let's create a /dist folder containing just the files necessary for runtime.
-# Later, it will be copied as the / (root) of the output image.
-
-WORKDIR /dist
-RUN mkdir -p ./agent && cp -r /go/src/github.com/kerberos-io/agent/machinery/www ./agent/
-
-############################################
-# Publish main binary to GitHub release
-
-FROM alpine:latest
-
-############################
-# Protect by non-root user.
-
-RUN addgroup -S kerberosio && adduser -S agent -G kerberosio && addgroup agent video
-
-#################################
-# Copy files from previous images
-
-COPY --chown=0:0 --from=build-machinery /dist /
-COPY --chown=0:0 --from=build-ui /dist /
-
-RUN apk update && apk add ca-certificates curl ffmpeg libstdc++ libc6-compat --no-cache && rm -rf /var/cache/apk/*
-
-##################
-# Try running agent
-
-RUN mv /agent/* /home/agent/
-RUN /home/agent/main version
-
-#######################
-# Make template config
-
-RUN cp /home/agent/data/config/config.json /home/agent/data/config.template.json
-
-###########################
-# Set permissions correctly
-
-RUN chown -R agent:kerberosio /home/agent/data
-RUN chown -R agent:kerberosio /home/agent/www
-
-###########################
-# Grant the necessary root capabilities to the process trying to bind to the privileged port
-RUN apk add libcap && setcap 'cap_net_bind_service=+ep' /home/agent/main
-
-###################
-# Run non-root user
-
-USER agent
-
-######################################
-# By default the app runs on port 80
-
-EXPOSE 80
-
-######################################
-# Check if agent is still running
-
-HEALTHCHECK CMD curl --fail http://localhost:80 || exit 1   
-
-###################################################
-# Leeeeettttt'ssss goooooo!!!
-# Run the shizzle from the right working directory.
-WORKDIR /home/agent
-CMD ["./main", "-action", "run", "-port", "80"]

README.md

@@ -412,13 +412,53 @@ Remember the build step of the `web` part, during build time we move the build d
 
 ## Building for Docker
 
-Inside the root of this `agent` repository, you will find a `Dockerfile`. This file contains the instructions for building and shipping a **Kerberos Agent**. Important to note is that you start from a prebuilt base image, `kerberos/base:xxx`.
-This base image already contains a couple of tools, such as Golang, FFmpeg and OpenCV. We do this for faster compilation times.
+Inside the root of this `agent` repository, you will find a `Dockerfile`. This file contains the instructions for building and shipping a **Kerberos Agent**. It uses Debian Trixie to support the native dependencies used by the Agent, including Media over QUIC.
 
 By running the `docker build` command, you will create the Kerberos Agent Docker image. After building you can simply run the image as a Docker container.
 
     docker build -t kerberos/agent .
 
+### Media over QUIC
+
+The standard AMD64 and ARM64 images include the optional MoQ publisher. Its Rust
+FFI archive requires CGO and glibc 2.38 or newer, which is why the standard image
+uses Debian Trixie. The publisher is disabled unless explicitly enabled at runtime:
+
+    docker run --rm -p 80:80 \
+      -e AGENT_LIVE_MOQ_ENABLED=true \
+      -e AGENT_LIVE_MOQ_URL=https://relay.uug.ai/anon \
+      kerberos/agent
+
+`AGENT_LIVE_MOQ_BROADCAST_PREFIX` defaults to `devices`. MoQ viewers subscribe to
+a relay and never negotiate with the Agent, so every quality tier is published as
+its own broadcast and switching quality is simply a resubscribe:
+
+| Tier   | Broadcast                             | Source                                     |
+| ------ | ------------------------------------- | ------------------------------------------ |
+| `high` | `devices/<agent-key>/live.hang`       | highest-resolution camera stream           |
+| `low`  | `devices/<agent-key>/live-low.hang`   | sub stream (main stream when none is set)  |
+
+Each tier only uploads while it has at least one subscriber, so the tier nobody
+watches costs virtually no bandwidth. `AGENT_LIVE_MOQ_QUALITY` accepts `high` or
+`low` to pin the Agent to a single tier; viewers requesting the other tier then
+find no broadcast. Any other value (including the default) publishes both. The
+initial implementation publishes H.264 video only.
+
+The `/anon` relay route is intended for interoperability testing. Production
+deployments must set `AGENT_LIVE_MOQ_URL` to a short-lived, device-scoped
+publisher URL issued by Hub API.
+
+To verify the native SDK in a development container, rebuild the Agent or shared
+monorepo devcontainer so it uses the Trixie base, then run the VS Code task
+`agent: moq verify`. The same check is available from a terminal:
+
+    cd machinery
+    bash ./verify-moq-devcontainer.sh
+
+The check requires glibc 2.38 or newer, runs the tagged package tests, links the
+complete Agent with `-tags moq`, and executes the resulting binary's version
+command. Both devcontainers also run this check during their post-create setup.
+
 ## What is new?
 
 This repository contains the next generation of Kerberos.io, **Kerberos Agent (v3)**, and is the successor of the machinery and web repositories. A switch in technologies and architecture has been made. This version is still under active development and can be followed on the [develop branch](https://github.com/kerberos-io/agent/tree/develop) and [project overview](https://github.com/kerberos-io/agent/projects/1).

machinery/go.mod

@@ -24,8 +24,9 @@ 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/moq-dev/moq-go v0.5.7
 	github.com/nfnt/resize v0.0.0-20180221191011-83c6a9932646
 	github.com/op/go-logging v0.0.0-20160315200505-970db520ece7
 	github.com/pion/interceptor v0.1.40
@@ -95,6 +96,7 @@ require (
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/montanaflynn/stats v0.7.1 // indirect
+	github.com/moq-dev/moq-go-ffi v0.3.7 // indirect
 	github.com/nxadm/tail v1.4.11 // indirect
 	github.com/pelletier/go-toml/v2 v2.2.3 // indirect
 	github.com/pion/datachannel v1.5.10 // 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=
@@ -845,6 +845,10 @@ github.com/modern-go/reflect2 v1.0.2 h1:xBagoLtFs94CBntxluKeaWgTMpvLxC4ur3nMaC9G
 github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk=
 github.com/montanaflynn/stats v0.7.1 h1:etflOAAHORrCC44V+aR6Ftzort912ZU+YLiSTuV8eaE=
 github.com/montanaflynn/stats v0.7.1/go.mod h1:etXPPgVO6n31NxCd9KQUMvCM+ve0ruNzt6R8Bnaayow=
+github.com/moq-dev/moq-go v0.5.7 h1:LfFpgAU8FRMcnU85L5Lb03HzxhYbQ2+BhT/pJQdjy5U=
+github.com/moq-dev/moq-go v0.5.7/go.mod h1:5K8zjKKjWe5lzfCtlfxoAfdCr3KX6EBuTOZsz0WBnzw=
+github.com/moq-dev/moq-go-ffi v0.3.7 h1:+xwPOzTJHvB0tuTnW6znbvvOgT4yLnN7sMqXXZYh02M=
+github.com/moq-dev/moq-go-ffi v0.3.7/go.mod h1:zxpOlUetvaoxWBnbXTdILtLLUytaoVuLvO36lftwWO0=
 github.com/nfnt/resize v0.0.0-20180221191011-83c6a9932646 h1:zYyBkD/k9seD2A7fsi6Oo2LfFZAehjjQMERAvZLEDnQ=
 github.com/nfnt/resize v0.0.0-20180221191011-83c6a9932646/go.mod h1:jpp1/29i3P1S/RLdc7JQKbRpFeM1dOBd8T9ki5s+AY8=
 github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e/go.mod h1:zD1mROLANZcx1PVRCS0qkT7pwLkGfwJo4zjcN/Tysno=

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 = 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,64 @@
+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)
+	}
+	expectedFPS := mp4Video.AverageFPS()
+	if stored.FileName != "recording.mp4" || stored.DeviceKey != "device-key" || stored.Timestamp != 1785934709414 || stored.Duration != 20452 || math.Abs(stored.FPS-expectedFPS) > 1e-9 {
+		t.Fatalf("upload marker = %+v", stored)
+	}
+	if stored.FPS == math.Floor(stored.FPS) {
+		t.Fatalf("upload marker FPS = %v, want fractional precision", stored.FPS)
+	}
+}
+
+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 = 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())
 						}

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/livemoq/annexb.go

@@ -0,0 +1,103 @@
+package livemoq
+
+import (
+	"bytes"
+	"strings"
+
+	"github.com/bluenviron/mediacommon/pkg/codecs/h264"
+	"github.com/kerberos-io/agent/machinery/src/models"
+)
+
+var annexBStartCode = []byte{0x00, 0x00, 0x00, 0x01}
+
+// H264NormalizationStats reports malformed duplication removed from an access unit.
+type H264NormalizationStats struct {
+	DuplicateIDRNALUs int
+}
+
+// EnsureAnnexB restores the start code stripped by the Agent capture queue.
+func EnsureAnnexB(payload []byte) []byte {
+	if hasAnnexBStartCode(payload) {
+		return payload
+	}
+
+	framed := make([]byte, 0, len(annexBStartCode)+len(payload))
+	framed = append(framed, annexBStartCode...)
+	return append(framed, payload...)
+}
+
+// NormalizeH264AccessUnit removes delimiters and exact duplicate parameter-set
+// or IDR NALUs that can confuse older MoQ splitters and decoders.
+func NormalizeH264AccessUnit(payload []byte) ([]byte, error) {
+	normalized, _, err := NormalizeH264AccessUnitWithStats(payload)
+	return normalized, err
+}
+
+// NormalizeH264AccessUnitWithStats also reports exact duplicate IDR NALUs.
+func NormalizeH264AccessUnitWithStats(payload []byte) ([]byte, H264NormalizationStats, error) {
+	nalus, err := h264.AnnexBUnmarshal(EnsureAnnexB(payload))
+	if err != nil {
+		return nil, H264NormalizationStats{}, err
+	}
+
+	stats := H264NormalizationStats{}
+	normalized := make([][]byte, 0, len(nalus))
+	for _, nalu := range nalus {
+		if len(nalu) == 0 || nalu[0]&0x1f == 9 {
+			continue
+		}
+		naluType := nalu[0] & 0x1f
+		if naluType == 7 || naluType == 8 || naluType == 5 {
+			duplicate := false
+			for _, existing := range normalized {
+				if bytes.Equal(existing, nalu) {
+					duplicate = true
+					break
+				}
+			}
+			if duplicate {
+				if naluType == 5 {
+					stats.DuplicateIDRNALUs++
+				}
+				continue
+			}
+		}
+		normalized = append(normalized, nalu)
+	}
+
+	result, err := h264.AnnexBMarshal(normalized)
+	return result, stats, err
+}
+
+// BroadcastPath returns the relay path a quality tier is published on. Every
+// tier gets its own broadcast so a viewer switches between the camera's main and
+// sub stream by resubscribing to another path, without any control channel back
+// to the Agent. The high tier keeps the historical ".../live.hang" path so
+// existing viewers keep working; the low tier lives next to it on
+// ".../live-low.hang".
+func BroadcastPath(prefix string, deviceKey string, quality string) string {
+	prefix = strings.Trim(prefix, "/")
+	if prefix == "" {
+		prefix = "devices"
+	}
+	name := "live.hang"
+	if quality == models.StreamQualityLow {
+		name = "live-low.hang"
+	}
+	return prefix + "/" + strings.Trim(deviceKey, "/") + "/" + name
+}
+
+// TimestampUs converts the capture presentation timestamp from milliseconds.
+// CompositionTime must not be added: it is already represented in the PTS and
+// is only used by muxers to derive DTS for streams containing B-frames.
+func TimestampUs(presentationTimeMs int64) uint64 {
+	if presentationTimeMs < 0 {
+		return 0
+	}
+	return uint64(presentationTimeMs) * 1000
+}
+
+func hasAnnexBStartCode(payload []byte) bool {
+	return len(payload) >= 4 && payload[0] == 0 && payload[1] == 0 &&
+		((payload[2] == 0 && payload[3] == 1) || payload[2] == 1)
+}

machinery/src/cloud/livemoq/annexb_test.go

@@ -0,0 +1,116 @@
+package livemoq
+
+import (
+	"bytes"
+	"testing"
+
+	"github.com/kerberos-io/agent/machinery/src/models"
+)
+
+func TestEnsureAnnexB(t *testing.T) {
+	tests := []struct {
+		name    string
+		payload []byte
+		want    []byte
+	}{
+		{
+			name:    "missing start code",
+			payload: []byte{0x41, 0x01},
+			want:    []byte{0x00, 0x00, 0x00, 0x01, 0x41, 0x01},
+		},
+		{
+			name:    "four byte start code",
+			payload: []byte{0x00, 0x00, 0x00, 0x01, 0x65},
+			want:    []byte{0x00, 0x00, 0x00, 0x01, 0x65},
+		},
+		{
+			name:    "three byte start code",
+			payload: []byte{0x00, 0x00, 0x01, 0x41},
+			want:    []byte{0x00, 0x00, 0x01, 0x41},
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			if got := EnsureAnnexB(test.payload); !bytes.Equal(got, test.want) {
+				t.Fatalf("EnsureAnnexB() = %x, want %x", got, test.want)
+			}
+		})
+	}
+}
+
+func TestNormalizeH264AccessUnit(t *testing.T) {
+	startCode := []byte{0x00, 0x00, 0x00, 0x01}
+	sps := []byte{0x67, 0x42, 0x00, 0x1f}
+	pps := []byte{0x68, 0xce, 0x06, 0xe2}
+	aud := []byte{0x09, 0xf0}
+	idr := []byte{0x65, 0x88, 0x84}
+
+	payload := make([]byte, 0)
+	for _, nalu := range [][]byte{sps, pps, aud, sps, pps, idr} {
+		payload = append(payload, startCode...)
+		payload = append(payload, nalu...)
+	}
+
+	got, err := NormalizeH264AccessUnit(payload)
+	if err != nil {
+		t.Fatal(err)
+	}
+	want := make([]byte, 0)
+	for _, nalu := range [][]byte{sps, pps, idr} {
+		want = append(want, startCode...)
+		want = append(want, nalu...)
+	}
+	if !bytes.Equal(got, want) {
+		t.Fatalf("NormalizeH264AccessUnit() = %x, want %x", got, want)
+	}
+}
+
+func TestNormalizeH264AccessUnitRemovesOnlyExactDuplicateIDRSlices(t *testing.T) {
+	startCode := []byte{0x00, 0x00, 0x00, 0x01}
+	idrSlice1 := []byte{0x65, 0x88, 0x84}
+	idrSlice2 := []byte{0x65, 0x44, 0x22}
+
+	payload := make([]byte, 0)
+	for _, nalu := range [][]byte{idrSlice1, idrSlice1, idrSlice2} {
+		payload = append(payload, startCode...)
+		payload = append(payload, nalu...)
+	}
+
+	got, stats, err := NormalizeH264AccessUnitWithStats(payload)
+	if err != nil {
+		t.Fatal(err)
+	}
+	want := make([]byte, 0)
+	for _, nalu := range [][]byte{idrSlice1, idrSlice2} {
+		want = append(want, startCode...)
+		want = append(want, nalu...)
+	}
+	if !bytes.Equal(got, want) {
+		t.Fatalf("NormalizeH264AccessUnitWithStats() = %x, want %x", got, want)
+	}
+	if stats.DuplicateIDRNALUs != 1 {
+		t.Fatalf("DuplicateIDRNALUs = %d, want 1", stats.DuplicateIDRNALUs)
+	}
+}
+
+func TestBroadcastPath(t *testing.T) {
+	if got := BroadcastPath("/devices/", "/camera-1/", models.StreamQualityHigh); got != "devices/camera-1/live.hang" {
+		t.Fatalf("BroadcastPath() high = %q", got)
+	}
+	if got := BroadcastPath("", "camera-1", models.StreamQualityHigh); got != "devices/camera-1/live.hang" {
+		t.Fatalf("BroadcastPath() default = %q", got)
+	}
+	if got := BroadcastPath("", "camera-1", models.StreamQualityLow); got != "devices/camera-1/live-low.hang" {
+		t.Fatalf("BroadcastPath() low = %q", got)
+	}
+}
+
+func TestTimestampUs(t *testing.T) {
+	if got := TimestampUs(1234); got != 1_234_000 {
+		t.Fatalf("TimestampUs() = %d, want 1234000", got)
+	}
+	if got := TimestampUs(-1); got != 0 {
+		t.Fatalf("TimestampUs() negative = %d, want 0", got)
+	}
+}

machinery/src/cloud/livemoq/dedup.go

@@ -0,0 +1,41 @@
+package livemoq
+
+import (
+	"crypto/sha256"
+	"time"
+)
+
+type KeyframeDeduplicator struct {
+	hasPrevious  bool
+	timestampMs  int64
+	capturedAtMs int64
+	observedAt   time.Time
+	digest       [sha256.Size]byte
+}
+
+func (d *KeyframeDeduplicator) Reset() {
+	*d = KeyframeDeduplicator{}
+}
+
+// IsDuplicate reports exact repeated keyframe access units observed close
+// together. Distinct IDR slices within one access unit remain untouched.
+func (d *KeyframeDeduplicator) IsDuplicate(timestampMs int64, capturedAtMs int64, payload []byte, observedAt time.Time, window time.Duration) bool {
+	digest := sha256.Sum256(payload)
+	duplicate := d.hasPrevious && d.timestampMs == timestampMs && d.digest == digest
+	if duplicate {
+		if capturedAtMs > 0 && d.capturedAtMs > 0 {
+			gap := time.Duration(capturedAtMs-d.capturedAtMs) * time.Millisecond
+			duplicate = gap >= 0 && gap <= window
+		} else {
+			gap := observedAt.Sub(d.observedAt)
+			duplicate = gap >= 0 && gap <= window
+		}
+	}
+
+	d.hasPrevious = true
+	d.timestampMs = timestampMs
+	d.capturedAtMs = capturedAtMs
+	d.observedAt = observedAt
+	d.digest = digest
+	return duplicate
+}

machinery/src/cloud/livemoq/dedup_test.go

@@ -0,0 +1,60 @@
+package livemoq
+
+import (
+	"testing"
+	"time"
+)
+
+func TestKeyframeDeduplicator(t *testing.T) {
+	now := time.UnixMilli(10_000)
+	window := 500 * time.Millisecond
+	payload := []byte{0x00, 0x00, 0x00, 0x01, 0x65, 0x88}
+	deduplicator := KeyframeDeduplicator{}
+
+	if deduplicator.IsDuplicate(1_000, 10_000, payload, now, window) {
+		t.Fatal("first keyframe reported as duplicate")
+	}
+	if !deduplicator.IsDuplicate(1_000, 10_020, payload, now.Add(20*time.Millisecond), window) {
+		t.Fatal("exact repeated keyframe was not reported as duplicate")
+	}
+	if deduplicator.IsDuplicate(2_000, 11_000, payload, now.Add(time.Second), window) {
+		t.Fatal("same payload with a new timestamp reported as duplicate")
+	}
+	if deduplicator.IsDuplicate(2_000, 11_020, append(payload, 0x01), now.Add(1020*time.Millisecond), window) {
+		t.Fatal("different payload with the same timestamp reported as duplicate")
+	}
+}
+
+func TestKeyframeDeduplicatorAllowsTimestampReuseOutsideWindow(t *testing.T) {
+	now := time.UnixMilli(10_000)
+	payload := []byte{0x00, 0x00, 0x00, 0x01, 0x65, 0x88}
+	deduplicator := KeyframeDeduplicator{}
+
+	deduplicator.IsDuplicate(1_000, 10_000, payload, now, 500*time.Millisecond)
+	if deduplicator.IsDuplicate(1_000, 20_000, payload, now.Add(10*time.Second), 500*time.Millisecond) {
+		t.Fatal("later keyframe after timestamp reset reported as duplicate")
+	}
+}
+
+func TestKeyframeDeduplicatorFallsBackToObservationTime(t *testing.T) {
+	now := time.UnixMilli(10_000)
+	payload := []byte{0x65, 0x88}
+	deduplicator := KeyframeDeduplicator{}
+
+	deduplicator.IsDuplicate(1_000, 0, payload, now, 500*time.Millisecond)
+	if !deduplicator.IsDuplicate(1_000, 0, payload, now.Add(20*time.Millisecond), 500*time.Millisecond) {
+		t.Fatal("duplicate without capture time was not reported")
+	}
+}
+
+func TestKeyframeDeduplicatorReset(t *testing.T) {
+	now := time.UnixMilli(10_000)
+	payload := []byte{0x65, 0x88}
+	deduplicator := KeyframeDeduplicator{}
+
+	deduplicator.IsDuplicate(1_000, 10_000, payload, now, 500*time.Millisecond)
+	deduplicator.Reset()
+	if deduplicator.IsDuplicate(1_000, 10_020, payload, now.Add(20*time.Millisecond), 500*time.Millisecond) {
+		t.Fatal("first keyframe after reset reported as duplicate")
+	}
+}

machinery/src/cloud/livemoq/recovery.go

@@ -0,0 +1,55 @@
+package livemoq
+
+import "time"
+
+type FrameGateEvent uint8
+
+const (
+	FrameGateEventNone FrameGateEvent = iota
+	FrameGateEventStarted
+	FrameGateEventLagging
+	FrameGateEventRecovered
+)
+
+// FrameGate keeps publication on a decodable, recent GOP.
+type FrameGate struct {
+	started    bool
+	recovering bool
+}
+
+// Reset closes the gate so publication resumes on the next keyframe. It is used
+// when the publisher stopped writing for a reason unrelated to the stream health
+// (no subscribers), so the next viewer never receives a partial GOP.
+func (g *FrameGate) Reset() {
+	g.started = false
+	g.recovering = false
+}
+
+// Allow rejects stale frames and waits for a fresh keyframe before reopening.
+func (g *FrameGate) Allow(isKeyFrame bool, capturedAtMs int64, now time.Time, maxAge time.Duration) (bool, FrameGateEvent) {
+	if capturedAtMs > 0 && now.Sub(time.UnixMilli(capturedAtMs)) > maxAge {
+		event := FrameGateEventNone
+		if g.started {
+			if !g.recovering {
+				event = FrameGateEventLagging
+			}
+			g.started = false
+			g.recovering = true
+		}
+		return false, event
+	}
+
+	if !g.started {
+		if !isKeyFrame {
+			return false, FrameGateEventNone
+		}
+		g.started = true
+		if g.recovering {
+			g.recovering = false
+			return true, FrameGateEventRecovered
+		}
+		return true, FrameGateEventStarted
+	}
+
+	return true, FrameGateEventNone
+}

machinery/src/cloud/livemoq/recovery_test.go

@@ -0,0 +1,49 @@
+package livemoq
+
+import (
+	"testing"
+	"time"
+)
+
+func TestFrameGateRecoversAtFreshKeyframe(t *testing.T) {
+	now := time.UnixMilli(10_000)
+	maxAge := 1500 * time.Millisecond
+	gate := FrameGate{}
+
+	tests := []struct {
+		name         string
+		isKeyFrame   bool
+		capturedAtMs int64
+		wantAllowed  bool
+		wantEvent    FrameGateEvent
+	}{
+		{name: "waits for initial keyframe", capturedAtMs: 10_000},
+		{name: "starts at initial keyframe", isKeyFrame: true, capturedAtMs: 10_000, wantAllowed: true, wantEvent: FrameGateEventStarted},
+		{name: "publishes fresh delta", capturedAtMs: 10_020, wantAllowed: true},
+		{name: "detects stale packet", capturedAtMs: 8_000, wantEvent: FrameGateEventLagging},
+		{name: "rejects fresh delta while recovering", capturedAtMs: 10_040},
+		{name: "rejects stale keyframe without duplicate event", isKeyFrame: true, capturedAtMs: 8_000},
+		{name: "recovers at fresh keyframe", isKeyFrame: true, capturedAtMs: 10_060, wantAllowed: true, wantEvent: FrameGateEventRecovered},
+		{name: "publishes delta after recovery", capturedAtMs: 10_080, wantAllowed: true},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			allowed, event := gate.Allow(test.isKeyFrame, test.capturedAtMs, now, maxAge)
+			if allowed != test.wantAllowed {
+				t.Fatalf("Allow() allowed = %t, want %t", allowed, test.wantAllowed)
+			}
+			if event != test.wantEvent {
+				t.Fatalf("Allow() event = %d, want %d", event, test.wantEvent)
+			}
+		})
+	}
+}
+
+func TestFrameGateAllowsMissingCaptureTime(t *testing.T) {
+	gate := FrameGate{}
+	allowed, event := gate.Allow(true, 0, time.Now(), time.Second)
+	if !allowed || event != FrameGateEventStarted {
+		t.Fatalf("Allow() = (%t, %d), want (true, %d)", allowed, event, FrameGateEventStarted)
+	}
+}

machinery/src/cloud/livemoq_disabled.go

@@ -0,0 +1,8 @@
+//go:build !moq
+
+package cloud
+
+import "github.com/kerberos-io/agent/machinery/src/models"
+
+// StartLiveStreamMoQ is disabled in the standard Agent build.
+func StartLiveStreamMoQ(_ *models.Configuration, _ *models.Communication, _ bool) {}

machinery/src/cloud/livemoq_enabled.go

@@ -0,0 +1,285 @@
+//go:build moq
+
+package cloud
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"strings"
+	"sync/atomic"
+	"time"
+
+	"github.com/kerberos-io/agent/machinery/src/cloud/livemoq"
+	"github.com/kerberos-io/agent/machinery/src/log"
+	"github.com/kerberos-io/agent/machinery/src/models"
+	"github.com/kerberos-io/agent/machinery/src/packets"
+	"github.com/moq-dev/moq-go/moq"
+)
+
+const (
+	defaultMoQRelayURL      = "https://relay.uug.ai/anon"
+	minMoQRetryDelay        = time.Second
+	maxMoQRetryDelay        = 30 * time.Second
+	maxMoQLivePacketAge     = 1500 * time.Millisecond
+	slowMoQWriteThreshold   = 100 * time.Millisecond
+	moQWriteWarningInterval = 10 * time.Second
+	duplicateKeyframeWindow = 500 * time.Millisecond
+)
+
+type liveMoQConfig struct {
+	relayURL    string
+	broadcast   string
+	quality     string
+	sourceLabel string
+	queue       *packets.Queue
+}
+
+// label identifies the tier in log lines, since one Agent runs a publisher per
+// quality tier.
+func (c liveMoQConfig) label() string {
+	return c.quality + " (" + c.sourceLabel + " stream)"
+}
+
+// StartLiveStreamMoQ starts the publisher only in the dedicated MoQ build and
+// only when explicitly enabled by the deployment.
+//
+// Unlike WebRTC and HLS — where a viewer negotiates a session with the Agent and
+// can therefore ask for another quality on the fly — MoQ viewers subscribe to a
+// relay and never talk to the Agent. The quality selector is honoured by
+// publishing each tier as its OWN broadcast (see livemoq.BroadcastPath): the
+// high tier from the camera's highest-resolution stream and the low tier from
+// its sub stream, so switching quality in the frontend is a resubscribe to the
+// other path. Each tier only uploads while it actually has subscribers, so the
+// second broadcast is close to free when nobody watches it.
+func StartLiveStreamMoQ(configuration *models.Configuration, communication *models.Communication, subStreamEnabled bool) {
+	if os.Getenv("AGENT_LIVE_MOQ_ENABLED") != "true" {
+		return
+	}
+
+	config := configuration.Config
+	if config.Offline == "true" || config.Capture.Liveview == "false" {
+		log.Log.Info("cloud.StartLiveStreamMoQ(): disabled by Agent live-view configuration")
+		return
+	}
+	if config.Key == "" {
+		log.Log.Warning("cloud.StartLiveStreamMoQ(): AGENT_KEY is required")
+		return
+	}
+
+	// Both tiers are published by default. AGENT_LIVE_MOQ_QUALITY pins the Agent
+	// to a single tier for deployments that must never publish the other one
+	// (viewers asking for the pinned-away tier then find no broadcast).
+	qualities := []string{models.StreamQualityHigh, models.StreamQualityLow}
+	switch strings.ToLower(strings.TrimSpace(os.Getenv("AGENT_LIVE_MOQ_QUALITY"))) {
+	case models.StreamQualityHigh:
+		qualities = []string{models.StreamQualityHigh}
+	case models.StreamQualityLow:
+		qualities = []string{models.StreamQualityLow}
+	}
+
+	relayURL := os.Getenv("AGENT_LIVE_MOQ_URL")
+	if relayURL == "" {
+		relayURL = defaultMoQRelayURL
+	}
+	broadcastPrefix := os.Getenv("AGENT_LIVE_MOQ_BROADCAST_PREFIX")
+
+	ctx := context.Background()
+	if communication.Context != nil {
+		ctx = *communication.Context
+	}
+
+	for _, quality := range qualities {
+		queue := communication.Queue
+		sourceLabel := "main"
+		if models.SelectSubStreamForQuality(config, quality, subStreamEnabled) && communication.SubQueue != nil {
+			queue = communication.SubQueue
+			sourceLabel = "sub"
+		}
+		if queue == nil {
+			log.Log.Warning("cloud.StartLiveStreamMoQ(): packet queue for the " + quality + " tier is unavailable")
+			continue
+		}
+		go runLiveStreamMoQ(ctx, liveMoQConfig{
+			relayURL:    relayURL,
+			broadcast:   livemoq.BroadcastPath(broadcastPrefix, config.Key, quality),
+			quality:     quality,
+			sourceLabel: sourceLabel,
+			queue:       queue,
+		})
+	}
+}
+
+func runLiveStreamMoQ(ctx context.Context, config liveMoQConfig) {
+	log.Log.Info(fmt.Sprintf(
+		"cloud.runLiveStreamMoQ(): publishing %s stream (quality=%s) to %s/%s",
+		config.sourceLabel, config.quality, strings.TrimRight(config.relayURL, "/"), config.broadcast,
+	))
+
+	retryDelay := minMoQRetryDelay
+	for ctx.Err() == nil {
+		connectedAt := time.Now()
+		err := publishLiveStreamMoQ(ctx, config)
+		if ctx.Err() != nil {
+			return
+		}
+		log.Log.Warning("cloud.runLiveStreamMoQ(): publisher stopped: " + err.Error())
+		if time.Since(connectedAt) >= time.Minute {
+			retryDelay = minMoQRetryDelay
+		}
+
+		timer := time.NewTimer(retryDelay)
+		select {
+		case <-ctx.Done():
+			timer.Stop()
+			return
+		case <-timer.C:
+		}
+		if retryDelay < maxMoQRetryDelay {
+			retryDelay *= 2
+			if retryDelay > maxMoQRetryDelay {
+				retryDelay = maxMoQRetryDelay
+			}
+		}
+	}
+}
+
+func publishLiveStreamMoQ(ctx context.Context, config liveMoQConfig) error {
+	client, err := moq.Dial(ctx, config.relayURL)
+	if err != nil {
+		return fmt.Errorf("connect to relay: %w", err)
+	}
+	defer client.Close()
+
+	broadcast, err := client.CreateBroadcast(config.broadcast)
+	if err != nil {
+		return fmt.Errorf("create broadcast: %w", err)
+	}
+	defer broadcast.Finish()
+
+	stream, err := broadcast.PublishMedia("avc3", nil)
+	if err != nil {
+		return fmt.Errorf("create H.264 media stream: %w", err)
+	}
+	defer stream.Finish()
+
+	// Only upload while this tier is actually being watched. `publishing` starts
+	// true so the track becomes discoverable on the relay even before the first
+	// subscriber ever arrives; from the moment a viewer has attached once, the
+	// subscriber watcher takes over and idles the tier again when everybody left.
+	watchCtx, cancelWatch := context.WithCancel(ctx)
+	defer cancelWatch()
+	publishing := &atomic.Bool{}
+	publishing.Store(true)
+	go watchLiveStreamMoQSubscribers(watchCtx, stream, publishing, config)
+
+	cursor := config.queue.Latest()
+	gate := livemoq.FrameGate{}
+	deduplicator := livemoq.KeyframeDeduplicator{}
+	var lastSlowWriteWarning time.Time
+	var lastDuplicateKeyframeWarning time.Time
+	idle := false
+	for {
+		packet, err := cursor.ReadPacket()
+		if err != nil {
+			return fmt.Errorf("read packet: %w", err)
+		}
+		if !publishing.Load() {
+			// Keep draining the cursor so we stay at the live edge, but publish
+			// nothing. The gate is closed so the next viewer resumes on a keyframe.
+			if !idle {
+				gate.Reset()
+				deduplicator.Reset()
+				idle = true
+			}
+			continue
+		}
+		idle = false
+		if !packet.IsVideo || len(packet.Data) == 0 || !strings.EqualFold(packet.Codec, "H264") {
+			continue
+		}
+		allowed, event := gate.Allow(packet.IsKeyFrame, packet.CurrentTime, time.Now(), maxMoQLivePacketAge)
+		switch event {
+		case livemoq.FrameGateEventStarted:
+			log.Log.Info("cloud.publishLiveStreamMoQ(): first H.264 keyframe received; " + config.label() + " broadcast is live")
+		case livemoq.FrameGateEventLagging:
+			log.Log.Warning("cloud.publishLiveStreamMoQ(): " + config.label() + " stream is lagging; dropping packets until a recent keyframe")
+		case livemoq.FrameGateEventRecovered:
+			log.Log.Info("cloud.publishLiveStreamMoQ(): caught up with the " + config.label() + " live stream at a recent keyframe")
+		}
+		if !allowed {
+			continue
+		}
+		payload, normalizationStats, err := livemoq.NormalizeH264AccessUnitWithStats(packet.Data)
+		if err != nil {
+			return fmt.Errorf("normalize H.264 access unit: %w", err)
+		}
+		if normalizationStats.DuplicateIDRNALUs > 0 && time.Since(lastDuplicateKeyframeWarning) >= moQWriteWarningInterval {
+			log.Log.Warning(fmt.Sprintf(
+				"cloud.publishLiveStreamMoQ(): %s removed %d duplicate IDR NALU(s) from H.264 keyframe (timestamp_ms=%d)",
+				config.label(), normalizationStats.DuplicateIDRNALUs, packet.Time,
+			))
+			lastDuplicateKeyframeWarning = time.Now()
+		}
+		if packet.IsKeyFrame && deduplicator.IsDuplicate(packet.Time, packet.CurrentTime, payload, time.Now(), duplicateKeyframeWindow) {
+			if time.Since(lastDuplicateKeyframeWarning) >= moQWriteWarningInterval {
+				log.Log.Warning(fmt.Sprintf(
+					"cloud.publishLiveStreamMoQ(): %s dropping duplicate H.264 keyframe (timestamp_ms=%d)",
+					config.label(), packet.Time,
+				))
+				lastDuplicateKeyframeWarning = time.Now()
+			}
+			continue
+		}
+		frame := moq.Frame{
+			Payload:     payload,
+			TimestampUs: livemoq.TimestampUs(packet.Time),
+		}
+		writeStartedAt := time.Now()
+		if err := stream.WriteFrame(frame); err != nil {
+			return fmt.Errorf("write H.264 access unit: %w", err)
+		}
+		writeDuration := time.Since(writeStartedAt)
+		if writeDuration >= slowMoQWriteThreshold && time.Since(lastSlowWriteWarning) >= moQWriteWarningInterval {
+			packetAge := time.Duration(0)
+			if packet.CurrentTime > 0 {
+				packetAge = time.Since(time.UnixMilli(packet.CurrentTime))
+				if packetAge < 0 {
+					packetAge = 0
+				}
+			}
+			log.Log.Warning(fmt.Sprintf(
+				"cloud.publishLiveStreamMoQ(): %s WriteFrame blocked for %s (packet_age=%s keyframe=%t)",
+				config.label(), writeDuration.Round(time.Millisecond), packetAge.Round(time.Millisecond), packet.IsKeyFrame,
+			))
+			lastSlowWriteWarning = time.Now()
+		}
+	}
+}
+
+// watchLiveStreamMoQSubscribers flips the publisher between uploading and idling
+// as viewers subscribe to and leave this tier's broadcast. Used and Unused both
+// block, so they are followed from their own goroutine.
+//
+// It deliberately never turns publishing off before the first subscriber has
+// been observed: the relay catalog is only complete once media has flowed, so
+// going idle up front could keep the tier undiscoverable. On any error it fails
+// open (keeps publishing) — a stalled watcher must never take the live view down.
+func watchLiveStreamMoQSubscribers(ctx context.Context, stream *moq.MediaProducer, publishing *atomic.Bool, config liveMoQConfig) {
+	for ctx.Err() == nil {
+		if err := stream.Used(ctx); err != nil {
+			publishing.Store(true)
+			return
+		}
+		if publishing.CompareAndSwap(false, true) {
+			log.Log.Info("cloud.watchLiveStreamMoQSubscribers(): viewer subscribed, resuming the " + config.label() + " broadcast")
+		}
+
+		if err := stream.Unused(ctx); err != nil {
+			publishing.Store(true)
+			return
+		}
+		publishing.Store(false)
+		log.Log.Info("cloud.watchLiveStreamMoQSubscribers(): no viewers left, idling the " + config.label() + " broadcast")
+	}
+}

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 || math.IsInf(metadata.FPS, 0) || math.IsNaN(metadata.FPS) {
+			return ""
+		}
+		return strconv.FormatFloat(metadata.FPS, 'f', -1, 64)
+	}
+
+	// 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

@@ -305,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)
 	}
@@ -326,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

@@ -243,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,
@@ -261,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.97}`)
 
 	uploaded, responded, supported, _, err := uploadVaultResumable(testVault(ts.URL), "pk", "dev", fileName, "test", "primary")
 	if err != nil {
@@ -275,6 +287,103 @@ 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.97" {
+		t.Fatalf("POST metadata fps = %q, want %q", got, "29.97")
+	}
+	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 fractional", fps: `{"fps":17.35}`, want: "17.35"},
+		{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) {
@@ -579,6 +688,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 {
@@ -649,6 +759,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

@@ -305,6 +305,10 @@ func RunAgent(configDirectory string, configuration *models.Configuration, commu
 	// watching.
 	go cloud.HandleLiveStreamHLS(configuration, communication, mqttClient, subStreamEnabled)
 
+	// MoQ is available only in the dedicated CGO/glibc build. The standard
+	// static Alpine build resolves this hook to a no-op.
+	cloud.StartLiveStreamMoQ(configuration, communication, subStreamEnabled)
+
 	// Handle livestream HD (high resolution over WEBRTC). Both the main and sub
 	// stream are exposed as separate broadcasters so a viewer can request the
 	// high (main) or low (sub) resolution per peer connection; "auto" prefers the

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.Warning("computervision.main.ProcessMotion(): motion detection is DISABLED because pixelChangeThreshold is set to 0 or less (nil/unset would default to 150). If motion detection is expected to be running, set capture.pixelChangeThreshold to a positive value (150 recommended) or AGENT_CAPTURE_PIXEL_CHANGE, then restart/update the agent.")
-
-	} 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.")
 
@@ -236,9 +227,9 @@ func ProcessMotion(motionCursor *packets.QueueCursor, configuration *models.Conf
 													// 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,
+													"pixelChangeThreshold": pixelThreshold,
+												},
 											},
-										},
 										}
 										payload, err := models.PackageMQTTMessage(configuration, message)
 										if err == nil {

machinery/src/config/main.go

@@ -651,13 +651,12 @@ func applyAgentEnvVars(configuration *models.Configuration, prefix string, apply
 		}
 	}
 
-	// Motion sensitivity: nil/unset must still resolve to the default (150), not
-	// be left nil. An explicit 0 (temporary "disable motion detection" switch
-	// from the UI) is a real, non-nil value and must NOT be touched here. Only
-	// applied for the effective configuration (applyDefaults), not for the
-	// separate global/custom views, so a missing value in one layer can still be
-	// inherited from the other instead of being masked by this default.
-	if applyDefaults && configuration.Config.Capture.PixelChangeThreshold == nil {
+	// 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
 	}

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/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       float64 `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/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)
+	}
 }

machinery/verify-moq-devcontainer.sh

@@ -0,0 +1,23 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+required_glibc="2.38"
+current_glibc="$(getconf GNU_LIBC_VERSION | awk '{print $2}')"
+
+if ! dpkg --compare-versions "$current_glibc" ge "$required_glibc"; then
+    echo "MoQ requires glibc ${required_glibc}+; this container has ${current_glibc}." >&2
+    echo "Rebuild the devcontainer with the Trixie base, then run this check again." >&2
+    exit 1
+fi
+
+echo "==> Testing MoQ packages (glibc ${current_glibc})"
+GOWORK=off go test -tags moq ./src/cloud/livemoq ./src/cloud
+
+binary="${TMPDIR:-/tmp}/agent-moq"
+trap 'rm -f "$binary"' EXIT
+
+echo "==> Linking the MoQ Agent"
+GOWORK=off go build -tags moq -o "$binary" ./main.go
+
+echo "==> Running the linked binary"
+"$binary" -action version