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 && \

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 && \

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
@@ -550,6 +684,12 @@ func compositionOffsetMs(ext dtsExtractor, au [][]byte, pts int64, clockRate int
 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) {
@@ -813,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.
@@ -983,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/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.")

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
@@ -74,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.
@@ -275,7 +299,101 @@ func (mp4 *MP4) flushPendingVideoSample(nextPTS uint64) bool {
 // 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++
@@ -437,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))
 
@@ -571,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())
@@ -597,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_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)
+}