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

@@ -26,6 +26,7 @@ require (
 	github.com/kerberos-io/joy4 v1.0.64
 	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

@@ -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/main.go

@@ -62,7 +62,7 @@ func recordingUploadMetadata(name, deviceKey string, timestamp int64, mp4Video *
 	}
 	value := mp4Video.AverageFPS()
 	if value > 0 && value <= 240 && !math.IsInf(value, 0) && !math.IsNaN(value) {
-		metadata.FPS = int(math.Floor(value))
+		metadata.FPS = value
 	}
 	return metadata
 }

machinery/src/capture/main_test.go

@@ -29,9 +29,13 @@ func TestQueueRecordingForUploadStoresFinalizedMetadata(t *testing.T) {
 	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 {
+	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) {
@@ -44,7 +48,7 @@ func TestQueueRecordingForUploadKeepsUnknownFPSCompatible(t *testing.T) {
 
 			metadata := models.RecordingUploadMetadata{FileName: "recording.mp4"}
 			if fps >= 1 && fps <= 240 && !math.IsNaN(fps) && !math.IsInf(fps, 0) {
-				metadata.FPS = int(math.Floor(fps))
+				metadata.FPS = fps
 			}
 			queueRecordingForUpload(configDirectory, metadata)
 

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

@@ -28,10 +28,10 @@ func queuedRecordingFPS(fileName string) string {
 	marker := strings.TrimSpace(string(value))
 	if strings.HasPrefix(marker, "{") {
 		metadata, ok := decodeRecordingUploadMetadata(value)
-		if !ok || metadata.FPS <= 0 || metadata.FPS > 240 {
+		if !ok || metadata.FPS <= 0 || metadata.FPS > 240 || math.IsInf(metadata.FPS, 0) || math.IsNaN(metadata.FPS) {
 			return ""
 		}
-		return strconv.Itoa(metadata.FPS)
+		return strconv.FormatFloat(metadata.FPS, 'f', -1, 64)
 	}
 
 	// Compatibility with markers created before upload metadata used JSON.

machinery/src/cloud/tus_client_test.go

@@ -272,7 +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}`)
+	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 {
@@ -289,8 +289,8 @@ func TestUploadVaultResumable_HappyPath(t *testing.T) {
 	}
 	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["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")
@@ -307,6 +307,7 @@ func TestQueuedRecordingFPSValidation(t *testing.T) {
 		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}`},

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/models/recording_upload_metadata.go

@@ -11,11 +11,11 @@ const RecordingUploadMetadataExtension = ".metadata"
 // 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"`
+	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

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