Drop truncated GOPs at loop/restart seams

Buffer and conditionally drop a pending GOP when an upstream loop/restart emits a premature IDR. mp4.go: add gopBuffer and bufferedSample types, hold samples until the next video keyframe, detect a seam by comparing the new keyframe interval against the previous cadence (SeamGapDivisor) and drop the short/truncated tail GOP or commit buffered samples. Add commitBufferedGOP and commitSampleToTrack helpers and flush the final buffered GOP on Close. mp4_loopseam_test.go: update test descriptions, expectations and names to assert the truncated tail GOP is dropped exactly once across different GOP sizes. Dockerfile and Dockerfile.arm64: re-declare ARG VERSION inside the build stage and only derive git describe when VERSION is unset or the default 0.0.0 so build-arg values are respected. Add a binary MP4 fixture used by the tests.

.github/workflows/pr-description.yaml

@@ -2,25 +2,16 @@ name: Autofill PR description
 
 on: pull_request
 
-env:
-  ORGANIZATION: uugai
-  PROJECT: ${{ github.event.repository.name }}
-  PR_NUMBER: ${{ github.event.number }}
-  
 jobs:
   openai-pr-description:
-    runs-on: ubuntu-22.04
-
-    steps:
-      - uses: actions/checkout@v4
-      - name: Autofill PR description if empty using OpenAI
-        uses: cedricve/azureopenai-pr-description@master
-        with:
-          github_token: ${{ secrets.TOKEN }}
-          openai_api_key: ${{ secrets.OPENAI_API_KEY }}
-          azure_openai_api_key: ${{ secrets.AZURE_OPENAI_API_KEY }}
-          azure_openai_endpoint: ${{ secrets.AZURE_OPENAI_ENDPOINT }}
-          azure_openai_version: ${{ secrets.AZURE_OPENAI_VERSION }}
-          openai_model: ${{ secrets.OPENAI_MODEL }}
-          pull_request_url: https://pr${{ env.PR_NUMBER }}.api.kerberos.lol
-          overwrite_description: true
+    uses: uug-ai/workflows/.github/workflows/pr-description.yml@main
+    with:
+      pr_number: ${{ github.event.number }}
+      pull_request_url: ""
+      overwrite_description: true
+    secrets:
+      TOKEN: ${{ secrets.TOKEN }}
+      AZURE_OPENAI_API_KEY: ${{ secrets.AZURE_OPENAI_API_KEY }}
+      OPENAI_MODEL: ${{ secrets.OPENAI_MODEL }}
+      AZURE_OPENAI_ENDPOINT: ${{ secrets.AZURE_OPENAI_ENDPOINT }}
+      AZURE_OPENAI_VERSION: ${{ secrets.AZURE_OPENAI_VERSION }}

.github/workflows/release-bump.yml

@@ -0,0 +1,168 @@
+name: Bump release
+
+on:
+  workflow_dispatch:
+    inputs:
+      bump:
+        description: "Which part of the version to bump"
+        required: true
+        default: patch
+        type: choice
+        options:
+          - major
+          - minor
+          - patch
+
+permissions:
+  contents: write
+
+env:
+  REPO: kerberos/agent
+
+jobs:
+  # Determine the next version, create the GitHub release and expose the tag.
+  bump-release:
+    uses: uug-ai/workflows/.github/workflows/release-bump.yml@main
+    with:
+      bump: ${{ github.event.inputs.bump }}
+    secrets: inherit
+
+  # Publish the platform image to the uug-ai GitHub Container Registry
+  # (ghcr.io/uug-ai/agent-platform).
+  release:
+    needs: bump-release
+    uses: uug-ai/workflows/.github/workflows/release-create.yml@main
+    with:
+      organization: uug-ai
+      project: ${{ github.event.repository.name }}
+      tag: ${{ needs.bump-release.outputs.tag }}
+      docker_context: "."
+      create_gitops_pr: false
+      runner_matrix: >-
+        [
+          {"architecture":"amd64","runner":"ubuntu-24.04"},
+          {"architecture":"arm64","runner":"ubuntu-24.04-arm"}
+        ]
+    secrets: inherit
+
+  # Everything below mirrors the agent's own release-create.yml pipeline and
+  # publishes the multi-arch image to the kerberos/agent Docker Hub repo, driven
+  # by the freshly bumped tag instead of a `release: created` event.
+  build-amd64:
+    needs: bump-release
+    runs-on: ubuntu-24.04
+    permissions:
+      contents: write
+    strategy:
+      matrix:
+        architecture: [amd64]
+    steps:
+      - name: Login to DockerHub
+        uses: docker/login-action@v2
+        with:
+          username: ${{ secrets.DOCKER_USERNAME }}
+          password: ${{ secrets.DOCKER_PASSWORD }}
+      - name: Checkout
+        uses: actions/checkout@v3
+      - uses: benjlevesque/short-sha@v2.1
+        id: short-sha
+        with:
+          length: 7
+      - name: Run Build
+        run: |
+          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}
+      - name: Strip binary
+        run: tar -cf agent-${{matrix.architecture}}.tar -C output-${{matrix.architecture}} . && rm -rf output-${{matrix.architecture}}
+      - name: Build and push Docker image
+        run: |
+          docker tag ${{matrix.architecture}} $REPO-arch:arch-${{matrix.architecture}}-${{ needs.bump-release.outputs.tag }}
+          docker push $REPO-arch:arch-${{matrix.architecture}}-${{ needs.bump-release.outputs.tag }}
+      - name: Upload artifact
+        uses: actions/upload-artifact@v4
+        with:
+          name: agent-${{matrix.architecture}}.tar
+          path: agent-${{matrix.architecture}}.tar
+
+  build-arm64:
+    needs: bump-release
+    runs-on: ubuntu-24.04-arm
+    permissions:
+      contents: write
+    strategy:
+      matrix:
+        architecture: [arm64]
+    steps:
+      - name: Login to DockerHub
+        uses: docker/login-action@v2
+        with:
+          username: ${{ secrets.DOCKER_USERNAME }}
+          password: ${{ secrets.DOCKER_PASSWORD }}
+      - name: Checkout
+        uses: actions/checkout@v3
+      - uses: benjlevesque/short-sha@v2.1
+        id: short-sha
+        with:
+          length: 7
+      - name: Run Build
+        run: |
+          docker build --provenance=false --build-arg VERSION=${{ needs.bump-release.outputs.tag }} -t ${{matrix.architecture}} -f Dockerfile.arm64 .
+          CID=$(docker create ${{matrix.architecture}})
+          docker cp ${CID}:/home/agent ./output-${{matrix.architecture}}
+          docker rm ${CID}
+      - name: Strip binary
+        run: tar -cf agent-${{matrix.architecture}}.tar -C output-${{matrix.architecture}} . && rm -rf output-${{matrix.architecture}}
+      - name: Build and push Docker image
+        run: |
+          docker tag ${{matrix.architecture}} $REPO-arch:arch-${{matrix.architecture}}-${{ needs.bump-release.outputs.tag }}
+          docker push $REPO-arch:arch-${{matrix.architecture}}-${{ needs.bump-release.outputs.tag }}
+      - name: Upload artifact
+        uses: actions/upload-artifact@v4
+        with:
+          name: agent-${{matrix.architecture}}.tar
+          path: agent-${{matrix.architecture}}.tar
+
+  create-manifest:
+    runs-on: ubuntu-24.04
+    needs: [bump-release, build-amd64, build-arm64]
+    steps:
+      - name: Login to DockerHub
+        uses: docker/login-action@v2
+        with:
+          username: ${{ secrets.DOCKER_USERNAME }}
+          password: ${{ secrets.DOCKER_PASSWORD }}
+      - name: Create and push multi-arch manifest
+        run: |
+          docker manifest create $REPO:${{ needs.bump-release.outputs.tag }} \
+            $REPO-arch:arch-amd64-${{ needs.bump-release.outputs.tag }} \
+            $REPO-arch:arch-arm64-${{ needs.bump-release.outputs.tag }}
+          docker manifest push $REPO:${{ needs.bump-release.outputs.tag }}
+      - name: Create and push latest manifest
+        run: |
+          docker manifest create $REPO:latest \
+            $REPO-arch:arch-amd64-${{ needs.bump-release.outputs.tag }} \
+            $REPO-arch:arch-arm64-${{ needs.bump-release.outputs.tag }}
+          docker manifest push $REPO:latest
+
+  create-release:
+    runs-on: ubuntu-24.04
+    needs: [bump-release, build-amd64, build-arm64]
+    permissions:
+      contents: write
+    steps:
+      - name: Download all artifacts
+        uses: actions/download-artifact@v4
+      - name: Create a release
+        uses: ncipollo/release-action@v1
+        with:
+          latest: true
+          allowUpdates: true
+          name: ${{ needs.bump-release.outputs.tag }}
+          tag: ${{ needs.bump-release.outputs.tag }}
+          generateReleaseNotes: false
+          omitBodyDuringUpdate: true
+          artifacts: "agent-*.tar/agent-*.tar"
+
+  

Dockerfile

@@ -4,6 +4,11 @@ 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
@@ -35,7 +40,9 @@ RUN cat /go/src/github.com/kerberos-io/agent/machinery/version
 
 RUN cd /go/src/github.com/kerberos-io/agent/machinery && \
 	go mod download && \
-	VERSION=$(cd /go/src/github.com/kerberos-io/agent && git describe --tags --always 2>/dev/null || echo "${VERSION}") && \
+	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 && \
@@ -60,7 +67,7 @@ RUN cp -r /agent ./
 
 RUN /dist/agent/main version
 
-FROM node:18.14.0-alpine3.16 AS build-ui
+FROM node:22-alpine AS build-ui
 
 RUN apk update && apk upgrade --available && sync
 

Dockerfile.arm64

@@ -4,6 +4,11 @@ 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
@@ -35,7 +40,9 @@ RUN cat /go/src/github.com/kerberos-io/agent/machinery/version
 
 RUN cd /go/src/github.com/kerberos-io/agent/machinery && \
 	go mod download && \
-	VERSION=$(cd /go/src/github.com/kerberos-io/agent && git describe --tags --always 2>/dev/null || echo "${VERSION}") && \
+	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 && \
@@ -60,7 +67,7 @@ RUN cp -r /agent ./
 
 RUN /dist/agent/main version
 
-FROM node:18.14.0-alpine3.16 AS build-ui
+FROM node:22-alpine AS build-ui
 
 RUN apk update && apk upgrade --available && sync
 

README.md

@@ -65,6 +65,7 @@ There are a myriad of cameras out there (USB, IP and other cameras), and it migh
 
 ### Contributing
 
+1. [Security vulnerability reporting](#security-vulnerability-reporting)
 1. [Contribute with Codespaces](#contribute-with-codespaces)
 2. [Develop and build](#develop-and-build)
 3. [Building from source](#building-from-source)
@@ -301,6 +302,10 @@ If we talk about video encoders and decoders (codecs) there are 2 major video co
 
 Conclusion: depending on the use case you might choose one over the other, and you can use both at the same time. For example you can use H264 (main stream) for livestreaming, and H265 (sub stream) for recording. If you wish to play recordings in a cross-platform and cross-browser environment, you might opt for H264 for better support.
 
+## Security vulnerability reporting
+
+If you found a potential security vulnerability, please use the private channels described in [SECURITY.md](SECURITY.md). Avoid opening public GitHub issues for sensitive findings.
+
 ## Contribute with Codespaces
 
 One of the major blockers for letting you contribute to an Open Source project is to set up your local development machine. Why? Because you might already have some tools and libraries installed that are used for other projects, and the libraries you would need for Kerberos Agent, for example FFmpeg, might require a different version. Welcome to dependency hell...

SECURITY.md

@@ -0,0 +1,40 @@
+# Security Policy
+
+## Supported Versions
+
+We only provide security fixes for the latest release series on the `master` branch.
+
+## Reporting a Vulnerability
+
+Please do **not** open a public GitHub issue for potential security vulnerabilities.
+
+Use one of the private channels below:
+
+1. Preferred: GitHub private vulnerability reporting
+   - https://github.com/kerberos-io/agent/security/advisories/new
+2. Fallback: Email
+   - support@kerberos.io
+   - Optional CC: support@uug.ai
+
+Please include:
+
+- A short summary and impact.
+- Reproduction steps or proof of concept.
+- Affected version(s), commit hash, or deployment details.
+- Any proposed mitigation/workaround.
+- Your preferred attribution name.
+
+For faster triage, use this subject format in email:
+
+`[Security][Kerberos Agent] <short title>`
+
+## Response Expectations
+
+- Acknowledgement target: within 3 business days.
+- Triage/update target: within 7 business days after acknowledgement.
+
+If you do not receive a response in time, please resend your report and include your original timestamp.
+
+## Disclosure and Credits
+
+We follow coordinated disclosure. After a fix is available, we will credit reporters unless they prefer to stay anonymous.

machinery/README.md

@@ -22,4 +22,8 @@ https://brianmacdonald.github.io/Ethonate/address#0xf4a759C9436E2280Ea9cdd23d314
 
 [**Docker Hub**](https://hub.docker.com/r/kerberos/agent) | [**Documentation**](https://doc.kerberos.io) | [**Website**](https://kerberos.io)
 
-Kerberos Open source (v3) is a cutting edge video surveillance management system made available as Open Source under the MIT License. This means that all the source code is available for you or your company, and you can use, transform and distribute the source code; as long you keep a reference of the original license. Kerberos Open Source (v3) can be used for commercial usage (which was not the case for v2). Read more [about the license here](LICENSE).
+Kerberos Open source (v3) is a cutting edge video surveillance management system made available as Open Source under the MIT License. This means that all the source code is available for you or your company, and you can use, transform and distribute the source code; as long you keep a reference of the original license. Kerberos Open Source (v3) can be used for commercial usage (which was not the case for v2). Read more [about the license here](LICENSE).
+
+## Security reporting
+
+For sensitive vulnerabilities, use private disclosure channels documented in [../SECURITY.md](../SECURITY.md).

machinery/src/capture/gortsplib.go

@@ -95,6 +95,129 @@ type Golibrtsp struct {
 	keyframeBufferSize       int
 	keyframeBufferIndex      int
 	keyframeMutex            sync.Mutex
+
+	// Stream health instrumentation. Used to pinpoint the root cause behind
+	// "RTP packets lost" + watchdog restarts by separating downstream
+	// back-pressure from upstream network/camera stalls.
+	health      *streamHealth
+	streamLabel string
+}
+
+// streamHealth instruments the RTSP read path. gortsplib delivers every RTP
+// packet on a single read goroutine; queue.WritePacket() is synchronous, so if
+// a downstream consumer (recording, muxing, WebRTC) is slow or the process is
+// CPU-starved, WritePacket() blocks, the TCP socket is not drained, and the
+// camera advances RTP sequence numbers -> "RTP packets lost". This type makes
+// the two failure modes distinguishable:
+//   - large writeMax / writeAvg  => downstream back-pressure (our side).
+//   - large gapMax with fast writes => upstream network / camera stall.
+type streamHealth struct {
+	mu          sync.Mutex
+	windowStart time.Time
+	lastPacket  time.Time
+	frames      int64
+	writeSum    time.Duration
+	writeMax    time.Duration
+	gapMax      time.Duration
+	lost        uint64
+	decodeErrs  int64
+}
+
+const (
+	streamHealthWindow    = 10 * time.Second
+	streamHealthWriteWarn = 150 * time.Millisecond
+	streamHealthGapWarn   = 1500 * time.Millisecond
+)
+
+func newStreamHealth() *streamHealth {
+	now := time.Now()
+	return &streamHealth{windowStart: now, lastPacket: now}
+}
+
+// observePacket records one processed video frame: the wall-clock gap since the
+// previous frame (arrival cadence) and how long WritePacket() blocked
+// (back-pressure). It emits an immediate warning when either side stalls and a
+// periodic summary every streamHealthWindow.
+func (h *streamHealth) observePacket(streamType string, writeDur time.Duration) {
+	if h == nil {
+		return
+	}
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	now := time.Now()
+	var gap time.Duration
+	if h.frames == 0 {
+		// First frame: initialize timing to avoid counting RTSP setup time as a stall.
+		h.windowStart = now
+		h.lastPacket = now
+		gap = 0
+	} else {
+		gap = now.Sub(h.lastPacket)
+		h.lastPacket = now
+	}
+	h.frames++
+	h.writeSum += writeDur
+	if writeDur > h.writeMax {
+		h.writeMax = writeDur
+	}
+	if gap > h.gapMax {
+		h.gapMax = gap
+	}
+	if writeDur >= streamHealthWriteWarn {
+		log.Log.Warning(fmt.Sprintf(
+			"capture.golibrtsp.health(%s): WritePacket blocked %dms — downstream back-pressure / CPU starvation",
+			streamType, writeDur.Milliseconds()))
+	}
+	if gap >= streamHealthGapWarn {
+		log.Log.Warning(fmt.Sprintf(
+			"capture.golibrtsp.health(%s): %dms since previous frame — upstream network / camera stall",
+			streamType, gap.Milliseconds()))
+	}
+	if now.Sub(h.windowStart) >= streamHealthWindow {
+		elapsed := now.Sub(h.windowStart).Seconds()
+		var avgWriteMs float64
+		if h.frames > 0 {
+			avgWriteMs = float64(h.writeSum.Milliseconds()) / float64(h.frames)
+		}
+		log.Log.Info(fmt.Sprintf(
+			"capture.golibrtsp.health(%s): %.0fs window — frames=%d (%.1f/s) writeAvg=%.1fms writeMax=%dms gapMax=%dms lost=%d decodeErrs=%d",
+			streamType, elapsed, h.frames, float64(h.frames)/elapsed, avgWriteMs,
+			h.writeMax.Milliseconds(), h.gapMax.Milliseconds(), h.lost, h.decodeErrs))
+		h.windowStart = now
+		h.frames = 0
+		h.writeSum = 0
+		h.writeMax = 0
+		h.gapMax = 0
+		h.lost = 0
+		h.decodeErrs = 0
+	}
+}
+
+// observeLost is invoked by gortsplib when RTP sequence numbers skip. On a TCP
+// transport this means the sender (camera) dropped packets because we were not
+// reading fast enough, not loss on the wire.
+func (h *streamHealth) observeLost(streamType string, lost uint64) {
+	if h == nil {
+		return
+	}
+	h.mu.Lock()
+	h.lost += lost
+	h.mu.Unlock()
+	log.Log.Warning(fmt.Sprintf(
+		"capture.golibrtsp.health(%s): %d RTP packet(s) lost — sender-side gap (receiver not draining TCP fast enough)",
+		streamType, lost))
+}
+
+// observeDecodeError is invoked by gortsplib on incomplete/invalid access units,
+// which are a downstream symptom of the loss reported by observeLost.
+func (h *streamHealth) observeDecodeError(streamType string, err error) {
+	if h == nil {
+		return
+	}
+	h.mu.Lock()
+	h.decodeErrs++
+	h.mu.Unlock()
+	log.Log.Debug(fmt.Sprintf("capture.golibrtsp.health(%s): decode error: %s", streamType, err.Error()))
 }
 
 // fpsTracker holds per-stream state for PTS-based FPS calculation.
@@ -195,9 +318,20 @@ func (g *Golibrtsp) Connect(ctx context.Context, ctxOtel context.Context) (err e
 	defer span.End()
 
 	transport := gortsplib.TransportTCP
+	g.health = newStreamHealth()
 	g.Client = gortsplib.Client{
 		RequestBackChannels: false,
 		Transport:           &transport,
+		// Route gortsplib's packet-loss / decode-error reporting through our
+		// structured logger with stream context (replaces its plain stdout
+		// logging). These hooks are what let us tell whether the camera is
+		// dropping packets because we can't drain the socket fast enough.
+		OnPacketsLost: func(lost uint64) {
+			g.health.observeLost(g.streamLabel, lost)
+		},
+		OnDecodeError: func(err error) {
+			g.health.observeDecodeError(g.streamLabel, err)
+		},
 	}
 
 	// parse URL
@@ -517,10 +651,45 @@ func (g *Golibrtsp) ConnectBackChannel(ctx context.Context, ctxRunAgent context.
 	return
 }
 
+// dtsExtractor abstracts the codec-specific DTS extractors from mediacommon
+// (h264.DTSExtractor2 and h265.DTSExtractor2), which expose the same method.
+type dtsExtractor interface {
+	Extract(au [][]byte, pts int64) (int64, error)
+}
+
+// compositionOffsetMs returns the composition time offset (PTS - DTS) in
+// milliseconds for a coded access unit. Streams that contain B-frames deliver
+// access units in decode order with non-monotonic PTS; the fragmented MP4
+// writer needs a monotonic DTS timeline plus a per-sample composition offset
+// so browsers (Media Source Extensions) can decode the chained segments.
+//
+// It returns 0 when the codec has no frame reordering (the common case, e.g.
+// baseline "IPPP" streams) or when extraction fails, making it a safe no-op.
+func compositionOffsetMs(ext dtsExtractor, au [][]byte, pts int64, clockRate int) int64 {
+	if ext == nil || clockRate <= 0 {
+		return 0
+	}
+	dts, err := ext.Extract(au, pts)
+	if err != nil {
+		return 0
+	}
+	offset := pts - dts
+	if offset <= 0 {
+		return 0
+	}
+	return offset * 1000 / int64(clockRate)
+}
+
 // Start the RTSP client, and start reading packets.
 func (g *Golibrtsp) Start(ctx context.Context, streamType string, queue *packets.Queue, configuration *models.Configuration, communication *models.Communication) (err error) {
 	log.Log.Debug("capture.golibrtsp.Start(): started")
 
+	// Label this client's loss/decode/health logging with the stream type.
+	g.streamLabel = streamType
+	if g.health == nil {
+		g.health = newStreamHealth()
+	}
+
 	// called when a MULAW audio RTP packet arrives
 	if g.AudioG711Media != nil && g.AudioG711Forma != nil {
 		g.Client.OnPacketRTP(g.AudioG711Media, g.AudioG711Forma, func(rtppkt *rtp.Packet) {
@@ -602,7 +771,9 @@ func (g *Golibrtsp) Start(ctx context.Context, streamType string, queue *packets
 	var filteredAU [][]byte
 	if g.VideoH264Media != nil && g.VideoH264Forma != nil {
 
-		//dtsExtractor := h264.NewDTSExtractor2()
+		// Extracts DTS from the bitstream to support B-frame H264 streams.
+		// Created once per stream (tracks reorder state across access units).
+		h264DTSExtractor := h264.NewDTSExtractor2()
 
 		g.Client.OnPacketRTP(g.VideoH264Media, g.VideoH264Forma, func(rtppkt *rtp.Packet) {
 
@@ -742,6 +913,11 @@ func (g *Golibrtsp) Start(ctx context.Context, streamType string, queue *packets
 					return
 				}
 
+				// Composition time offset (PTS - DTS) in milliseconds. Non-zero
+				// only for streams with B-frames; the MP4 writer uses it to keep a
+				// monotonic decode timeline and present frames in PTS order.
+				compositionOffset := compositionOffsetMs(h264DTSExtractor, au, pts2, g.VideoH264Forma.ClockRate())
+
 				pkt := packets.Packet{
 					IsKeyFrame:      idrPresent,
 					Packet:          rtppkt,
@@ -749,7 +925,7 @@ func (g *Golibrtsp) Start(ctx context.Context, streamType string, queue *packets
 					Time:            pts2,
 					TimeLegacy:      pts,
 					CurrentTime:     time.Now().UnixMilli(),
-					CompositionTime: pts2,
+					CompositionTime: compositionOffset,
 					Idx:             g.VideoH264Index,
 					IsVideo:         true,
 					IsAudio:         false,
@@ -777,15 +953,38 @@ func (g *Golibrtsp) Start(ctx context.Context, streamType string, queue *packets
 
 				pkt.Data = pkt.Data[4:]
 				if pkt.IsKeyFrame {
-					annexbNALUStartCode := func() []byte { return []byte{0x00, 0x00, 0x00, 0x01} }
-					pkt.Data = append(annexbNALUStartCode(), pkt.Data...)
-					pkt.Data = append(g.VideoH264Forma.PPS, pkt.Data...)
-					pkt.Data = append(annexbNALUStartCode(), pkt.Data...)
-					pkt.Data = append(g.VideoH264Forma.SPS, pkt.Data...)
-					pkt.Data = append(annexbNALUStartCode(), pkt.Data...)
+					// Prepend SPS/PPS (when available) in front of every keyframe so the
+					// access unit is self-contained. Downstream decoders (and the MP4 writer's
+					// in-band parameter-set recovery) rely on this; a recording whose first
+					// frame lacks SPS/PPS produces an MP4 with an empty avcC, which makes FFmpeg
+					// report "non-existing PPS 0 referenced".
+					//
+					// Build the payload in a freshly allocated buffer. The previous code
+					// did append(g.VideoH264Forma.PPS, pkt.Data...): because the SPS/PPS
+					// slices are sub-slices of the RTP reassembly buffer (spare capacity),
+					// that append wrote into - and corrupted - the shared parameter-set
+					// backing arrays, occasionally poisoning the SPS/PPS stored for the
+					// recording.
+					startCode := []byte{0x00, 0x00, 0x00, 0x01}
+					out := make([]byte, 0, len(g.VideoH264Forma.SPS)+len(g.VideoH264Forma.PPS)+len(pkt.Data)+12)
+					if len(g.VideoH264Forma.SPS) > 0 {
+						out = append(out, startCode...)
+						out = append(out, g.VideoH264Forma.SPS...)
+					}
+					if len(g.VideoH264Forma.PPS) > 0 {
+						out = append(out, startCode...)
+						out = append(out, g.VideoH264Forma.PPS...)
+					}
+					out = append(out, startCode...)
+					out = append(out, pkt.Data...)
+					pkt.Data = out
 				}
 
+				writeStart := time.Now()
 				queue.WritePacket(pkt)
+				// Records WritePacket() blocking time and frame arrival cadence so
+				// we can tell back-pressure from a network/camera stall.
+				g.health.observePacket(streamType, time.Since(writeStart))
 
 				// This will check if we need to stop the thread,
 				// because of a reconfiguration.
@@ -817,6 +1016,11 @@ func (g *Golibrtsp) Start(ctx context.Context, streamType string, queue *packets
 
 	// called when a video RTP packet arrives for H265
 	if g.VideoH265Media != nil && g.VideoH265Forma != nil {
+
+		// Extracts DTS from the bitstream to support B-frame H265 streams.
+		// Created once per stream (tracks reorder state across access units).
+		h265DTSExtractor := h265.NewDTSExtractor2()
+
 		g.Client.OnPacketRTP(g.VideoH265Media, g.VideoH265Forma, func(rtppkt *rtp.Packet) {
 
 			// This will check if we need to stop the thread,
@@ -860,6 +1064,10 @@ func (g *Golibrtsp) Start(ctx context.Context, streamType string, queue *packets
 					}
 				}
 
+				// Preserve the decoded access unit (in decode order) for DTS
+				// extraction before we rewrite it into the filtered/annexb form.
+				decodedAU := au
+
 				filteredAU = [][]byte{
 					{byte(h265.NALUType_AUD_NUT) << 1, 1, 0x50},
 				}
@@ -902,6 +1110,9 @@ func (g *Golibrtsp) Start(ctx context.Context, streamType string, queue *packets
 					return
 				}
 
+				// Composition time offset (PTS - DTS) in milliseconds; see H264 handler.
+				compositionOffset := compositionOffsetMs(h265DTSExtractor, decodedAU, pts2, g.VideoH265Forma.ClockRate())
+
 				pkt := packets.Packet{
 					IsKeyFrame:      isRandomAccess,
 					Packet:          rtppkt,
@@ -909,7 +1120,7 @@ func (g *Golibrtsp) Start(ctx context.Context, streamType string, queue *packets
 					Time:            pts2,
 					TimeLegacy:      pts,
 					CurrentTime:     time.Now().UnixMilli(),
-					CompositionTime: pts2,
+					CompositionTime: compositionOffset,
 					Idx:             g.VideoH265Index,
 					IsVideo:         true,
 					IsAudio:         false,
@@ -935,7 +1146,11 @@ func (g *Golibrtsp) Start(ctx context.Context, streamType string, queue *packets
 					}
 				}
 
+				writeStart := time.Now()
 				queue.WritePacket(pkt)
+				// Records WritePacket() blocking time and frame arrival cadence so
+				// we can tell back-pressure from a network/camera stall.
+				g.health.observePacket(streamType, time.Since(writeStart))
 
 				// This will check if we need to stop the thread,
 				// because of a reconfiguration.

machinery/src/capture/main.go

@@ -140,23 +140,8 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
 				if start && // If already recording and current frame is a keyframe and we should stop recording
 					nextPkt.IsKeyFrame && (startRecording+postRecording-now <= 0 || now-startRecording > maxRecordingPeriod-500) {
 
-					pts := convertPTS(pkt.TimeLegacy)
-					if pkt.IsVideo {
-						// Write the last packet
-						if err := mp4Video.AddSampleToTrack(videoTrack, pkt.IsKeyFrame, pkt.Data, pts); err != nil {
-							log.Log.Error("capture.main.HandleRecordStream(continuous): " + err.Error())
-						}
-					} else if pkt.IsAudio {
-						// Write the last packet
-						if pkt.Codec == "AAC" {
-							if err := mp4Video.AddSampleToTrack(audioTrack, pkt.IsKeyFrame, pkt.Data, pts); err != nil {
-								log.Log.Error("capture.main.HandleRecordStream(continuous): " + err.Error())
-							}
-						} else if pkt.Codec == "PCM_MULAW" {
-							// TODO: transcode to AAC, some work to do..
-							log.Log.Debug("capture.main.HandleRecordStream(continuous): no AAC audio codec detected, skipping audio track.")
-						}
-					}
+					// Write the last packet before closing the recording.
+					writeSampleToMP4(mp4Video, videoTrack, audioTrack, pkt)
 
 					// Close mp4
 					if len(mp4Video.SPSNALUs) == 0 && len(configuration.Config.Capture.IPCamera.SPSNALUs) > 0 {
@@ -311,43 +296,12 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
 						log.Log.Debug("capture.main.HandleRecordStream(continuous): no AAC audio codec detected, skipping audio track.")
 					}
 
-					pts := convertPTS(pkt.TimeLegacy)
-					if pkt.IsVideo {
-						if err := mp4Video.AddSampleToTrack(videoTrack, pkt.IsKeyFrame, pkt.Data, pts); err != nil {
-							log.Log.Error("capture.main.HandleRecordStream(continuous): " + err.Error())
-						}
-					} else if pkt.IsAudio {
-						if pkt.Codec == "AAC" {
-							if err := mp4Video.AddSampleToTrack(audioTrack, pkt.IsKeyFrame, pkt.Data, pts); err != nil {
-								log.Log.Error("capture.main.HandleRecordStream(continuous): " + err.Error())
-							}
-						} else if pkt.Codec == "PCM_MULAW" {
-							// TODO: transcode to AAC, some work to do..
-							// We might need to use ffmpeg to transcode the audio to AAC.
-							// For now we will skip the audio track.
-							log.Log.Debug("capture.main.HandleRecordStream(continuous): no AAC audio codec detected, skipping audio track.")
-						}
-					}
+					writeSampleToMP4(mp4Video, videoTrack, audioTrack, pkt)
 					recordingStatus = "started"
 
 				} else if start {
 
-					pts := convertPTS(pkt.TimeLegacy)
-					if pkt.IsVideo {
-						// New method using new mp4 library
-						if err := mp4Video.AddSampleToTrack(videoTrack, pkt.IsKeyFrame, pkt.Data, pts); err != nil {
-							log.Log.Error("capture.main.HandleRecordStream(continuous): " + err.Error())
-						}
-					} else if pkt.IsAudio {
-						if pkt.Codec == "AAC" {
-							if err := mp4Video.AddSampleToTrack(audioTrack, pkt.IsKeyFrame, pkt.Data, pts); err != nil {
-								log.Log.Error("capture.main.HandleRecordStream(continuous): " + err.Error())
-							}
-						} else if pkt.Codec == "PCM_MULAW" {
-							// TODO: transcode to AAC, some work to do..
-							log.Log.Debug("capture.main.HandleRecordStream(continuous): no AAC audio codec detected, skipping audio track.")
-						}
-					}
+					writeSampleToMP4(mp4Video, videoTrack, audioTrack, pkt)
 				}
 				pkt = nextPkt
 			}
@@ -571,29 +525,7 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
 						start = true
 					}
 					if start {
-						pts := convertPTS(pkt.TimeLegacy)
-						if pkt.IsVideo {
-							log.Log.Debug("capture.main.HandleRecordStream(motiondetection): add video sample")
-							if mp4Video != nil {
-								if err := mp4Video.AddSampleToTrack(videoTrack, pkt.IsKeyFrame, pkt.Data, pts); err != nil {
-									log.Log.Error("capture.main.HandleRecordStream(motiondetection): " + err.Error())
-								}
-							}
-						} else if pkt.IsAudio {
-							log.Log.Debug("capture.main.HandleRecordStream(motiondetection): add audio sample")
-							if pkt.Codec == "AAC" {
-								if mp4Video != nil {
-									if err := mp4Video.AddSampleToTrack(audioTrack, pkt.IsKeyFrame, pkt.Data, pts); err != nil {
-										log.Log.Error("capture.main.HandleRecordStream(motiondetection): " + err.Error())
-									}
-								}
-							} else if pkt.Codec == "PCM_MULAW" {
-								// TODO: transcode to AAC, some work to do..
-								// We might need to use ffmpeg to transcode the audio to AAC.
-								// For now we will skip the audio track.
-								log.Log.Debug("capture.main.HandleRecordStream(motiondetection): no AAC audio codec detected, skipping audio track.")
-							}
-						}
+						writeSampleToMP4(mp4Video, videoTrack, audioTrack, pkt)
 					}
 
 					pkt = nextPkt
@@ -867,6 +799,41 @@ func convertPTS(v time.Duration) uint64 {
 	return uint64(v.Milliseconds())
 }
 
+// writeSampleToMP4 writes a single capture packet to the fragmented MP4.
+//
+// For video it derives the decode timestamp (DTS) from the packet PTS using the
+// per-packet composition offset (PTS - DTS), which is non-zero only for streams
+// that contain B-frames. Passing the monotonic DTS as the sample timestamp keeps
+// the fragment timeline (tfdt/sidx) monotonic, while the composition offset is
+// forwarded so frames are still presented in PTS order.
+func writeSampleToMP4(mp4Video *video.MP4, videoTrack, audioTrack uint32, pkt packets.Packet) {
+	if mp4Video == nil {
+		return
+	}
+
+	pts := convertPTS(pkt.TimeLegacy)
+
+	if pkt.IsVideo {
+		compositionOffset := pkt.CompositionTime
+		dts := pts
+		if compositionOffset > 0 && uint64(compositionOffset) <= pts {
+			dts = pts - uint64(compositionOffset)
+		}
+		if err := mp4Video.AddSampleToTrack(videoTrack, pkt.IsKeyFrame, pkt.Data, dts, compositionOffset); err != nil {
+			log.Log.Error("capture.main.writeSampleToMP4(): " + err.Error())
+		}
+	} else if pkt.IsAudio {
+		if pkt.Codec == "AAC" {
+			if err := mp4Video.AddSampleToTrack(audioTrack, pkt.IsKeyFrame, pkt.Data, pts, 0); err != nil {
+				log.Log.Error("capture.main.writeSampleToMP4(): " + err.Error())
+			}
+		} else if pkt.Codec == "PCM_MULAW" {
+			// TODO: transcode to AAC, some work to do..
+			log.Log.Debug("capture.main.writeSampleToMP4(): no AAC audio codec detected, skipping audio track.")
+		}
+	}
+}
+
 /*func convertPTS2(v int64) uint64 {
 	return uint64(v) / 100
 }*/

machinery/src/cloud/kerberos_hub.go

@@ -68,9 +68,9 @@ func UploadKerberosHub(configuration *models.Configuration, fileName string) (bo
 		tr := &http.Transport{
 			TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
 		}
-		client = &http.Client{Transport: tr}
+		client = &http.Client{Transport: tr, CheckRedirect: stripHubCredentialsOnCrossHostRedirect}
 	} else {
-		client = &http.Client{}
+		client = &http.Client{CheckRedirect: stripHubCredentialsOnCrossHostRedirect}
 	}
 
 	resp, err := client.Do(req)
@@ -129,3 +129,20 @@ func UploadKerberosHub(configuration *models.Configuration, fileName string) (bo
 	log.Log.Info(errorMessage)
 	return false, true, errors.New(errorMessage)
 }
+
+// stripHubCredentialsOnCrossHostRedirect removes the custom Kerberos Hub
+// credential headers on a redirect that crosses to a different host. net/http
+// already strips the standard sensitive headers (Authorization, Cookie,
+// WWW-Authenticate) on a cross-host redirect, but it does NOT strip
+// custom-named headers, so without this the Hub private/public keys would be
+// forwarded to any host the configured HubURI redirects to.
+func stripHubCredentialsOnCrossHostRedirect(req *http.Request, via []*http.Request) error {
+	if len(via) == 0 {
+		return nil
+	}
+	if req.URL.Host != via[0].URL.Host {
+		req.Header.Del("X-Kerberos-Hub-PrivateKey")
+		req.Header.Del("X-Kerberos-Hub-PublicKey")
+	}
+	return nil
+}

machinery/src/components/Kerberos.go

@@ -2,6 +2,7 @@ package components
 
 import (
 	"context"
+	"fmt"
 	"os"
 	"strconv"
 	"sync/atomic"
@@ -446,6 +447,37 @@ func RunAgent(configDirectory string, configuration *models.Configuration, commu
 	return status
 }
 
+// packetAgeString returns a human readable age (e.g. "12s") since the last
+// packet timestamp stored in the given atomic.Value, or "unknown" when no
+// packet has been received yet. Used to add context to watchdog restart logs.
+func packetAgeString(timer *atomic.Value) string {
+	if timer == nil {
+		return "unknown"
+	}
+
+	// atomic.Value panics on Load() if it was never initialized via Store().
+	var v any
+	func() {
+		defer func() {
+			if recover() != nil {
+				v = nil
+			}
+		}()
+		v = timer.Load()
+	}()
+
+	last, ok := v.(int64)
+	if !ok || last == 0 {
+		return "unknown"
+	}
+
+	age := time.Now().Unix() - last
+	if age < 0 {
+		age = 0
+	}
+	return strconv.FormatInt(age, 10) + "s"
+}
+
 // ControlAgent will check if the camera is still connected, if not it will restart the agent.
 // In the other thread we are keeping track of the number of packets received, and particular the keyframe packets.
 // Once we are not receiving any packets anymore, we will restart the agent.
@@ -480,7 +512,8 @@ func ControlAgent(communication *models.Communication) {
 
 				// After 15 seconds without activity this is thrown..
 				if occurence == 3 {
-					log.Log.Info("components.Kerberos.ControlAgent(): Restarting machinery because of blocking mainstream.")
+					log.Log.Info(fmt.Sprintf("components.Kerberos.ControlAgent(): Restarting machinery because of blocking mainstream. (stalledKeyframeCounter=%d, lastPacket=%s ago, isConfiguring=%t)",
+						packetsR, packetAgeString(communication.LastPacketTimer), communication.IsConfiguring.IsSet()))
 					select {
 					case communication.HandleBootstrap <- "restart":
 						log.Log.Info("components.Kerberos.ControlAgent(): Restarting machinery because of blocking substream.")
@@ -507,6 +540,8 @@ func ControlAgent(communication *models.Communication) {
 
 					// After 15 seconds without activity this is thrown..
 					if occurenceSub == 3 {
+						log.Log.Info(fmt.Sprintf("components.Kerberos.ControlAgent(): substream stalled (stalledKeyframeCounter=%d, lastPacket=%s ago, isConfiguring=%t)",
+							packetsSubR, packetAgeString(communication.LastPacketTimerSub), communication.IsConfiguring.IsSet()))
 						select {
 						case communication.HandleBootstrap <- "restart":
 							log.Log.Info("components.Kerberos.ControlAgent(): Restarting machinery because of blocking substream.")
@@ -752,10 +787,24 @@ func GetSnapshotRaw(c *gin.Context, captureDevice *capture.Capture, configuratio
 // @Description Get the current configuration.
 // @Success 200
 func GetConfig(c *gin.Context, captureDevice *capture.Capture, configuration *models.Configuration, communication *models.Communication) {
-	// We'll try to get a snapshot from the camera.
-	base64Image := capture.Base64Image(captureDevice, communication, configuration)
-	if base64Image != "" {
-		communication.Image = base64Image
+	// We'll try to get a fresh snapshot from the camera. Capturing a snapshot
+	// reads a keyframe from the live stream, which blocks until one arrives.
+	// When the camera is offline or the stream is stalled (no packets being
+	// received) this would block the /config endpoint indefinitely, making the
+	// agent appear unreachable even though its HTTP server is healthy. We
+	// therefore bound the snapshot fetch with a short timeout and fall back to
+	// the last cached snapshot, so /config always responds promptly.
+	snapshot := make(chan string, 1)
+	go func() {
+		snapshot <- capture.Base64Image(captureDevice, communication, configuration)
+	}()
+	select {
+	case base64Image := <-snapshot:
+		if base64Image != "" {
+			communication.Image = base64Image
+		}
+	case <-time.After(2 * time.Second):
+		log.Log.Info("components.Kerberos.GetConfig(): snapshot timed out (stream stalled or camera offline), returning configuration with the last cached snapshot.")
 	}
 
 	c.JSON(200, gin.H{

machinery/src/config/main.go

@@ -193,8 +193,104 @@ func OpenConfig(configDirectory string, configuration *models.Configuration) {
 	return
 }
 
-// This function will override the configuration with environment variables.
+// OverrideWithEnvironmentVariables builds the effective configuration from the
+// environment variables.
+//
+// In ConfigMap/standalone mode (DEPLOYMENT empty or "agent") the global
+// configuration is delivered as GLOBAL_AGENT_* environment variables and the
+// per-agent configuration as AGENT_* environment variables. We parse them into
+// the separate global and custom configurations and build the effective
+// configuration as "global overridden by custom", mirroring the MongoDB-backed
+// factory behaviour. This keeps the global and per-agent (custom) configuration
+// separated so the factory edit page can distinguish inherited global settings
+// from per-agent overrides.
 func OverrideWithEnvironmentVariables(configuration *models.Configuration) {
+	if os.Getenv("DEPLOYMENT") == "" || os.Getenv("DEPLOYMENT") == "agent" {
+		initConfigPointers(&configuration.Config)
+
+		// Parse the global configuration from the GLOBAL_AGENT_* variables.
+		globalWrap := &models.Configuration{Config: configuration.GlobalConfig}
+		initConfigPointers(&globalWrap.Config)
+		applyAgentEnvVars(globalWrap, "GLOBAL_", false)
+		configuration.GlobalConfig = globalWrap.Config
+
+		// Parse the per-agent (custom) configuration from the AGENT_* variables.
+		// In ConfigMap mode the per-agent overrides are delivered exclusively
+		// through AGENT_* environment variables, so we must start from an empty
+		// configuration rather than the bundled config.json that OpenConfig loaded
+		// into CustomConfig. Otherwise defaults from that file (e.g. cloud="s3")
+		// would leak into the custom config and be mistaken for explicit per-agent
+		// overrides, hiding inherited global settings (the factory edit page would
+		// show the local default instead of the inherited global persistence).
+		customBase := configuration.CustomConfig
+		if isConfigMapMode() {
+			customBase = models.Config{}
+		}
+		customWrap := &models.Configuration{Config: customBase}
+		initConfigPointers(&customWrap.Config)
+		applyAgentEnvVars(customWrap, "", false)
+		configuration.CustomConfig = customWrap.Config
+
+		// Build the effective configuration: global base, then per-agent
+		// overrides on top. Defaults (e.g. signing) are applied on the last
+		// pass only.
+		applyAgentEnvVars(configuration, "GLOBAL_", false)
+		applyAgentEnvVars(configuration, "", true)
+	} else {
+		// Factory/MongoDB mode: the global and custom configurations are already
+		// loaded and merged from MongoDB; we only override the effective
+		// configuration with any AGENT_* environment variables.
+		applyAgentEnvVars(configuration, "", true)
+	}
+}
+
+// isConfigMapMode reports whether the agent is running in ConfigMap mode, i.e.
+// whether a global configuration layer is delivered separately through
+// GLOBAL_AGENT_* environment variables. In that mode the per-agent (custom)
+// configuration must be built solely from the AGENT_* overrides and must not be
+// seeded with the bundled config.json defaults, so that inherited global
+// settings remain distinguishable from explicit per-agent overrides.
+func isConfigMapMode() bool {
+	for _, env := range os.Environ() {
+		if strings.HasPrefix(env, "GLOBAL_AGENT_") {
+			return true
+		}
+	}
+	return false
+}
+
+// initConfigPointers ensures all pointer sub-structs are non-nil so that the
+// environment-variable parsing can assign into them without dereferencing a nil
+// pointer.
+func initConfigPointers(config *models.Config) {
+	if config.KStorage == nil {
+		config.KStorage = &models.KStorage{}
+	}
+	if config.KStorageSecondary == nil {
+		config.KStorageSecondary = &models.KStorage{}
+	}
+	if config.S3 == nil {
+		config.S3 = &models.S3{}
+	}
+	if config.Encryption == nil {
+		config.Encryption = &models.Encryption{}
+	}
+	if config.Signing == nil {
+		config.Signing = &models.Signing{}
+	}
+	if config.Dropbox == nil {
+		config.Dropbox = &models.Dropbox{}
+	}
+	if config.Region == nil {
+		config.Region = &models.Region{}
+	}
+}
+
+// applyAgentEnvVars applies the AGENT_* environment variables (optionally
+// carrying the given prefix, e.g. "GLOBAL_") onto configuration.Config. When
+// applyDefaults is true, defaults (such as the signing key) are applied after
+// parsing; this should only be done for the effective configuration.
+func applyAgentEnvVars(configuration *models.Configuration, prefix string, applyDefaults bool) {
 	environmentVariables := os.Environ()
 
 	// Initialize the configuration for some new fields.
@@ -203,9 +299,10 @@ func OverrideWithEnvironmentVariables(configuration *models.Configuration) {
 	}
 
 	for _, env := range environmentVariables {
-		if strings.Contains(env, "AGENT_") {
-			key := strings.Split(env, "=")[0]
-			value := os.Getenv(key)
+		fullKey := strings.SplitN(env, "=", 2)[0]
+		if strings.HasPrefix(fullKey, prefix+"AGENT_") && !(prefix == "" && strings.HasPrefix(fullKey, "GLOBAL_AGENT_")) {
+			key := strings.TrimPrefix(fullKey, prefix)
+			value := os.Getenv(fullKey)
 			switch key {
 
 			/* General configuration */
@@ -545,13 +642,20 @@ func OverrideWithEnvironmentVariables(configuration *models.Configuration) {
 		}
 	}
 
-	// Signing is a new feature, so if empty we set default values.
-	if configuration.Config.Signing == nil || configuration.Config.Signing.PrivateKey == "" {
+	// Signing is a new feature, so if empty we set default values. Only applied
+	// for the effective configuration (applyDefaults), not for the separate
+	// global/custom views.
+	if applyDefaults && (configuration.Config.Signing == nil || configuration.Config.Signing.PrivateKey == "") {
 		configuration.Config.Signing = &models.Signing{
 			Enabled:    "true",
 			PrivateKey: "-----BEGIN PRIVATE KEY-----\nMIIJQgIBADANBgkqhkiG9w0BAQEFAASCCSwwggkoAgEAAoICAQDoSxjyw08lRxF4Yoqmcaewjq3XjB55dMy4tlN5MGLdr8aAPuNR9Mwh3jlh1bDpwQXNgZkHDV/q9bpdPGGi7SQo2xw+rDuo5Y1f3wdzz+iuCTPbzoGFalE+1PZlU5TEtUtlbt7MRc4pxTaLP3u0P3EtW3KnzcUarcJWZJYxzv7gqVNCA/47BN+1ptqjwz3LAlah5yaftEvVjkaANOsafUswbS4VT44XfSlbKgebORCKDuNgQiyhuV5gU+J0TOaqRWwwMAWV0UoScyJLfhHRBCrUwrCUTwqH9jfkB7pgRFsYoZJd4MKMeHJjFSum+QXCBqInSnwu8c2kJChiLMWqJ+mhpTdfUAmSkeUSStfbbcavIPbDABvMgzOcmYMIVXXe57twU0xdu3AqWLtc9kw1BkUgZblM9pSSpYrIDheEyMs2/hiLgXsIaM0nVQtqwrA7rbeEGuPblzA6hvHgwN9K6HaBqdlGSlpYZ0v3SWIMwmxRB+kIojlyuggm8Qa4mqL97GFDGl6gOBGlNUFTBUVEa3EaJ7NJpGobRGsh/9dXzcW4aYmT9WxlzTlIKksI1ro6KdRfuVWfEs4AnG8bVEJmofK8EUrueB9IdXlcJZB49xolnOZPFohtMe/0U7evQOQP3sZnX+KotCsE7OXJvL09oF58JKoqmK9lPp0+pFBU4g6NjQIDAQABAoICAA+RSWph1t+q5R3nxUxFTYMrhv5IjQe2mDxJpF3B409zolC9OHxgGUisobTY3pBqs0DtKbxUeH2A0ehUH/axEosWHcz3cmIbgxHE9kdlJ9B3Lmss6j/uw+PWutu1sgm5phaIFIvuNNRWhPB6yXUwU4sLRat1+Z9vTmIQiKdtLIrtJz/n2VDvrJxn1N+yAsE20fnrksFKyZuxVsJaZPiX/t5Yv1/z0LjFjVoL7GUA5/Si7csN4ftqEhUrkNr2BvcZlTyffrF4lZCXrtl76RNUaxhqIu3H0gFbV2UfBpuckkfAhNRpXJ4iFSxm4nQbk4ojV8+l21RFOBeDN2Z7Ocu6auP5MnzpopR66vmDCmPoid498VGgDzFQEVkOar8WAa4v9h85QgLKrth6FunmaWJUT6OggQD3yY58GSwp5+ARMETMBP2x6Eld+PGgqoJvPT1+l/e9gOw7/SJ+Wz6hRXZAm/eiXMppHtB7sfea5rscNanPjJkK9NvPM0MX9cq/iA6QjXuETkMbubjo+Cxk3ydZiIQmWQDAx/OgxTyHbeRCVhLPcAphX0clykCuHZpI9Mvvj643/LoE0mjTByWJXf/WuGJA8ElHkjSdokVJ7jumz8OZZHfq0+V7+la2opsObeQANHW5MLWrnHlRVzTGV0IRZDXh7h1ptUJ4ubdvw/GJ2NeTAoIBAQD0lXXdjYKWC4uZ4YlgydP8b1CGda9cBV5RcPt7q9Ya1R2E4ieYyohmzltopvdaOXdsTZzhtdzOzKF+2qNcbBKhBTleYZ8GN5RKbo7HwXWpzfCTjseKHOD/QPwvBKXzLVWNtXn1NrLR79Rv0wbkYF6DtoqpEPf5kMs4bx79yW+mz8FUgdEeMjKphx6Jd5RYlTUxS64K6bnK7gjHNCF2cwdxsh4B6EB649GKeNz4JXi+oQBmOcX5ncXnkJrbju+IjtCkQ40HINVNdX7XeEaaw6KGaImVjw61toPUuDaioYUojufayoyXaUJnDbHQ2tNekEpq5iwnenZCbUKWmSeRe7dLAoIBAQDzIscYujsrmPxiTj2prhG0v36NRNP99mShnnJGowiIs+UBS0EMdOmBFa2sC9uFs/VnreQNYPDJdfr7O5VK9kfbH/PSiiKJ+wVebfdAlWkJYH27JN2Kl2l/OsvRVelNvF3BWIYF46qzGxIM0axaz3T2ZAJ9SrUgeAYhak6uyM4fbexEWXxDgPGu6C0jB6IAzmHJnnh+j5+4ZXqjVyUxBYtUsWXF/TXomVcT9jxj7aUmS2/Us0XTVOVNpALqqYcekrzsX/wX0OEi5HkivYXHcNaDHx3NuUf6KdYof5DwPUM76qe+5/kWlSIHP3M6rIFK3pYFUnkHn2E8jNWcO97Aio+HAoIBAA+bcff/TbPxbKkXIUMR3fsfx02tONFwbkJYKVQM9Q6lRsrx+4Dee7HDvUWCUgpp3FsG4NnuVvbDTBLiNMZzBwVLZgvFwvYMmePeBjJs/+sj/xQLamQ/z4O6S91cOJK589mlGPEy2lpXKYExQCFWnPFetp5vPMOqH62sOZgMQJmubDHOTt/UaDM1Mhenj8nPS6OnpqV/oKF4awr7Ip+CW5k/unZ4sZSl8PsbF06mZXwUngfn6+Av1y8dpSQZjONz6ZBx1w/7YmEc/EkXnbnGfhqBlTX7+P5TdTofvyzFjc+2vsjRYANRbjFRSGWBcTd5kaYcpfim8eDvQ+6EO2gnMt0CggEAH2ln1Y8B5AEQ4lZ/avOdP//ZhsDUrqPtnl/NHckkahzrwj4JumVEYbP+SxMBGoYEd4+kvgG/OhfvBBRPlm65G9tF8fZ8vdzbdba5UfO7rUV1GP+LS8OCErjy6imySaPDbR5Vul8Oh7NAor1YCidxUf/bvnovanF3QUvtvHEfCDp4YuA4yLPZBaLjaforePUw9w5tPNSravRZYs74dBvmQ1vj7S9ojpN5B5AxfyuNwaPPX+iFZec69MvywISEe3Ozysof1Kfc3lgsOkvIA9tVK32SqSh93xkWnQbWH+OaUxxe7bAko0FDMzKEXZk53wVg1nEwR8bUljEPy+6EOdXs8wKCAQEAsEOWYMY5m7HkeG2XTTvX7ECmmdGl/c4ZDVwzB4IPxqUG7XfLmtsON8YoKOEUpJoc4ANafLXzmU+esUGbH4Ph22IWgP9jzws7jxaN/Zoku64qrSjgEZFTRIpKyhFk/ImWbS9laBW4l+m0tqTTRqoE0QEJf/2uv/04q65zrA70X9z2+KTrAtqOiRQPWl/IxRe9U4OEeGL+oD+YlXKCDsnJ3rwUIOZgJx0HWZg7K35DKwqs1nVi56FBdljiTRKAjVLRedjgDCSfGS1yUZ3krHzpaPt1qgnT3rdtYcIdbYDr66V2/gEEaz6XMGHuTk/ewjzUJxq9UTVeXOCbkRPXgVJg1w==\n-----END PRIVATE KEY-----",
 		}
 	}
+
+	// When the agent is configured through environment variables the global and
+	// custom configurations were already parsed separately (see
+	// OverrideWithEnvironmentVariables), so there is no need to mirror the
+	// effective configuration into CustomConfig anymore.
 }
 
 func SaveConfig(configDirectory string, config models.Config, configuration *models.Configuration, communication *models.Communication) error {

machinery/src/packets/packet.go

@@ -14,7 +14,7 @@ type Packet struct {
 	IsKeyFrame      bool   // video packet is key frame
 	Idx             int8   // stream index in container format
 	Codec           string // codec name
-	CompositionTime int64  // packet presentation time minus decode time for H264 B-Frame
+	CompositionTime int64  // composition offset (PTS - DTS) in milliseconds, non-zero for H264/H265 B-frames
 	Time            int64  // packet decode time
 	TimeLegacy      time.Duration
 	CurrentTime     int64  // current time in milliseconds (UNIX timestamp)

machinery/src/routers/http/jwt_middleware.go

@@ -14,7 +14,14 @@ import (
 func JWTMiddleWare() jwt.GinJWTMiddleware {
 
 	identityKey := "id"
-	myKey := "TOBECHANGED"
+	// Allow the JWT signing secret to be configured through an environment
+	// variable so that tokens issued by another service (e.g. the Kerberos
+	// Factory) can be validated by the agent. Falls back to the historic
+	// default to preserve backwards compatibility.
+	myKey := os.Getenv("AGENT_JWT_SECRET")
+	if myKey == "" {
+		myKey = "TOBECHANGED"
+	}
 
 	m := jwt.GinJWTMiddleware{
 		Realm:       "kerberosio",
@@ -106,7 +113,11 @@ func JWTMiddleWare() jwt.GinJWTMiddleware {
 		// - "query:<name>"
 		// - "cookie:<name>"
 		// - "param:<name>"
-		TokenLookup: "header: Authorization, query: token, cookie: jwt",
+		// X-Authorization is included because requests proxied through the
+		// Kubernetes apiserver service-proxy have their Authorization header
+		// consumed by the apiserver; the original bearer token is forwarded in
+		// the X-Authorization header instead.
+		TokenLookup: "header: Authorization, header: X-Authorization, query: token, cookie: jwt",
 		// TokenLookup: "query:token",
 		// TokenLookup: "cookie:token",
 

machinery/src/routers/mqtt/main.go

@@ -11,6 +11,7 @@ import (
 	"math/rand"
 	"strconv"
 	"strings"
+	"sync"
 	"time"
 
 	mqtt "github.com/eclipse/paho.mqtt.golang"
@@ -170,11 +171,45 @@ func ConfigureMQTT(configDirectory string, configuration *models.Configuration,
 	return nil
 }
 
-// maxSignalingAge is the maximum age of a WebRTC signaling message (request-hd-stream,
-// receive-hd-candidates) before it is considered stale and discarded. With CleanSession=false
-// the MQTT broker may replay queued messages from previous sessions; this prevents the agent
-// from setting up peer connections for viewers that are no longer waiting.
-const maxSignalingAge = 30 * time.Second
+// recentHDSessions tracks recently-seen WebRTC viewer session IDs so we can
+// dedupe duplicate request-hd-stream messages without relying on the broker's
+// (and the viewer's) wall clock. The viewer's offer-republish loop can fire
+// the same request several times for the same session_id while waiting for an
+// answer; the broker can also redeliver a message after a reconnect with
+// CleanSession=false. In both cases we want to handle the session exactly
+// once.
+//
+// Entries expire after recentHDSessionTTL. The map is small (one entry per
+// active viewer over the TTL window) so a periodic sweep is sufficient.
+const recentHDSessionTTL = 60 * time.Second
+
+var (
+	recentHDSessionsMu sync.Mutex
+	recentHDSessions   = make(map[string]time.Time)
+)
+
+// markHDSessionSeen returns true if this session_id was already processed
+// within the TTL window (i.e. this message should be treated as a duplicate).
+// It also opportunistically prunes expired entries.
+func markHDSessionSeen(sessionID string) bool {
+	if sessionID == "" {
+		return false
+	}
+	recentHDSessionsMu.Lock()
+	defer recentHDSessionsMu.Unlock()
+	now := time.Now()
+	// Lazy GC — cheap given the expected map size.
+	for k, t := range recentHDSessions {
+		if now.Sub(t) > recentHDSessionTTL {
+			delete(recentHDSessions, k)
+		}
+	}
+	if _, exists := recentHDSessions[sessionID]; exists {
+		return true
+	}
+	recentHDSessions[sessionID] = now
+	return false
+}
 
 func MQTTListenerHandler(mqttClient mqtt.Client, hubKey string, configDirectory string, configuration *models.Configuration, communication *models.Communication) {
 	if hubKey == "" {
@@ -282,16 +317,13 @@ func MQTTListenerHandler(mqttClient mqtt.Client, hubKey string, configDirectory
 				// We'll find out which message we received, and act accordingly.
 				log.Log.Info("routers.mqtt.main.MQTTListenerHandler(): received message with action: " + payload.Action)
 
-				// For time-sensitive WebRTC signaling messages, discard stale ones that may
-				// have been queued by the broker while CleanSession=false.
-				if payload.Action == "request-hd-stream" || payload.Action == "receive-hd-candidates" {
-					messageAge := time.Since(time.Unix(message.Timestamp, 0))
-					if messageAge > maxSignalingAge {
-						log.Log.Info("routers.mqtt.main.MQTTListenerHandler(): discarding stale " + payload.Action +
-							" message (age: " + messageAge.Round(time.Second).String() + ")")
-						return
-					}
-				}
+				// NOTE: We intentionally do NOT discard request-hd-stream /
+				// receive-hd-candidates messages based on a wall-clock age. The
+				// viewer and agent clocks can drift (especially on embedded
+				// devices), which previously caused valid requests to be
+				// silently dropped and forced the user to refresh the page.
+				// Duplicate handling for request-hd-stream is done by session_id
+				// inside HandleRequestHDStream (see markHDSessionSeen).
 
 				switch payload.Action {
 				case "record":
@@ -536,6 +568,15 @@ func HandleRequestHDStream(mqttClient mqtt.Client, hubKey string, payload models
 
 	if requestHDStreamPayload.Timestamp != 0 {
 		if communication.CameraConnected {
+			// Dedupe by session_id: the viewer republishes its offer while
+			// waiting for an answer (and the broker may redeliver), and we
+			// don't want to spawn multiple peer connections for the same
+			// browser session.
+			if markHDSessionSeen(requestHDStreamPayload.SessionID) {
+				log.Log.Info("routers.mqtt.main.HandleRequestHDStream(): duplicate request for session " +
+					requestHDStreamPayload.SessionID + ", ignoring")
+				return
+			}
 			// Set the Hub key, so we can send back the answer.
 			requestHDStreamPayload.HubKey = hubKey
 			if communication.HandleLiveHDHandshake == nil {

machinery/src/video/mp4.go

@@ -32,6 +32,16 @@ const MacEpochOffset uint64 = 2082844800
 // resulting in ~3 second fragments (assuming a typical GOP interval).
 const FragmentDurationMs = 3000
 
+// SeamGapDivisor controls loop-seam detection. A keyframe is treated as an
+// upstream loop/restart seam when it arrives in less than (previous keyframe
+// interval / SeamGapDivisor) — i.e. far sooner than the established keyframe
+// cadence. Comparing against the *previous* interval (rather than a fixed
+// millisecond threshold) makes the check scale automatically with the camera's
+// configured GOP size: it works the same whether keyframes are 0.5s, 1s, 2s or
+// more apart, and does not misfire on legitimately short-GOP or all-intra
+// streams (where every interval is similar, so none looks anomalously short).
+const SeamGapDivisor = 2
+
 type MP4 struct {
 	// FileName is the name of the file
 	FileName                string
@@ -56,7 +66,6 @@ type MP4 struct {
 	FreeBoxSize             int64
 	FragmentStartRawPTS     uint64            // Raw PTS for timing when to flush fragments
 	FragmentStartDTS        uint64            // Accumulated VideoTotalDuration at fragment start (matches tfdt)
-	LastKeyframeRawPTS      uint64            // Raw PTS of the most recently seen keyframe (in any fragment)
 	MoofBoxes               int64             // Number of moof boxes in the file
 	MoofBoxSizes            []int64           // Sizes of each moof box
 	SegmentDurations        []uint64          // Duration of each segment in timescale units
@@ -75,6 +84,20 @@ type MP4 struct {
 	TotalKeyframesWritten   int               // Total keyframes written to trun boxes
 	FragmentKeyframeCount   int               // Keyframes in the current fragment
 	PendingSampleIsKeyframe bool              // Whether the pending video sample is a keyframe
+	LastKeyframeRawPTS      uint64            // Raw PTS of the most recently seen keyframe (across fragments)
+	LastKeyframeGapMs       uint64            // Interval (ms) between the two most recent keyframes; reference cadence for seam detection
+	gopBuffer               []bufferedSample  // Current, not-yet-committed GOP (video frames + interleaved audio), held so a loop-seam GOP can be dropped before it reaches the file
+}
+
+// bufferedSample is a single sample (video or audio) held in the current-GOP
+// buffer until we know whether the GOP should be committed to the file or
+// dropped as an upstream loop-seam artifact (see AddSampleToTrack).
+type bufferedSample struct {
+	trackID           uint32
+	isKeyframe        bool
+	data              []byte
+	pts               uint64
+	compositionOffset int64
 }
 
 // NewMP4 creates a new MP4 object.
@@ -234,24 +257,6 @@ func (mp4 *MP4) flushPendingVideoSample(nextPTS uint64) bool {
 	var duration uint64
 	if nextPTS > 0 && nextPTS > mp4.VideoFullSample.DecodeTime {
 		duration = nextPTS - mp4.VideoFullSample.DecodeTime
-		// Guard against forward PTS jumps (e.g. when looping a source MP4
-		// through virtual-rtsp the upstream ffmpeg may insert a large offset
-		// at the loop boundary, or the RTSP stream may stall briefly).
-		// Without this clamp the sample gets a huge duration which appears
-		// as a discontinuity in the trun/sidx/mvhd and causes browsers
-		// (Video.js / MSE) to abort playback with a "media corruption"
-		// error around the loop boundary.
-		var maxPlausible uint64 = 1000 // 1 second hard ceiling
-		if mp4.LastVideoSampleDTS > 0 && mp4.LastVideoSampleDTS*10 < maxPlausible {
-			maxPlausible = mp4.LastVideoSampleDTS * 10
-		}
-		if duration > maxPlausible {
-			log.Log.Warning(fmt.Sprintf("mp4.flushPendingVideoSample(): video PTS jumped forward (nextPTS=%d, prevDTS=%d, gap=%d ms) - clamping to %d ms (likely source loop/stall discontinuity)", nextPTS, mp4.VideoFullSample.DecodeTime, duration, maxPlausible))
-			duration = mp4.LastVideoSampleDTS
-			if duration == 0 {
-				duration = 33
-			}
-		}
 	} else {
 		// No valid nextPTS (Close case) or PTS went backwards (jitter/discontinuity)
 		if nextPTS > 0 {
@@ -285,7 +290,110 @@ func (mp4 *MP4) flushPendingVideoSample(nextPTS uint64) bool {
 	return true
 }
 
-func (mp4 *MP4) AddSampleToTrack(trackID uint32, isKeyframe bool, data []byte, pts uint64) error {
+// AddSampleToTrack appends a sample to the given track.
+//
+// For video, pts is the decode timestamp (DTS, in milliseconds) and
+// compositionOffset is the composition time offset (PTS - DTS, in milliseconds).
+// The offset is non-zero only for streams that contain B-frames; it is written
+// as the sample's signed composition time offset so the decoder presents frames
+// in PTS order while the fragment timeline stays monotonic in DTS.
+//
+// For audio, pts is the sample timestamp and compositionOffset should be 0.
+//
+// Samples are not written straight through. Each video GOP is held in a small
+// buffer (gopBuffer) until the next keyframe arrives, so a GOP belonging to an
+// upstream source-loop / restart seam can be dropped before it ever reaches the
+// file. When a source MP4 is looped through virtual-rtsp
+// (ffmpeg `-stream_loop -1 -re`), the loop boundary leaves a truncated tail GOP
+// whose first inter-frame is incomplete: software decoders conceal the missing
+// macroblocks, but hardware decoders (macOS VideoToolbox) reject it with
+// kVTVideoDecoderBadDataErr (-12909) and MSE players (Video.js / Chromium /
+// Firefox) report media corruption, freezing playback at the seam (e.g. the
+// ~10s mark in the original recordings). The seam IDR that follows is a clean
+// random-access point, so dropping the truncated GOP lets playback continue
+// seamlessly. Holding back at most one GOP only delays on-disk fragments; for
+// any recording without a seam the finalized file is identical to the straight
+// pass-through output (Close flushes the final buffered GOP).
+func (mp4 *MP4) AddSampleToTrack(trackID uint32, isKeyframe bool, data []byte, pts uint64, compositionOffset int64) error {
+	isVideoKeyframe := isKeyframe && trackID == uint32(mp4.VideoTrack)
+	if !isVideoKeyframe {
+		// Part of the current GOP window (P/B frame or interleaved audio): hold it
+		// until the GOP is committed or dropped at the next video keyframe.
+		mp4.gopBuffer = append(mp4.gopBuffer, bufferedSample{
+			trackID:           trackID,
+			isKeyframe:        isKeyframe,
+			data:              data,
+			pts:               pts,
+			compositionOffset: compositionOffset,
+		})
+		return nil
+	}
+
+	// A video keyframe ends the GOP we have been buffering. Decide whether that
+	// buffered GOP is genuine (commit it) or the truncated tail GOP at an upstream
+	// loop/restart seam (drop it).
+	//
+	// The GOP size is configurable per camera, so we do NOT compare against a
+	// fixed millisecond threshold. Instead we compare this keyframe interval to
+	// the previous one and only flag a *sudden* shortening: a seam IDR arrives in
+	// less than (previous interval / SeamGapDivisor). Deriving the threshold from
+	// the observed cadence keeps detection correct for any configured GOP (0.5s,
+	// 1s, 2s, ...) and avoids false positives on steady short-GOP / all-intra
+	// streams (where consecutive intervals are similar, so none looks anomalously
+	// short). Because the reference is the immediately preceding interval, a burst
+	// of close keyframes only drops a single GOP instead of cascading.
+	seam := false
+	if mp4.LastKeyframeRawPTS > 0 && pts > mp4.LastKeyframeRawPTS {
+		gap := pts - mp4.LastKeyframeRawPTS
+		if mp4.LastKeyframeGapMs > 0 && gap*SeamGapDivisor < mp4.LastKeyframeGapMs {
+			seam = true
+			log.Log.Warning(fmt.Sprintf("mp4.AddSampleToTrack(): dropping truncated GOP at unexpectedly close keyframe (interval=%d ms, previous interval=%d ms, buffered samples=%d) - likely upstream loop/restart discontinuity", gap, mp4.LastKeyframeGapMs, len(mp4.gopBuffer)))
+		}
+		mp4.LastKeyframeGapMs = gap
+	}
+	mp4.LastKeyframeRawPTS = pts
+
+	if seam {
+		// Discard the truncated tail GOP; this keyframe is a clean restart point.
+		mp4.gopBuffer = mp4.gopBuffer[:0]
+	} else {
+		// Genuine GOP boundary: commit the GOP we just finished buffering.
+		mp4.commitBufferedGOP()
+	}
+
+	// Begin buffering the new GOP, starting with this keyframe.
+	mp4.gopBuffer = append(mp4.gopBuffer, bufferedSample{
+		trackID:           trackID,
+		isKeyframe:        isKeyframe,
+		data:              data,
+		pts:               pts,
+		compositionOffset: compositionOffset,
+	})
+	return nil
+}
+
+// commitBufferedGOP writes every sample currently held in gopBuffer to the file
+// in arrival order, then clears the buffer. Committing in arrival order
+// preserves the original audio/video interleave and lets commitSampleToTrack's
+// pending-sample mechanism derive each sample's duration from the next one, so
+// the on-disk result matches a straight pass-through.
+func (mp4 *MP4) commitBufferedGOP() {
+	if len(mp4.gopBuffer) == 0 {
+		return
+	}
+	buffered := mp4.gopBuffer
+	mp4.gopBuffer = nil // detach so commitSampleToTrack never observes a half-cleared buffer
+	for _, s := range buffered {
+		if err := mp4.commitSampleToTrack(s.trackID, s.isKeyframe, s.data, s.pts, s.compositionOffset); err != nil {
+			log.Log.Error("mp4.commitBufferedGOP(): " + err.Error())
+		}
+	}
+}
+
+// commitSampleToTrack appends a single buffered sample to the current fragment.
+// It is the low-level writer behind AddSampleToTrack and is only ever invoked
+// from commitBufferedGOP, after a GOP has been confirmed as non-seam.
+func (mp4 *MP4) commitSampleToTrack(trackID uint32, isKeyframe bool, data []byte, pts uint64, compositionOffset int64) error {
 
 	if isKeyframe && trackID == uint32(mp4.VideoTrack) {
 		mp4.TotalKeyframesReceived++
@@ -308,27 +416,6 @@ func (mp4 *MP4) AddSampleToTrack(trackID uint32, isKeyframe bool, data []byte, p
 		}
 		shouldFlush := !mp4.Start || elapsed >= FragmentDurationMs
 
-		// Detect upstream source-loop / restart discontinuity. When an MP4 is
-		// looped through virtual-rtsp (ffmpeg `-stream_loop -1 -re`) the loop
-		// seam emits a fresh IDR much sooner than a normal GOP would. PTS keeps
-		// growing monotonically, so the timing-only `elapsed` check above does
-		// not catch it and the seam IDR ends up as a mid-fragment sync sample.
-		// MSE-based players (Video.js / Chromium / Firefox) reject the resulting
-		// fragment with a "media corruption" error because the inner IDR resets
-		// frame_num/POC inside what they expect to be a single GOP. Force a
-		// fragment boundary whenever two consecutive keyframes arrive much
-		// closer than a normal GOP (here: < 500 ms apart). This isolates the
-		// seam IDR into its own fragment so each fragment stays a clean GOP.
-		if !shouldFlush && trackID == uint32(mp4.VideoTrack) && mp4.Start &&
-			mp4.LastKeyframeRawPTS > 0 && pts > mp4.LastKeyframeRawPTS &&
-			pts-mp4.LastKeyframeRawPTS < 500 {
-			log.Log.Warning(fmt.Sprintf("mp4.AddSampleToTrack(): forcing fragment flush at unexpectedly close keyframe (gap=%d ms, fragment elapsed=%d ms) - likely upstream loop/restart discontinuity", pts-mp4.LastKeyframeRawPTS, elapsed))
-			shouldFlush = true
-		}
-		if trackID == uint32(mp4.VideoTrack) {
-			mp4.LastKeyframeRawPTS = pts
-		}
-
 		if shouldFlush {
 			// Write the previous segment to the file
 			if mp4.Start {
@@ -415,7 +502,7 @@ func (mp4 *MP4) AddSampleToTrack(trackID uint32, isKeyframe bool, data []byte, p
 				fullSample.Sample = mp4ff.Sample{
 					Size:                  uint32(len(fullSample.Data)),
 					Flags:                 flags,
-					CompositionTimeOffset: 0, // No composition time offset for video
+					CompositionTimeOffset: int32(compositionOffset), // PTS-DTS, non-zero for B-frames
 				}
 				mp4.VideoFullSample = &fullSample
 				mp4.PendingSampleIsKeyframe = isKeyframe
@@ -433,14 +520,6 @@ func (mp4 *MP4) AddSampleToTrack(trackID uint32, isKeyframe bool, data []byte, p
 					if started {
 						dts = 1
 					}
-					// Guard against forward PTS jumps (e.g. virtual-rtsp loop
-					// boundary or upstream stalls). Without this clamp the
-					// audio trun would carry an enormous sample duration that
-					// renders the recording unplayable in browsers.
-					if mp4.LastAudioSampleDTS > 0 && dts > mp4.LastAudioSampleDTS*10 {
-						log.Log.Warning(fmt.Sprintf("mp4.AddSampleToTrack(): audio PTS jumped forward (pts=%d, prevDTS=%d, gap=%d) - clamping to last known duration", pts, mp4.AudioFullSample.DecodeTime, dts))
-						dts = mp4.LastAudioSampleDTS
-					}
 					mp4.LastAudioSampleDTS = dts
 					//fmt.Printf("Adding sample to track %d, PTS: %d, Duration: %d, size: %d\n", trackID, pts, dts, len(aac[7:]))
 					mp4.AudioTotalDuration += dts
@@ -476,6 +555,10 @@ func (mp4 *MP4) AddSampleToTrack(trackID uint32, isKeyframe bool, data []byte, p
 
 func (mp4 *MP4) Close(config *models.Config) {
 
+	// Commit the final buffered GOP held back for seam detection. The last GOP of
+	// a recording is never a loop seam, so it must always be written out.
+	mp4.commitBufferedGOP()
+
 	log.Log.Info(fmt.Sprintf("mp4.Close(): KEYFRAME SUMMARY - totalReceived=%d, totalWritten=%d, segments=%d, lastFragmentKF=%d",
 		mp4.TotalKeyframesReceived, mp4.TotalKeyframesWritten, mp4.SegmentCount, mp4.FragmentKeyframeCount))
 
@@ -610,6 +693,13 @@ func (mp4 *MP4) Close(config *models.Config) {
 		includePS := true
 		spsNALUs, ppsNALUs := normalizeH264ParameterSets(mp4.SPSNALUs, mp4.PPSNALUs)
 		log.Log.Debug("mp4.Close(): AVC parameter sets: SPS=" + formatNaluDebug(spsNALUs) + ", PPS=" + formatNaluDebug(ppsNALUs))
+		if len(spsNALUs) == 0 || len(ppsNALUs) == 0 {
+			// An avcC without both SPS and PPS is invalid: downstream FFmpeg-based
+			// pipelines decoding this file will report "non-existing PPS 0 referenced"
+			// and fail to extract any frame. Surface it loudly so the capture-side
+			// parameter-set handling can be diagnosed.
+			log.Log.Error(fmt.Sprintf("mp4.Close(): incomplete H264 parameter sets (SPS=%d, PPS=%d) - the avcC will be invalid and downstream decoders will report 'non-existing PPS 0 referenced'", len(spsNALUs), len(ppsNALUs)))
+		}
 		err := init.Moov.Traks[0].SetAVCDescriptor("avc1", spsNALUs, ppsNALUs, includePS)
 		if err != nil {
 			log.Log.Error("mp4.Close(): error setting AVC descriptor: " + err.Error())
@@ -636,6 +726,11 @@ func (mp4 *MP4) Close(config *models.Config) {
 		includePS := true
 		vpsNALUs, spsNALUs, ppsNALUs := normalizeH265ParameterSets(mp4.VPSNALUs, mp4.SPSNALUs, mp4.PPSNALUs)
 		log.Log.Debug("mp4.Close(): HEVC parameter sets: VPS=" + formatNaluDebug(vpsNALUs) + ", SPS=" + formatNaluDebug(spsNALUs) + ", PPS=" + formatNaluDebug(ppsNALUs))
+		if len(vpsNALUs) == 0 || len(spsNALUs) == 0 || len(ppsNALUs) == 0 {
+			// An hvcC missing VPS/SPS/PPS is invalid and downstream FFmpeg-based
+			// pipelines will fail to decode the recording. Surface it loudly.
+			log.Log.Error(fmt.Sprintf("mp4.Close(): incomplete H265 parameter sets (VPS=%d, SPS=%d, PPS=%d) - the hvcC will be invalid and downstream decoders will fail to process the recording", len(vpsNALUs), len(spsNALUs), len(ppsNALUs)))
+		}
 		err := init.Moov.Traks[0].SetHEVCDescriptor("hvc1", vpsNALUs, spsNALUs, ppsNALUs, [][]byte{}, includePS)
 		if err != nil {
 			log.Log.Error("mp4.Close(): error setting HEVC descriptor: " + err.Error())

machinery/src/video/mp4_duration_test.go

@@ -49,7 +49,7 @@ func TestMP4Duration(t *testing.T) {
 	for i := 0; i < numFrames; i++ {
 		pts := uint64(i) * frameDuration
 		isKeyframe := i%gopSize == 0
-		err := mp4Video.AddSampleToTrack(videoTrack, isKeyframe, makeFrame(isKeyframe), pts)
+		err := mp4Video.AddSampleToTrack(videoTrack, isKeyframe, makeFrame(isKeyframe), pts, 0)
 		if err != nil {
 			t.Fatalf("AddSampleToTrack failed at frame %d: %v", i, err)
 		}

machinery/src/video/mp4_loopseam_test.go

@@ -0,0 +1,194 @@
+package video
+
+import (
+	"os"
+	"testing"
+
+	mp4ff "github.com/Eyevinn/mp4ff/mp4"
+	"github.com/kerberos-io/agent/machinery/src/models"
+)
+
+// runLoopSeamScenario builds a fragmented MP4 that reproduces the loop-seam
+// pattern observed in the failing virtual-rtsp recordings (e.g.
+// thales_1781196512_3-138_2top_0-0-0-0_-1_30219.mp4): a steady GOP cadence, but
+// at the source-MP4 loop boundary the source restarts and emits a fresh IDR far
+// sooner than a normal GOP. In the real recordings the short tail GOP left just
+// before that premature IDR contains a truncated inter-frame - software decoders
+// conceal the missing macroblocks, but hardware decoders (macOS VideoToolbox,
+// kVTVideoDecoderBadDataErr / -12909) and MSE players reject it and freeze
+// playback at the seam (~10s in the original file).
+//
+// The fix detects the premature seam IDR and drops the truncated tail GOP that
+// precedes it. The seam IDR is itself a clean random-access point, so playback
+// resumes seamlessly. This scenario asserts that the tail GOP is removed -
+// exactly one GOP fewer than emitted - while every healthy GOP is preserved in
+// full and no two IDRs are left bunched in a fragment.
+//
+// gopFrames is the number of frames per GOP, so the same scenario can be
+// exercised at different (configurable) camera GOP sizes. The fix derives its
+// threshold from the observed keyframe cadence, so the truncated tail GOP is
+// dropped regardless of GOP size.
+func runLoopSeamScenario(t *testing.T, gopFrames int) {
+	t.Helper()
+
+	tmpFile, err := os.CreateTemp("", "test_loop_seam_*.mp4")
+	if err != nil {
+		t.Fatalf("create temp: %v", err)
+	}
+	tmpFile.Close()
+	defer os.Remove(tmpFile.Name())
+
+	sps := []byte{0x67, 0x42, 0xc0, 0x1e, 0xd9, 0x00, 0xa0, 0x47, 0xfe, 0xc8}
+	pps := []byte{0x68, 0xce, 0x38, 0x80}
+	mp4Video := NewMP4(tmpFile.Name(), [][]byte{sps}, [][]byte{pps}, nil, 60)
+	mp4Video.SetWidth(1920)
+	mp4Video.SetHeight(1080)
+	v := mp4Video.AddVideoTrack("H264")
+
+	mk := func(k bool) []byte {
+		nt := byte(0x01)
+		if k {
+			nt = 0x65
+		}
+		f := []byte{0, 0, 0, 1, nt}
+		for i := 0; i < 200; i++ {
+			f = append(f, byte(i))
+		}
+		return f
+	}
+
+	frameDur := uint64(33)
+	normalGOPms := uint64(gopFrames) * frameDur
+	pts := uint64(0)
+	emitFrame := func(isKey bool) {
+		// compositionOffset is 0: synthetic stream has no B-frames.
+		mp4Video.AddSampleToTrack(v, isKey, mk(isKey), pts, 0)
+		pts += frameDur
+	}
+	emitP := func(n int) {
+		for i := 0; i < n; i++ {
+			emitFrame(false)
+		}
+	}
+	// emitGOP emits one GOP: a leading keyframe followed by gopFrames-1 P-frames.
+	emitGOP := func() {
+		emitFrame(true)
+		emitP(gopFrames - 1)
+	}
+
+	// Several healthy GOPs to establish the cadence and fill a couple of
+	// fragments, then the truncated tail GOP: a keyframe followed by only a few
+	// P-frames before the source loops. This is the GOP that must be dropped.
+	for g := 0; g < 9; g++ {
+		emitGOP()
+	}
+	emitFrame(true)
+	seamLead := gopFrames / 5 // tail GOP is only ~20% of a normal GOP before the loop
+	if seamLead < 1 {
+		seamLead = 1
+	}
+	emitP(seamLead)
+	// Loop seam: the source recording restarts, emitting a fresh IDR far sooner
+	// than the normal GOP. The short tail GOP emitted just above is the truncated
+	// one that must be dropped; this seam IDR opens a fresh, healthy GOP.
+	emitFrame(true)
+	emitP(gopFrames - 1)
+	// The recording continues with normal GOPs to the end.
+	for g := 0; g < 10; g++ {
+		emitGOP()
+	}
+
+	mp4Video.Close(&models.Config{Signing: &models.Signing{PrivateKey: ""}})
+
+	f, err := os.Open(tmpFile.Name())
+	if err != nil {
+		t.Fatalf("open: %v", err)
+	}
+	defer f.Close()
+	parsed, err := mp4ff.DecodeFile(f)
+	if err != nil {
+		t.Fatalf("decode: %v", err)
+	}
+
+	// After the fix, the truncated tail GOP that precedes the premature seam IDR
+	// is dropped entirely (its first inter-frame is the incomplete one that
+	// freezes hardware decoders), while every other GOP is preserved in full.
+	//
+	// 9 lead GOPs + the seam's own (healthy) GOP + 10 trailing GOPs = 20 committed
+	// GOPs. The standalone "tail" keyframe and its seamLead P-frames are the
+	// dropped truncated GOP, so the output must contain exactly one GOP fewer than
+	// emitted and a whole number of complete GOPs.
+	const committedGOPs = 9 + 1 + 10
+	wantSync := committedGOPs
+	wantSamples := committedGOPs * gopFrames
+
+	// A healthy fragment only ever contains keyframes spaced ~normalGOPms apart.
+	// If any fragment contains two keyframes closer than half a normal GOP, the
+	// premature seam IDR was not dropped and the file will freeze on playback.
+	maxBunchMs := normalGOPms / 2
+
+	totalSamples := 0
+	totalSync := 0
+	fragIdx := 0
+	for _, seg := range parsed.Segments {
+		for _, fr := range seg.Fragments {
+			for _, traf := range fr.Moof.Trafs {
+				if traf.Tfhd.TrackID != 1 {
+					continue
+				}
+				tfdt := traf.Tfdt.BaseMediaDecodeTime()
+				offset := uint64(0)
+				var keys []uint64
+				for _, trun := range traf.Truns {
+					for _, s := range trun.Samples {
+						totalSamples++
+						// sample_depends_on == 2 => "does not depend on others" => IDR/sync.
+						if (s.Flags>>24)&0x03 == 0x02 {
+							keys = append(keys, offset)
+							totalSync++
+						}
+						offset += uint64(s.Dur)
+					}
+				}
+				t.Logf("gop=%dframes frag %d tfdt=%d samples_dur=%d keys@%v", gopFrames, fragIdx, tfdt, offset, keys)
+				for i := 1; i < len(keys); i++ {
+					gap := keys[i] - keys[i-1]
+					if gap < maxBunchMs {
+						t.Errorf("gop=%dframes frag %d (tfdt=%d): two IDRs only %d ms apart in same fragment (< %d) - seam was not dropped",
+							gopFrames, fragIdx, tfdt, gap, maxBunchMs)
+					}
+				}
+				fragIdx++
+			}
+		}
+	}
+
+	if totalSync != wantSync {
+		t.Errorf("gop=%dframes: got %d keyframes in output, want %d - the truncated seam GOP was not dropped exactly once",
+			gopFrames, totalSync, wantSync)
+	}
+	if totalSamples != wantSamples {
+		t.Errorf("gop=%dframes: got %d video samples in output, want %d (= %d committed GOPs x %d frames) - the seam GOP drop removed the wrong frames",
+			gopFrames, totalSamples, wantSamples, committedGOPs, gopFrames)
+	}
+}
+
+// TestMP4LoopSeamDrop exercises the ~1s GOP case (30 frames @ ~33ms),
+// matching the original failing recording.
+func TestMP4LoopSeamDrop(t *testing.T) {
+	runLoopSeamScenario(t, 30)
+}
+
+// TestMP4LoopSeamDropLargeGOP exercises a larger ~2s GOP (60 frames). The
+// GOP size is configurable per camera; this guards against regressing to a
+// fixed-millisecond threshold that would only work for ~1s GOPs.
+func TestMP4LoopSeamDropLargeGOP(t *testing.T) {
+	runLoopSeamScenario(t, 60)
+}
+
+// TestMP4LoopSeamDropShortGOP exercises a short ~0.5s GOP (15 frames),
+// where a fixed ~1s threshold would misfire on every keyframe. The relative
+// detection must only drop the genuine premature seam's truncated tail GOP.
+func TestMP4LoopSeamDropShortGOP(t *testing.T) {
+	runLoopSeamScenario(t, 15)
+}

machinery/src/webrtc/main.go

@@ -26,7 +26,12 @@ import (
 
 const (
 	// Channel buffer sizes
-	candidateChannelBuffer = 100
+	// candidateChannelBuffer: large enough to absorb the burst of trickled ICE
+	// candidates that can arrive over MQTT before the SetRemoteDescription
+	// goroutine starts draining them. A small buffer caused candidates to be
+	// dropped silently on restrictive networks, leaving ICE stuck in
+	// "checking" until the viewer refreshed.
+	candidateChannelBuffer = 512
 	rtcpBufferSize         = 1500
 
 	// Timeouts and intervals
@@ -116,6 +121,22 @@ func (cm *ConnectionManager) RemovePeerConnection(sessionKey string) {
 	}
 }
 
+// CloseExistingPeerConnection closes and removes any peer connection currently
+// registered under sessionKey. Returns true if one was found. This is used to
+// reset state cleanly when a new request-hd-stream arrives for a session id
+// that the agent thinks is still active (for example after a viewer reload
+// where the previous PC hasn't yet been timed out by ICE).
+func (cm *ConnectionManager) CloseExistingPeerConnection(sessionKey string) bool {
+	cm.mu.RLock()
+	wrapper, exists := cm.peerConnections[sessionKey]
+	cm.mu.RUnlock()
+	if !exists || wrapper == nil {
+		return false
+	}
+	cleanupPeerConnection(sessionKey, wrapper)
+	return true
+}
+
 // QueueCandidate safely queues a candidate for a session without racing with channel closure.
 func (cm *ConnectionManager) QueueCandidate(sessionKey string, candidate string) bool {
 	cm.mu.Lock()
@@ -341,6 +362,17 @@ func InitializeWebRTCConnection(configuration *models.Configuration, communicati
 
 	// We create a channel which will hold the candidates for this session.
 	sessionKey := config.Key + "/" + handshakePayload.SessionID
+
+	// If a previous peer connection for this exact session is still hanging
+	// around (e.g. a viewer reloaded before pion's ICE timeout fired) close it
+	// first so we start from a clean slate. Without this, the new request would
+	// race against a stale PC that still owns the per-peer broadcaster tracks.
+	if globalConnectionManager.CloseExistingPeerConnection(sessionKey) {
+		log.Log.Info("webrtc.main.InitializeWebRTCConnection(): closed stale peer connection for session " + handshakePayload.SessionID)
+	}
+	// Drain/reset the candidate channel too \u2014 leftover candidates from the
+	// prior session are not valid for the new ICE agent.
+	globalConnectionManager.CloseCandidateChannel(sessionKey)
 	candidateChannel := globalConnectionManager.GetOrCreateCandidateChannel(sessionKey)
 
 	// Set variables