Merge pull request #288 from kerberos-io/feature/resumable-uploads-to-hub

feature/resumable-uploads-to-hub

.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"
+
+  

.gitignore

@@ -14,5 +14,7 @@ machinery/test*
 machinery/init-dev.sh
 machinery/.env.local
 machinery/vendor
+machinery/go.work
+machinery/go.work.sum
 deployments/docker/private-docker-compose.yaml
 video.mp4

.vscode/launch.json

@@ -18,6 +18,9 @@
             ],
             "envFile": "${workspaceFolder}/machinery/.env.local",
             "buildFlags": "--tags dynamic",
+            "env": {
+                "GOWORK": "off"
+            },
         },
         {
             "name": "Launch React",

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
 

machinery/.env

@@ -27,5 +27,12 @@ AGENT_KERBEROSVAULT_SECONDARY_DIRECTORY=
 AGENT_KERBEROSVAULT_SECONDARY_ACCESS_KEY=
 AGENT_KERBEROSVAULT_SECONDARY_SECRET_KEY=
 
+# Resumable (tus) uploads to Kerberos Vault are enabled by default.
+# Set to true to fall back to the legacy single-shot POST /storage upload.
+#AGENT_DISABLE_RESUMABLE_UPLOAD=true
+# Bytes sent per PATCH request (default 1 MiB = 1048576). 0 disables chunking
+# and sends the whole file in a single PATCH.
+AGENT_TUS_CHUNK_SIZE_BYTES=1048576
+
 # Open telemetry tracing endpoint
 OTEL_EXPORTER_OTLP_ENDPOINT=

machinery/cmd/mp4analyze/main.go

@@ -0,0 +1,425 @@
+package main
+
+import (
+	"fmt"
+	"os"
+
+	"github.com/Eyevinn/mp4ff/avc"
+	mp4ff "github.com/Eyevinn/mp4ff/mp4"
+)
+
+func main() {
+	if len(os.Args) < 2 {
+		fmt.Println("usage: mp4analyze <file.mp4>")
+		os.Exit(1)
+	}
+	f, err := os.Open(os.Args[1])
+	if err != nil {
+		panic(err)
+	}
+	defer f.Close()
+	parsed, err := mp4ff.DecodeFile(f)
+	if err != nil {
+		panic(err)
+	}
+
+	// Movie-level info
+	if parsed.Init != nil && parsed.Init.Moov != nil {
+		moov := parsed.Init.Moov
+		fmt.Printf("ftyp/moov present. timescale(mvhd)=%d duration(mvhd)=%d\n",
+			moov.Mvhd.Timescale, moov.Mvhd.Duration)
+		for _, trak := range moov.Traks {
+			ts := trak.Mdia.Mdhd.Timescale
+			fmt.Printf("  trak id=%d handler=%s mdhd.timescale=%d mdhd.duration=%d\n",
+				trak.Tkhd.TrackID, trak.Mdia.Hdlr.HandlerType, ts, trak.Mdia.Mdhd.Duration)
+		}
+	} else {
+		fmt.Println("no Init/Moov (pure fragmented stream?)")
+	}
+
+	// sidx vs actual segment layout. MSE players use sidx to map presentation
+	// time -> byte ranges; if sidx references disagree with the real segment
+	// sizes/durations (e.g. after an early/short flush) the player fetches the
+	// wrong bytes and fails to decode — a failure that "heals" on seek.
+	fmt.Println("=== sidx references vs actual segments ===")
+	var sidxRefs []mp4ff.SidxRef
+	for _, c := range parsed.Children {
+		if s, ok := c.(*mp4ff.SidxBox); ok {
+			fmt.Printf("  sidx: timescale=%d earliestPresTime=%d firstOffset=%d refCount=%d anchor(after sidx)=%d\n",
+				s.Timescale, s.EarliestPresentationTime, s.FirstOffset, len(s.SidxRefs), s.AnchorPoint)
+			sidxRefs = s.SidxRefs
+		}
+	}
+	// Actual segment sizes (styp+moof+mdat) and fragment durations.
+	type segInfo struct {
+		size uint64
+		dur  uint64
+	}
+	var actual []segInfo
+	for _, seg := range parsed.Segments {
+		var sz uint64
+		if seg.Styp != nil {
+			sz += seg.Styp.Size()
+		}
+		if seg.Sidx != nil {
+			sz += seg.Sidx.Size()
+		}
+		var dur uint64
+		for _, fr := range seg.Fragments {
+			sz += fr.Moof.Size()
+			if fr.Mdat != nil {
+				sz += fr.Mdat.Size()
+			}
+			for _, traf := range fr.Moof.Trafs {
+				if traf.Tfhd.TrackID != 1 {
+					continue
+				}
+				for _, trun := range traf.Truns {
+					for _, s := range trun.Samples {
+						dur += uint64(s.Dur)
+					}
+				}
+			}
+		}
+		actual = append(actual, segInfo{size: sz, dur: dur})
+	}
+	for i := range actual {
+		refStr := "(no sidx ref)"
+		if i < len(sidxRefs) {
+			r := sidxRefs[i]
+			mark := ""
+			if uint64(r.ReferencedSize) != actual[i].size {
+				mark += fmt.Sprintf(" SIZE MISMATCH actual=%d", actual[i].size)
+			}
+			if uint64(r.SubSegmentDuration) != actual[i].dur {
+				mark += fmt.Sprintf(" DUR MISMATCH actual=%d", actual[i].dur)
+			}
+			refStr = fmt.Sprintf("sidx.size=%d sidx.dur=%d type=%d sap=%d/%d%s",
+				r.ReferencedSize, r.SubSegmentDuration, r.ReferenceType, r.StartsWithSAP, r.SAPType, mark)
+		}
+		fmt.Printf("  seg%02d actual.size=%d actual.dur=%d | %s\n", i, actual[i].size, actual[i].dur, refStr)
+	}
+
+	fmt.Println("=== fragments ===")
+	fragIdx := 0
+	var allKeyGlobal []uint64 // global keyframe decode times (track timescale units)
+	var prevTfdtEnd = map[uint32]uint64{}
+	for si, seg := range parsed.Segments {
+		for _, fr := range seg.Fragments {
+			for _, traf := range fr.Moof.Trafs {
+				tid := traf.Tfhd.TrackID
+				tfdt := traf.Tfdt.BaseMediaDecodeTime()
+				offset := uint64(0)
+				var keys []uint64    // keyframe offset-from-tfdt
+				var durs []uint64
+				zeroDur := 0
+				nSamples := 0
+				for _, trun := range traf.Truns {
+					for _, s := range trun.Samples {
+						nSamples++
+						if (s.Flags>>24)&0x03 == 0x02 { // sample_depends_on==2 => IDR/sync
+							keys = append(keys, offset)
+							if tid == 1 {
+								allKeyGlobal = append(allKeyGlobal, tfdt+offset)
+							}
+						}
+						if s.Dur == 0 {
+							zeroDur++
+						}
+						durs = append(durs, uint64(s.Dur))
+						offset += uint64(s.Dur)
+					}
+				}
+				cont := ""
+				if pe, ok := prevTfdtEnd[tid]; ok {
+					if tfdt != pe {
+						cont = fmt.Sprintf(" <-- tfdt GAP/JUMP prev_end=%d delta=%d", pe, int64(tfdt)-int64(pe))
+					}
+				}
+				prevTfdtEnd[tid] = tfdt + offset
+				if tid == 1 {
+					// in-fragment keyframe gaps
+					var gaps []int64
+					for i := 1; i < len(keys); i++ {
+						gaps = append(gaps, int64(keys[i])-int64(keys[i-1]))
+					}
+					fmt.Printf("seg%d frag%d trk%d tfdt=%d dur=%d nSamp=%d zeroDur=%d keys=%v inFragKeyGaps=%v%s\n",
+						si, fragIdx, tid, tfdt, offset, nSamples, zeroDur, keys, gaps, cont)
+				}
+			}
+			fragIdx++
+		}
+	}
+
+	fmt.Println("=== global video keyframe decode times & gaps ===")
+	for i, k := range allKeyGlobal {
+		gap := int64(0)
+		if i > 0 {
+			gap = int64(k) - int64(allKeyGlobal[i-1])
+		}
+		flag := ""
+		if i > 1 {
+			prevGap := int64(allKeyGlobal[i-1]) - int64(allKeyGlobal[i-2])
+			if gap > 0 && prevGap > 0 && gap*2 < prevGap {
+				flag = fmt.Sprintf("  <== SEAM? gap=%d < prevGap/2=%d", gap, prevGap/2)
+			}
+		}
+		fmt.Printf("  kf#%02d dt=%d gap=%d%s\n", i, k, gap, flag)
+	}
+
+	// Full sample timeline: DTS, CTS (=DTS+cto), composition offset, NAL types,
+	// to detect PTS non-monotonicity / gaps / param-set changes at the seam.
+	fmt.Println("=== per-sample timeline (full) — checking PTS monotonicity & nal types ===")
+	var trex *mp4ff.TrexBox
+	if parsed.Init != nil && parsed.Init.Moov != nil && parsed.Init.Moov.Mvex != nil {
+		for _, t := range parsed.Init.Moov.Mvex.Trexs {
+			if t.TrackID == 1 {
+				trex = t
+			}
+		}
+	}
+	var lastCTS int64 = -1
+	var lastDTS int64 = -1
+	sampIdx := 0
+	fragIdx = 0
+	for _, seg := range parsed.Segments {
+		for _, fr := range seg.Fragments {
+			fs, err := fr.GetFullSamples(trex)
+			if err != nil {
+				fmt.Printf("  frag%d GetFullSamples err: %v\n", fragIdx, err)
+				fragIdx++
+				continue
+			}
+			for _, s := range fs {
+				dts := int64(s.DecodeTime)
+				cts := dts + int64(s.CompositionTimeOffset)
+				nals := nalTypes(s.Data)
+				anomaly := ""
+				if lastCTS >= 0 && cts < lastCTS {
+					anomaly += fmt.Sprintf(" <== CTS BACKWARDS (prev=%d)", lastCTS)
+				}
+				if lastDTS >= 0 && dts < lastDTS {
+					anomaly += fmt.Sprintf(" <== DTS BACKWARDS (prev=%d)", lastDTS)
+				}
+				isSync := s.Flags&0x02000000 == 0 && (s.Flags>>24)&0x03 == 0x02
+				// Only print near the seam region and any anomalies, to keep output small.
+				near := dts >= 7800 && dts <= 8700
+				if near || anomaly != "" {
+					fmt.Printf("  s%04d frag%d dts=%d cts=%d cto=%d dur=%d size=%d sync=%v nal=%v%s\n",
+						sampIdx, fragIdx, dts, cts, s.CompositionTimeOffset, s.Dur, len(s.Data), isSync, nals, anomaly)
+				}
+				lastCTS = cts
+				lastDTS = dts
+				sampIdx++
+			}
+			fragIdx++
+		}
+	}
+
+	// Compare parameter sets: avcC (in moov) vs inline SPS/PPS at every IDR.
+	// A looping source that restarts may re-emit SPS/PPS that differ from the
+	// ones the player configured its decoder with from avcC — a classic cause
+	// of a freeze that "heals" when you seek past the seam.
+	fmt.Println("=== parameter set comparison (avcC vs inline IDR) ===")
+	var avccSPS, avccPPS [][]byte
+	if parsed.Init != nil && parsed.Init.Moov != nil {
+		for _, trak := range parsed.Init.Moov.Traks {
+			if trak.Mdia == nil || trak.Mdia.Minf == nil || trak.Mdia.Minf.Stbl == nil {
+				continue
+			}
+			stsd := trak.Mdia.Minf.Stbl.Stsd
+			if stsd == nil || stsd.AvcX == nil || stsd.AvcX.AvcC == nil {
+				continue
+			}
+			avccSPS = stsd.AvcX.AvcC.SPSnalus
+			avccPPS = stsd.AvcX.AvcC.PPSnalus
+		}
+	}
+	for i, s := range avccSPS {
+		fmt.Printf("  avcC SPS[%d] = %x\n", i, s)
+	}
+	for i, p := range avccPPS {
+		fmt.Printf("  avcC PPS[%d] = %x\n", i, p)
+	}
+	fragIdx = 0
+	sampIdx = 0
+	var baseSPS, basePPS []byte
+	if len(avccSPS) > 0 {
+		baseSPS = avccSPS[0]
+	}
+	if len(avccPPS) > 0 {
+		basePPS = avccPPS[0]
+	}
+	for _, seg := range parsed.Segments {
+		for _, fr := range seg.Fragments {
+			fs, err := fr.GetFullSamples(trex)
+			if err != nil {
+				fragIdx++
+				continue
+			}
+			for _, s := range fs {
+				spsList := nalsByType(s.Data, 7)
+				ppsList := nalsByType(s.Data, 8)
+				if len(spsList) > 0 || len(ppsList) > 0 {
+					dts := int64(s.DecodeTime)
+					note := ""
+					if len(spsList) > 0 {
+						if baseSPS == nil {
+							baseSPS = spsList[0]
+						} else if !bytesEqual(baseSPS, spsList[0]) {
+							note += " <== SPS CHANGED vs base/avcC"
+						}
+					}
+					if len(ppsList) > 0 {
+						if basePPS == nil {
+							basePPS = ppsList[0]
+						} else if !bytesEqual(basePPS, ppsList[0]) {
+							note += " <== PPS CHANGED vs base/avcC"
+						}
+					}
+					var spsHex, ppsHex string
+					if len(spsList) > 0 {
+						spsHex = fmt.Sprintf("%x", spsList[0])
+					}
+					if len(ppsList) > 0 {
+						ppsHex = fmt.Sprintf("%x", ppsList[0])
+					}
+					fmt.Printf("  IDR s%04d frag%d dts=%d SPS=%s PPS=%s%s\n",
+						sampIdx, fragIdx, dts, spsHex, ppsHex, note)
+				}
+				sampIdx++
+			}
+			fragIdx++
+		}
+	}
+
+	sliceHeaders(parsed, trex)
+}
+
+func sliceHeaders(parsed *mp4ff.File, trex *mp4ff.TrexBox) {
+	// Build SPS/PPS maps from avcC.
+	spsMap := map[uint32]*avc.SPS{}
+	ppsMap := map[uint32]*avc.PPS{}
+	if parsed.Init != nil && parsed.Init.Moov != nil {
+		for _, trak := range parsed.Init.Moov.Traks {
+			if trak.Mdia == nil || trak.Mdia.Minf == nil || trak.Mdia.Minf.Stbl == nil {
+				continue
+			}
+			stsd := trak.Mdia.Minf.Stbl.Stsd
+			if stsd == nil || stsd.AvcX == nil || stsd.AvcX.AvcC == nil {
+				continue
+			}
+			for _, s := range stsd.AvcX.AvcC.SPSnalus {
+				if sps, err := avc.ParseSPSNALUnit(s, true); err == nil {
+					spsMap[uint32(sps.ParameterID)] = sps
+				}
+			}
+			for _, p := range stsd.AvcX.AvcC.PPSnalus {
+				if pps, err := avc.ParsePPSNALUnit(p, spsMap); err == nil {
+					ppsMap[pps.PicParameterSetID] = pps
+				}
+			}
+		}
+	}
+
+	fmt.Println("=== slice headers near seam (frame_num / poc / idr_pic_id) ===")
+	fragIdx := 0
+	sampIdx := 0
+	for _, seg := range parsed.Segments {
+		for _, fr := range seg.Fragments {
+			fs, err := fr.GetFullSamples(trex)
+			if err != nil {
+				fragIdx++
+				continue
+			}
+			for _, s := range fs {
+				dts := int64(s.DecodeTime)
+				if dts < 6800 || dts > 9400 {
+					sampIdx++
+					continue
+				}
+				for _, nal := range splitAVCC(s.Data) {
+					t := nal[0] & 0x1f
+					if t == 1 || t == 5 { // non-IDR or IDR slice
+						sh, err := avc.ParseSliceHeader(nal, spsMap, ppsMap)
+						if err != nil {
+							fmt.Printf("  s%04d frag%d dts=%d nalType=%d sliceHeader ERR: %v\n", sampIdx, fragIdx, dts, t, err)
+							break
+						}
+						fmt.Printf("  s%04d frag%d dts=%d nalType=%d sliceType=%v frameNum=%d idrPicId=%d pocLsb=%d\n",
+							sampIdx, fragIdx, dts, t, sh.SliceType, sh.FrameNum, sh.IDRPicID, sh.PicOrderCntLsb)
+						break
+					}
+				}
+				sampIdx++
+			}
+			fragIdx++
+		}
+	}
+}
+
+// splitAVCC splits a length-prefixed (4-byte) AVCC buffer into NAL units.
+func splitAVCC(b []byte) [][]byte {
+	var out [][]byte
+	i := 0
+	for i+4 <= len(b) {
+		n := int(uint32(b[i])<<24 | uint32(b[i+1])<<16 | uint32(b[i+2])<<8 | uint32(b[i+3]))
+		i += 4
+		if n <= 0 || i+n > len(b) {
+			break
+		}
+		out = append(out, b[i:i+n])
+		i += n
+	}
+	return out
+}
+
+func bytesEqual(a, b []byte) bool {
+	if len(a) != len(b) {
+		return false
+	}
+	for i := range a {
+		if a[i] != b[i] {
+			return false
+		}
+	}
+	return true
+}
+
+// nalTypes returns the list of H.264 NAL unit types present in an AVCC
+// (length-prefixed) sample buffer.
+func nalTypes(b []byte) []int {
+	var out []int
+	i := 0
+	for i+4 <= len(b) {
+		n := int(uint32(b[i])<<24 | uint32(b[i+1])<<16 | uint32(b[i+2])<<8 | uint32(b[i+3]))
+		i += 4
+		if n <= 0 || i+n > len(b) {
+			break
+		}
+		out = append(out, int(b[i]&0x1f))
+		i += n
+	}
+	return out
+}
+
+// nalsByType returns the raw NAL payloads (without length prefix) of the given
+// type from an AVCC (length-prefixed) sample buffer.
+func nalsByType(b []byte, want int) [][]byte {
+	var out [][]byte
+	i := 0
+	for i+4 <= len(b) {
+		n := int(uint32(b[i])<<24 | uint32(b[i+1])<<16 | uint32(b[i+2])<<8 | uint32(b[i+3]))
+		i += 4
+		if n <= 0 || i+n > len(b) {
+			break
+		}
+		if int(b[i]&0x1f) == want {
+			nal := make([]byte, n)
+			copy(nal, b[i:i+n])
+			out = append(out, nal)
+		}
+		i += n
+	}
+	return out
+}

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/dropbox.go

@@ -46,7 +46,11 @@ func UploadDropbox(configuration *models.Configuration, fileName string) (bool,
 
 	file, err := os.OpenFile(fullname, os.O_RDWR, 0755)
 	if file != nil {
-		defer file.Close()
+		defer func() {
+			if cerr := file.Close(); cerr != nil {
+				log.Log.Error("UploadDropbox: Error closing file: " + cerr.Error())
+			}
+		}()
 	}
 
 	if err == nil {

machinery/src/cloud/kerberos_hub.go

@@ -34,6 +34,29 @@ func UploadKerberosHub(configuration *models.Configuration, fileName string) (bo
 
 	log.Log.Info("UploadKerberosHub: Uploading to Kerberos Hub (" + config.HubURI + ")")
 	log.Log.Info("UploadKerberosHub: Upload started for " + fileName)
+
+	// Prefer the resumable (tus) upload when enabled (the default). Kerberos Hub
+	// authenticates the agent with its Hub public/private key and proxies the
+	// resumable upload to the Kerberos Vault. When Hub does not expose a tus
+	// endpoint (older deployments) we transparently fall back to the legacy
+	// single-POST upload below.
+	if resumableUploadsEnabled() {
+		uploaded, _, supported, body, rerr := uploadHubResumable(&config, fileName, "UploadKerberosHub", "hub")
+		if supported {
+			if uploaded {
+				log.Log.Info("UploadKerberosHub: Upload Finished (resumable), " + body)
+				return true, true, nil
+			}
+			if rerr != nil {
+				log.Log.Info("UploadKerberosHub: resumable upload failed, " + rerr.Error())
+			} else {
+				log.Log.Info("UploadKerberosHub: resumable upload incomplete, " + body)
+			}
+			return false, true, rerr
+		}
+		log.Log.Info("UploadKerberosHub: resumable (tus) endpoint not available, falling back to legacy upload")
+	}
+
 	fullname := "data/recordings/" + fileName
 
 	// Check if we still have the file otherwise we abort the request.
@@ -68,9 +91,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 +152,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/cloud/kerberos_vault.go

@@ -30,6 +30,15 @@ func UploadKerberosVault(configuration *models.Configuration, fileName string) (
 		return false, false, errors.New(err)
 	}
 
+	// If the recording no longer exists on disk there is nothing to upload.
+	// This can happen when the file was already removed (e.g. cleanup, or an
+	// earlier successful upload). Skip it so the watcher drops the marker
+	// instead of retrying indefinitely.
+	if _, err := os.Stat("data/recordings/" + fileName); err != nil {
+		log.Log.Info("UploadKerberosVault: skipping " + fileName + ", file doesn't exist anymore")
+		return false, false, nil
+	}
+
 	// timestamp_microseconds_instanceName_regionCoordinates_numberOfChanges_token
 	// 1564859471_6-474162_oprit_577-283-727-375_1153_27.mp4
 	// - Timestamp
@@ -41,17 +50,6 @@ func UploadKerberosVault(configuration *models.Configuration, fileName string) (
 	// KerberosCloud, this means storage is disabled and proxy enabled.
 	log.Log.Info("UploadKerberosVault: Uploading to Kerberos Vault (" + config.KStorage.URI + ")")
 	log.Log.Info("UploadKerberosVault: Upload started for " + fileName)
-	fullname := "data/recordings/" + fileName
-
-	file, err := os.OpenFile(fullname, os.O_RDWR, 0755)
-	if file != nil {
-		defer file.Close()
-	}
-	if err != nil {
-		err := "UploadKerberosVault: Upload Failed, file doesn't exists anymore"
-		log.Log.Info(err)
-		return false, false, errors.New(err)
-	}
 
 	publicKey := config.KStorage.CloudKey
 	if config.HubKey != "" {
@@ -60,62 +58,30 @@ func UploadKerberosVault(configuration *models.Configuration, fileName string) (
 
 	// We need to check if we are in a retry timeout.
 	if kstorageRetryTimeout <= time.Now().Unix() {
+		uploaded, responded, body, err := sendToVault(*config.KStorage, publicKey, config.Key, fileName, "UploadKerberosVault", "primary")
+		if uploaded {
+			kstorageRetryCount = 0
+			log.Log.Info("UploadKerberosVault: Upload Finished, " + body)
+			return true, true, nil
+		}
 
-		req, err := http.NewRequest("POST", config.KStorage.URI+"/storage", file)
 		if err != nil {
-			errorMessage := "UploadKerberosVault: error reading request, " + config.KStorage.URI + "/storage: " + err.Error()
-			log.Log.Error(errorMessage)
-			return false, true, errors.New(errorMessage)
-		}
-		req.Header.Set("Content-Type", "video/mp4")
-		req.Header.Set("X-Kerberos-Storage-CloudKey", publicKey)
-		req.Header.Set("X-Kerberos-Storage-AccessKey", config.KStorage.AccessKey)
-		req.Header.Set("X-Kerberos-Storage-SecretAccessKey", config.KStorage.SecretAccessKey)
-		req.Header.Set("X-Kerberos-Storage-Provider", config.KStorage.Provider)
-		req.Header.Set("X-Kerberos-Storage-FileName", fileName)
-		req.Header.Set("X-Kerberos-Storage-Device", config.Key)
-		req.Header.Set("X-Kerberos-Storage-Capture", "IPCamera")
-		req.Header.Set("X-Kerberos-Storage-Directory", config.KStorage.Directory)
-
-		var client *http.Client
-		if os.Getenv("AGENT_TLS_INSECURE") == "true" {
-			tr := &http.Transport{
-				TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
-			}
-			client = &http.Client{Transport: tr}
-		} else {
-			client = &http.Client{}
-		}
-
-		resp, err := client.Do(req)
-		if resp != nil {
-			defer resp.Body.Close()
-		}
-
-		if err == nil {
-			if resp != nil {
-				body, err := io.ReadAll(resp.Body)
-				if err == nil {
-					if resp.StatusCode == 200 {
-						kstorageRetryCount = 0
-						log.Log.Info("UploadKerberosVault: Upload Finished, " + resp.Status + ", " + string(body))
-						return true, true, nil
-					} else {
-						// We increase the retry count, and set the timeout.
-						// If we have reached the retry policy, we set the timeout.
-						// This means we will not retry for the next 5 minutes.
-						if kstorageRetryCount < config.KStorage.MaxRetries {
-							kstorageRetryCount = (kstorageRetryCount + 1)
-						}
-						if kstorageRetryCount == config.KStorage.MaxRetries {
-							kstorageRetryTimeout = time.Now().Add(time.Duration(config.KStorage.Timeout) * time.Second).Unix()
-						}
-						log.Log.Info("UploadKerberosVault: Upload Failed, " + resp.Status + ", " + string(body))
-					}
-				}
-			}
-		} else {
 			log.Log.Info("UploadKerberosVault: Upload Failed, " + err.Error())
+		} else {
+			log.Log.Info("UploadKerberosVault: Upload Failed, " + body)
+		}
+
+		// We only advance the retry policy when the vault gave a definitive
+		// response (mirroring the original behaviour where transient network
+		// errors did not consume retries). When the retry count reaches the
+		// configured maximum we back off for the configured timeout.
+		if responded {
+			if kstorageRetryCount < config.KStorage.MaxRetries {
+				kstorageRetryCount = (kstorageRetryCount + 1)
+			}
+			if kstorageRetryCount == config.KStorage.MaxRetries {
+				kstorageRetryTimeout = time.Now().Add(time.Duration(config.KStorage.Timeout) * time.Second).Unix()
+			}
 		}
 	}
 
@@ -134,61 +100,116 @@ func UploadKerberosVault(configuration *models.Configuration, fileName string) (
 
 		log.Log.Info("UploadKerberosVault (Secondary): Uploading to Secondary Kerberos Vault (" + config.KStorageSecondary.URI + ")")
 
-		file, err = os.OpenFile(fullname, os.O_RDWR, 0755)
-		if file != nil {
-			defer file.Close()
-		}
-		if err != nil {
-			err := "UploadKerberosVault (Secondary): Upload Failed, file doesn't exists anymore"
-			log.Log.Info(err)
-			return false, false, errors.New(err)
+		uploaded, _, body, err := sendToVault(*config.KStorageSecondary, publicKey, config.Key, fileName, "UploadKerberosVault (Secondary)", "secondary")
+		if uploaded {
+			log.Log.Info("UploadKerberosVault (Secondary): Upload Finished to secondary, " + body)
+			return true, true, nil
 		}
 
-		req, err := http.NewRequest("POST", config.KStorageSecondary.URI+"/storage", file)
 		if err != nil {
-			errorMessage := "UploadKerberosVault (Secondary): error reading request, " + config.KStorageSecondary.URI + "/storage: " + err.Error()
-			log.Log.Error(errorMessage)
-			return false, true, errors.New(errorMessage)
-		}
-		req.Header.Set("Content-Type", "video/mp4")
-		req.Header.Set("X-Kerberos-Storage-CloudKey", publicKey)
-		req.Header.Set("X-Kerberos-Storage-AccessKey", config.KStorageSecondary.AccessKey)
-		req.Header.Set("X-Kerberos-Storage-SecretAccessKey", config.KStorageSecondary.SecretAccessKey)
-		req.Header.Set("X-Kerberos-Storage-Provider", config.KStorageSecondary.Provider)
-		req.Header.Set("X-Kerberos-Storage-FileName", fileName)
-		req.Header.Set("X-Kerberos-Storage-Device", config.Key)
-		req.Header.Set("X-Kerberos-Storage-Capture", "IPCamera")
-		req.Header.Set("X-Kerberos-Storage-Directory", config.KStorageSecondary.Directory)
-
-		var client *http.Client
-		if os.Getenv("AGENT_TLS_INSECURE") == "true" {
-			tr := &http.Transport{
-				TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
-			}
-			client = &http.Client{Transport: tr}
+			log.Log.Info("UploadKerberosVault (Secondary): Upload Failed to secondary, " + err.Error())
 		} else {
-			client = &http.Client{}
-		}
-
-		resp, err := client.Do(req)
-		if resp != nil {
-			defer resp.Body.Close()
-		}
-
-		if err == nil {
-			if resp != nil {
-				body, err := io.ReadAll(resp.Body)
-				if err == nil {
-					if resp.StatusCode == 200 {
-						log.Log.Info("UploadKerberosVault (Secondary): Upload Finished to secondary, " + resp.Status + ", " + string(body))
-						return true, true, nil
-					} else {
-						log.Log.Info("UploadKerberosVault (Secondary): Upload Failed to secondary, " + resp.Status + ", " + string(body))
-					}
-				}
-			}
+			log.Log.Info("UploadKerberosVault (Secondary): Upload Failed to secondary, " + body)
 		}
 	}
 
 	return false, true, nil
 }
+
+// sendToVault uploads a single recording to one Kerberos Vault. When resumable
+// uploads are enabled (the default) it attempts the tus protocol first and, if
+// the vault does not expose a tus endpoint (older deployments), transparently
+// falls back to the legacy single-shot POST.
+//
+// It returns whether the upload succeeded, whether the vault gave a definitive
+// HTTP response (so the caller can advance its retry policy), a short message
+// for logging, and a transport error if any.
+func sendToVault(vault models.KStorage, publicKey, deviceKey, fileName, label, slot string) (bool, bool, string, error) {
+	if resumableUploadsEnabled() {
+		uploaded, responded, supported, body, err := uploadVaultResumable(vault, publicKey, deviceKey, fileName, label, slot)
+		if supported {
+			return uploaded, responded, body, err
+		}
+		log.Log.Info(label + ": resumable (tus) endpoint not available, falling back to legacy upload")
+	}
+	return uploadVaultLegacy(vault, publicKey, deviceKey, fileName, label)
+}
+
+// uploadVaultLegacy performs the original single-request upload: the whole file
+// is sent as the body of a POST to {URI}/storage. Kept for backwards
+// compatibility with vault deployments that do not support resumable uploads.
+func uploadVaultLegacy(vault models.KStorage, publicKey, deviceKey, fileName, label string) (bool, bool, string, error) {
+	fullname := "data/recordings/" + fileName
+
+	file, err := os.Open(fullname)
+	if file != nil {
+		defer file.Close()
+	}
+	if err != nil {
+		msg := label + ": Upload Failed, file doesn't exists anymore"
+		log.Log.Info(msg)
+		return false, false, "", errors.New(msg)
+	}
+
+	uri := vault.URI
+	for len(uri) > 0 && uri[len(uri)-1] == '/' {
+		uri = uri[:len(uri)-1]
+	}
+
+	req, err := http.NewRequest("POST", uri+"/storage", file)
+	if err != nil {
+		errorMessage := label + ": error reading request, " + uri + "/storage: " + err.Error()
+		log.Log.Error(errorMessage)
+		return false, false, "", errors.New(errorMessage)
+	}
+	req.Header.Set("Content-Type", "video/mp4")
+	setVaultHeaders(req.Header, vault, publicKey, deviceKey, fileName)
+
+	client := newVaultHTTPClient(0)
+	resp, err := client.Do(req)
+	if resp != nil {
+		defer resp.Body.Close()
+	}
+	if err != nil {
+		return false, false, "", err
+	}
+
+	body, rerr := io.ReadAll(resp.Body)
+	if rerr != nil {
+		return false, false, "", rerr
+	}
+
+	if resp.StatusCode == 200 {
+		return true, true, resp.Status + ", " + string(body), nil
+	}
+	return false, true, resp.Status + ", " + string(body), nil
+}
+
+// setVaultHeaders sets the standard Kerberos Vault headers used by the legacy
+// single-POST upload.
+func setVaultHeaders(h http.Header, vault models.KStorage, publicKey, deviceKey, fileName string) {
+	h.Set("X-Kerberos-Storage-CloudKey", publicKey)
+	h.Set("X-Kerberos-Storage-AccessKey", vault.AccessKey)
+	h.Set("X-Kerberos-Storage-SecretAccessKey", vault.SecretAccessKey)
+	h.Set("X-Kerberos-Storage-Provider", vault.Provider)
+	h.Set("X-Kerberos-Storage-FileName", fileName)
+	h.Set("X-Kerberos-Storage-Device", deviceKey)
+	h.Set("X-Kerberos-Storage-Capture", "IPCamera")
+	h.Set("X-Kerberos-Storage-Directory", vault.Directory)
+}
+
+// newVaultHTTPClient builds an HTTP client honouring the AGENT_TLS_INSECURE
+// escape hatch. A timeout of 0 disables the client-level timeout, which is
+// required for streaming large upload bodies.
+func newVaultHTTPClient(timeout time.Duration) *http.Client {
+	client := &http.Client{}
+	if os.Getenv("AGENT_TLS_INSECURE") == "true" {
+		client.Transport = &http.Transport{
+			TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
+		}
+	}
+	if timeout > 0 {
+		client.Timeout = timeout
+	}
+	return client
+}

machinery/src/cloud/tus_client.go

@@ -0,0 +1,546 @@
+package cloud
+
+import (
+	"encoding/base64"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"net/http"
+	"net/url"
+	"os"
+	"path/filepath"
+	"sort"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/kerberos-io/agent/machinery/src/log"
+	"github.com/kerberos-io/agent/machinery/src/models"
+)
+
+// tusResumableVersion is the tus protocol version implemented by this client.
+const tusResumableVersion = "1.0.0"
+
+// tusUploadPath is appended to the configured Kerberos Vault URI to reach the
+// resumable upload endpoint. It mirrors how the legacy uploader appends
+// "/storage".
+const tusUploadPath = "/storage/tus/"
+
+// tusResumeState is persisted in a sidecar file next to the agent data so an
+// interrupted upload can be resumed across retries and even agent restarts.
+type tusResumeState struct {
+	UploadURL string `json:"upload_url"`
+	VaultURI  string `json:"vault_uri"`
+	Size      int64  `json:"size"`
+}
+
+// resumableUploadsEnabled reports whether the resumable (tus) upload path should
+// be attempted. It is enabled by default and can be disabled (falling back to
+// the legacy single POST) by setting AGENT_DISABLE_RESUMABLE_UPLOAD=true.
+func resumableUploadsEnabled() bool {
+	return os.Getenv("AGENT_DISABLE_RESUMABLE_UPLOAD") != "true"
+}
+
+// tusDefaultChunkSize is the number of bytes uploaded per PATCH request when no
+// explicit size is configured. Splitting the upload into chunks keeps each HTTP
+// request small enough for intermediary proxies/load balancers and checkpoints
+// progress frequently, so an interruption resumes with minimal re-upload.
+const tusDefaultChunkSize int64 = 1 << 20 // 1 MiB
+
+const tusProgressBucketPercent int64 = 10
+
+// tusChunkSize returns the number of bytes to send per PATCH request. It
+// defaults to tusDefaultChunkSize (1 MiB) and can be overridden with the
+// AGENT_TUS_CHUNK_SIZE_BYTES environment variable. A value of 0 (or negative)
+// disables chunking and sends the remaining bytes in a single PATCH.
+func tusChunkSize() int64 {
+	v := os.Getenv("AGENT_TUS_CHUNK_SIZE_BYTES")
+	if v == "" {
+		return tusDefaultChunkSize
+	}
+	n, err := strconv.ParseInt(v, 10, 64)
+	if err != nil {
+		return tusDefaultChunkSize
+	}
+	if n <= 0 {
+		return 0 // chunking disabled: send everything in one PATCH
+	}
+	return n
+}
+
+func tusProgressBucket(offset, size int64) int64 {
+	if size <= 0 {
+		return 100
+	}
+	percent := (offset * 100) / size
+	if percent > 100 {
+		percent = 100
+	}
+	return percent / tusProgressBucketPercent
+}
+
+func logTusUploadProgress(label string, offset, size int64, loggedBucket *int64) {
+	bucket := tusProgressBucket(offset, size)
+	if bucket <= *loggedBucket {
+		return
+	}
+	*loggedBucket = bucket
+	percent := bucket * tusProgressBucketPercent
+	if percent > 100 {
+		percent = 100
+	}
+	log.Log.Infof("%s: resumable upload progress %d%% (%d/%d bytes)", label, percent, offset, size)
+}
+
+// tusHeaderFunc sets the authentication and routing headers required on every
+// tus request for a particular upload target (Kerberos Vault directly, or
+// Kerberos Hub which proxies to a vault). fileName is only meaningful on the
+// creation request; it is empty on HEAD/PATCH/DELETE.
+type tusHeaderFunc func(h http.Header, fileName string)
+
+// runTusUpload performs a resumable (tus) upload of data/recordings/<fileName>
+// to baseURL, sending target-specific authentication/routing headers via
+// setHeaders on every request. It encapsulates the create/resume/chunk/finalize
+// state machine shared by the Kerberos Vault (direct) and Kerberos Hub (proxied)
+// upload paths.
+//
+// Return values:
+//   - uploaded:  the recording was fully received and persisted by the server.
+//   - responded: the server returned a definitive HTTP response (used by the
+//     caller to advance its retry/secondary-failover policy).
+//   - supported: the server exposes a tus endpoint. When false, the caller
+//     should fall back to the legacy single-POST upload (older deployments).
+//   - body:      a short message for logging.
+func runTusUpload(baseURL, metadata, fileName, label, slot string, setHeaders tusHeaderFunc) (uploaded bool, responded bool, supported bool, body string, err error) {
+	fullname := "data/recordings/" + fileName
+
+	file, ferr := os.Open(fullname)
+	if file != nil {
+		defer file.Close()
+	}
+	if ferr != nil {
+		msg := label + ": resumable upload failed, file doesn't exist anymore"
+		log.Log.Info(msg)
+		// The file is gone, so the legacy path cannot help either. Report it as
+		// "supported" to avoid a pointless fallback attempt.
+		return false, false, true, "", errors.New(msg)
+	}
+
+	info, serr := file.Stat()
+	if serr != nil {
+		return false, false, true, "", serr
+	}
+	size := info.Size()
+
+	client := newVaultHTTPClient(0)
+	client.CheckRedirect = func(req *http.Request, via []*http.Request) error {
+		if len(via) == 0 {
+			return nil
+		}
+		if req.URL.Host != via[0].URL.Host {
+			for k := range req.Header {
+				if strings.HasPrefix(http.CanonicalHeaderKey(k), "X-Kerberos-") {
+					req.Header.Del(k)
+				}
+			}
+		}
+		return nil
+	}
+
+	sidecar := tusSidecarPath(fileName, slot)
+	uploadURL := loadTusResumeState(sidecar, baseURL)
+
+	const maxAttempts = 4
+	restartedAfterComplete := false
+
+	for attempt := 0; attempt < maxAttempts; attempt++ {
+		// (1) Ensure we have an active upload URL, creating one if needed.
+		if uploadURL == "" {
+			created, status, cerr := tusCreate(client, baseURL, size, metadata, setHeaders, fileName)
+			if cerr != nil {
+				if status == http.StatusNotFound || status == http.StatusMethodNotAllowed || status == http.StatusNotImplemented {
+					// The vault does not implement tus; let the caller fall back.
+					return false, false, false, "", cerr
+				}
+				log.Log.Info(label + ": resumable create failed, " + cerr.Error())
+				tusBackoff(attempt)
+				continue
+			}
+			uploadURL = created
+			saveTusResumeState(sidecar, tusResumeState{UploadURL: uploadURL, VaultURI: baseURL, Size: size})
+		}
+
+		// (2) Query the current server-side offset.
+		offset, status, herr := tusHead(client, uploadURL, setHeaders)
+		if herr != nil {
+			if status == http.StatusNotFound || status == http.StatusGone {
+				// The upload expired/was removed server-side; start over.
+				removeTusResumeState(sidecar)
+				uploadURL = ""
+				continue
+			}
+			log.Log.Info(label + ": resumable head failed, " + herr.Error())
+			tusBackoff(attempt)
+			continue
+		}
+
+		// (3) All bytes are present but the upload was not finalized (e.g. the
+		// completion hook failed). A completed tus upload cannot be re-finalized
+		// with another PATCH, so delete it and re-upload to force a clean finalize.
+		if offset >= size {
+			if restartedAfterComplete {
+				return false, true, true, "resumable finalize did not complete", errors.New(label + ": resumable finalize did not complete")
+			}
+			tusTerminate(client, uploadURL, setHeaders)
+			removeTusResumeState(sidecar)
+			uploadURL = ""
+			restartedAfterComplete = true
+			continue
+		}
+
+		// (4) Stream the remaining bytes to the vault via PATCH, reading directly
+		// from disk so the recording is never fully buffered in memory. When a chunk
+		// size is configured the data is sent across several PATCH requests,
+		// checkpointing the offset after each one so an interruption resumes from the
+		// last completed chunk instead of re-uploading everything.
+		chunkSize := tusChunkSize()
+		progressed := false
+		patchFailed := false
+		var lastBody string
+		loggedProgressBucket := tusProgressBucket(offset, size)
+		for offset < size {
+			// Re-seek every chunk so the on-disk position always matches the
+			// server-acknowledged offset, even if a PATCH was partially accepted.
+			if _, sErr := file.Seek(offset, io.SeekStart); sErr != nil {
+				return false, false, true, "", sErr
+			}
+			patchLen := size - offset
+			if chunkSize > 0 && chunkSize < patchLen {
+				patchLen = chunkSize
+			}
+			newOffset, status, respBody, perr := tusPatch(client, uploadURL, offset, patchLen, file, setHeaders)
+			if perr != nil {
+				if status >= 400 {
+					// Definitive rejection (e.g. provider push failed during finalize).
+					// Re-evaluate via HEAD on the next iteration to decide retry/restart.
+					log.Log.Info(label + ": resumable patch rejected, " + perr.Error())
+				} else {
+					log.Log.Info(label + ": resumable patch failed, " + perr.Error())
+				}
+				tusBackoff(attempt)
+				patchFailed = true
+				break
+			}
+			if newOffset > offset {
+				progressed = true
+			}
+			offset = newOffset
+			lastBody = respBody
+			logTusUploadProgress(label, offset, size, &loggedProgressBucket)
+			if offset < size {
+				// Partial progress: persist so a later retry resumes from here.
+				saveTusResumeState(sidecar, tusResumeState{UploadURL: uploadURL, VaultURI: baseURL, Size: size})
+			}
+		}
+		if patchFailed {
+			if progressed {
+				// Forward progress refreshes the retry budget: maxAttempts bounds the
+				// number of consecutive failures, not the number of chunks needed for
+				// a large recording.
+				attempt = -1
+			}
+			continue
+		}
+
+		// All declared bytes have been sent and acknowledged: the upload is done.
+		removeTusResumeState(sidecar)
+		return true, true, true, lastBody, nil
+	}
+
+	return false, true, true, "resumable upload did not complete after retries", errors.New(label + ": resumable upload did not complete after retries")
+}
+
+// uploadVaultResumable uploads a recording directly to a Kerberos Vault using
+// the tus resumable upload protocol. Credentials travel in the
+// X-Kerberos-Storage-* headers on every request and routing (directory/provider)
+// is additionally carried in the tus Upload-Metadata.
+func uploadVaultResumable(vault models.KStorage, publicKey, deviceKey, fileName, label, slot string) (bool, bool, bool, string, error) {
+	baseURL := strings.TrimRight(vault.URI, "/") + tusUploadPath
+	metadata := encodeTusMetadata(map[string]string{
+		"filename":  fileName,
+		"device":    deviceKey,
+		"directory": vault.Directory,
+		"provider":  vault.Provider,
+		"capture":   "IPCamera",
+		"cloudkey":  publicKey,
+	})
+	setHeaders := func(h http.Header, fn string) {
+		setVaultTusHeaders(h, vault, publicKey, deviceKey, fn)
+	}
+	return runTusUpload(baseURL, metadata, fileName, label, slot, setHeaders)
+}
+
+// uploadHubResumable uploads a recording to Kerberos Hub's tus endpoint, which
+// authenticates the agent with its Hub public/private key and proxies the
+// resumable upload to the Kerberos Vault on the agent's behalf. The vault
+// directory and provider are resolved and injected by Kerberos Hub, so they are
+// intentionally omitted from the metadata here.
+func uploadHubResumable(config *models.Config, fileName, label, slot string) (bool, bool, bool, string, error) {
+	baseURL := strings.TrimRight(config.HubURI, "/") + tusUploadPath
+	metadata := encodeTusMetadata(map[string]string{
+		"filename": fileName,
+		"device":   config.Key,
+		"capture":  "IPCamera",
+	})
+	setHeaders := func(h http.Header, fn string) {
+		setHubTusHeaders(h, config, fn)
+	}
+	return runTusUpload(baseURL, metadata, fileName, label, slot, setHeaders)
+}
+
+// tusCreate performs the tus "creation" request (POST). On success it returns
+// the resolved upload URL the agent should use for subsequent HEAD/PATCH calls.
+func tusCreate(client *http.Client, baseURL string, size int64, metadata string, setHeaders tusHeaderFunc, fileName string) (string, int, error) {
+	req, err := http.NewRequest("POST", baseURL, nil)
+	if err != nil {
+		return "", 0, err
+	}
+	req.Header.Set("Tus-Resumable", tusResumableVersion)
+	req.Header.Set("Upload-Length", strconv.FormatInt(size, 10))
+	if metadata != "" {
+		req.Header.Set("Upload-Metadata", metadata)
+	}
+	setHeaders(req.Header, fileName)
+
+	resp, err := client.Do(req)
+	if resp != nil {
+		defer resp.Body.Close()
+	}
+	if err != nil {
+		return "", 0, err
+	}
+	io.Copy(io.Discard, resp.Body)
+
+	if resp.StatusCode != http.StatusCreated {
+		return "", resp.StatusCode, fmt.Errorf("unexpected status creating upload: %s", resp.Status)
+	}
+	location := resp.Header.Get("Location")
+	if location == "" {
+		return "", resp.StatusCode, errors.New("missing Location header in create response")
+	}
+	return resolveTusLocation(baseURL, location), resp.StatusCode, nil
+}
+
+// tusHead performs the tus "offset" request (HEAD) and returns the current
+// server-side upload offset.
+func tusHead(client *http.Client, uploadURL string, setHeaders tusHeaderFunc) (int64, int, error) {
+	req, err := http.NewRequest("HEAD", uploadURL, nil)
+	if err != nil {
+		return 0, 0, err
+	}
+	req.Header.Set("Tus-Resumable", tusResumableVersion)
+	setHeaders(req.Header, "")
+
+	resp, err := client.Do(req)
+	if resp != nil {
+		defer resp.Body.Close()
+	}
+	if err != nil {
+		return 0, 0, err
+	}
+	io.Copy(io.Discard, resp.Body)
+
+	if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusNoContent {
+		return 0, resp.StatusCode, fmt.Errorf("unexpected status on HEAD: %s", resp.Status)
+	}
+	offsetStr := resp.Header.Get("Upload-Offset")
+	offset, perr := strconv.ParseInt(offsetStr, 10, 64)
+	if perr != nil {
+		return 0, resp.StatusCode, fmt.Errorf("invalid Upload-Offset header: %q", offsetStr)
+	}
+	return offset, resp.StatusCode, nil
+}
+
+// tusPatch streams up to length bytes of the file (starting at offset) to the
+// upload URL using a single PATCH request. The body is read straight from the
+// *os.File, so the recording is never fully buffered in memory.
+func tusPatch(client *http.Client, uploadURL string, offset, length int64, file io.Reader, setHeaders tusHeaderFunc) (int64, int, string, error) {
+	req, err := http.NewRequest("PATCH", uploadURL, io.LimitReader(file, length))
+	if err != nil {
+		return offset, 0, "", err
+	}
+	req.ContentLength = length
+	req.Header.Set("Tus-Resumable", tusResumableVersion)
+	req.Header.Set("Content-Type", "application/offset+octet-stream")
+	req.Header.Set("Upload-Offset", strconv.FormatInt(offset, 10))
+	setHeaders(req.Header, "")
+
+	resp, err := client.Do(req)
+	if resp != nil {
+		defer resp.Body.Close()
+	}
+	if err != nil {
+		return offset, 0, "", err
+	}
+	bodyBytes, _ := io.ReadAll(resp.Body)
+	respBody := string(bodyBytes)
+
+	if resp.StatusCode != http.StatusNoContent {
+		return offset, resp.StatusCode, respBody, fmt.Errorf("unexpected status on PATCH: %s, %s", resp.Status, respBody)
+	}
+	newOffsetStr := resp.Header.Get("Upload-Offset")
+	newOffset, perr := strconv.ParseInt(newOffsetStr, 10, 64)
+	if perr != nil {
+		// A 204 without a parseable offset means this PATCH was fully accepted.
+		return offset + length, resp.StatusCode, respBody, nil
+	}
+	return newOffset, resp.StatusCode, respBody, nil
+}
+
+// tusTerminate best-effort deletes an upload server-side (DELETE).
+func tusTerminate(client *http.Client, uploadURL string, setHeaders tusHeaderFunc) {
+	req, err := http.NewRequest("DELETE", uploadURL, nil)
+	if err != nil {
+		return
+	}
+	req.Header.Set("Tus-Resumable", tusResumableVersion)
+	setHeaders(req.Header, "")
+
+	resp, derr := client.Do(req)
+	if resp != nil {
+		io.Copy(io.Discard, resp.Body)
+		resp.Body.Close()
+	}
+	_ = derr
+}
+
+// setVaultTusHeaders sets the Kerberos Vault authentication and routing headers
+// on every tus request. Credentials are sent on each request (and never stored
+// server-side in the upload metadata). When fileName is empty it is omitted, as
+// it is only useful on the creation request (routing also travels in the tus
+// Upload-Metadata).
+func setVaultTusHeaders(h http.Header, vault models.KStorage, publicKey, deviceKey, fileName string) {
+	h.Set("X-Kerberos-Storage-CloudKey", publicKey)
+	h.Set("X-Kerberos-Storage-AccessKey", vault.AccessKey)
+	h.Set("X-Kerberos-Storage-SecretAccessKey", vault.SecretAccessKey)
+	h.Set("X-Kerberos-Storage-Provider", vault.Provider)
+	h.Set("X-Kerberos-Storage-Device", deviceKey)
+	h.Set("X-Kerberos-Storage-Directory", vault.Directory)
+	h.Set("X-Kerberos-Storage-Capture", "IPCamera")
+	if fileName != "" {
+		h.Set("X-Kerberos-Storage-FileName", fileName)
+	}
+}
+
+// setHubTusHeaders sets the Kerberos Hub authentication headers on every tus
+// request of a hub-proxied resumable upload. The agent authenticates with its
+// Hub public/private key (exactly as the legacy single-POST hub upload does);
+// Kerberos Hub validates the subscription and injects the vault credentials and
+// directory/provider on the agent's behalf.
+func setHubTusHeaders(h http.Header, config *models.Config, fileName string) {
+	h.Set("X-Kerberos-Hub-PublicKey", config.HubKey)
+	h.Set("X-Kerberos-Hub-PrivateKey", config.HubPrivateKey)
+	h.Set("X-Kerberos-Hub-Region", config.S3.Region)
+	h.Set("X-Kerberos-Storage-Device", config.Key)
+	h.Set("X-Kerberos-Storage-Capture", "IPCamera")
+	if fileName != "" {
+		h.Set("X-Kerberos-Storage-FileName", fileName)
+	}
+}
+
+// encodeTusMetadata serializes a map into the tus Upload-Metadata header format:
+// a comma separated list of "key base64(value)" pairs. Keys are sorted for a
+// deterministic header value. Empty values are skipped.
+func encodeTusMetadata(pairs map[string]string) string {
+	parts := make([]string, 0, len(pairs))
+	for k, v := range pairs {
+		if v == "" {
+			continue
+		}
+		parts = append(parts, k+" "+base64.StdEncoding.EncodeToString([]byte(v)))
+	}
+	sort.Strings(parts)
+	return strings.Join(parts, ",")
+}
+
+// resolveTusLocation turns the Location header returned by the create request
+// into an absolute URL. To keep talking to the agent's configured vault host
+// (and avoid issues when the vault sits behind a proxy that rewrites the host),
+// it keeps the configured base URL and only appends the server-assigned upload
+// id taken from the Location.
+func resolveTusLocation(baseURL, location string) string {
+	if ref, err := url.Parse(location); err == nil {
+		trimmed := strings.Trim(ref.Path, "/")
+		if trimmed != "" {
+			segments := strings.Split(trimmed, "/")
+			id := segments[len(segments)-1]
+			if id != "" {
+				return strings.TrimRight(baseURL, "/") + "/" + id
+			}
+		}
+	}
+	// Fallback: resolve the reference against the base URL as-is.
+	if base, err := url.Parse(baseURL); err == nil {
+		if ref, err := url.Parse(location); err == nil {
+			return base.ResolveReference(ref).String()
+		}
+	}
+	return location
+}
+
+// tusSidecarDir is the directory where resume state files are kept. It is
+// intentionally separate from data/cloud (which is scanned for recordings to
+// upload) so the sidecar files are never mistaken for recordings.
+func tusSidecarDir() string {
+	return "data/tus"
+}
+
+func tusSidecarPath(fileName, slot string) string {
+	safe := strings.ReplaceAll(fileName, "/", "_")
+	safe = strings.ReplaceAll(safe, string(os.PathSeparator), "_")
+	return filepath.Join(tusSidecarDir(), safe+"."+slot+".json")
+}
+
+// loadTusResumeState returns a previously stored upload URL for the given
+// sidecar, but only if it was created against the same vault base URL. Any
+// mismatch or read/parse error yields an empty string (start fresh).
+func loadTusResumeState(path, baseURL string) string {
+	b, err := os.ReadFile(path)
+	if err != nil {
+		return ""
+	}
+	var state tusResumeState
+	if err := json.Unmarshal(b, &state); err != nil {
+		return ""
+	}
+	if state.UploadURL == "" || state.VaultURI != baseURL {
+		return ""
+	}
+	return state.UploadURL
+}
+
+func saveTusResumeState(path string, state tusResumeState) {
+	if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
+		return
+	}
+	b, err := json.Marshal(state)
+	if err != nil {
+		return
+	}
+	_ = os.WriteFile(path, b, 0o644)
+}
+
+func removeTusResumeState(path string) {
+	_ = os.Remove(path)
+}
+
+// tusBackoff sleeps for an exponentially increasing duration (capped) between
+// resume attempts to avoid hammering a temporarily unavailable vault.
+func tusBackoff(attempt int) {
+	delay := time.Duration(500*(1<<uint(attempt))) * time.Millisecond
+	if delay > 3*time.Second {
+		delay = 3 * time.Second
+	}
+	time.Sleep(delay)
+}

machinery/src/cloud/tus_client_test.go

@@ -0,0 +1,608 @@
+package cloud
+
+import (
+	"bytes"
+	"encoding/base64"
+	"fmt"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"path/filepath"
+	"strconv"
+	"strings"
+	"sync"
+	"testing"
+
+	"github.com/kerberos-io/agent/machinery/src/models"
+)
+
+// fakeUpload tracks the state of a single resumable upload on the fake server.
+type fakeUpload struct {
+	size   int64
+	offset int64
+}
+
+// recordedRequest captures the method and headers of a request received by the
+// fake tus server, so tests can assert the client's per-method auth headers.
+type recordedRequest struct {
+	method string
+	header http.Header
+}
+
+// fakeTus is a tiny in-memory implementation of the tus 1.0.0 server protocol,
+// sufficient to exercise the agent's resumable client.
+type fakeTus struct {
+	mu             sync.Mutex
+	uploads        map[string]*fakeUpload
+	counter        int
+	creates        int
+	lastPatchBytes int64
+	patchSizes     []int64
+
+	// unsupported makes the creation endpoint return 404, simulating an older
+	// vault without a tus endpoint.
+	unsupported bool
+	// failFinalize causes the next N completing PATCH requests to return 502
+	// after storing the bytes, simulating a failed completion hook.
+	failFinalize int
+
+	// requests records the headers of every received request (in order) so
+	// tests can assert which auth/routing headers the client sent per method.
+	requests []recordedRequest
+}
+
+func newFakeTus() *fakeTus {
+	return &fakeTus{uploads: map[string]*fakeUpload{}}
+}
+
+func (s *fakeTus) seed(size, offset int64) string {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.counter++
+	id := fmt.Sprintf("seed-%d", s.counter)
+	s.uploads[id] = &fakeUpload{size: size, offset: offset}
+	return id
+}
+
+func (s *fakeTus) totalBytes() int64 {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	var total int64
+	for _, u := range s.uploads {
+		total += u.offset
+	}
+	return total
+}
+
+func (s *fakeTus) lastPatch() int64 {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.lastPatchBytes
+}
+
+// patchCounts returns the number of PATCH requests received and the size of each.
+func (s *fakeTus) patchCounts() (int, []int64) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	sizes := make([]int64, len(s.patchSizes))
+	copy(sizes, s.patchSizes)
+	return len(s.patchSizes), sizes
+}
+
+func (s *fakeTus) createCount() int {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.creates
+}
+
+// requestsForMethod returns the recorded requests for the given HTTP method.
+func (s *fakeTus) requestsForMethod(method string) []recordedRequest {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	var out []recordedRequest
+	for _, req := range s.requests {
+		if req.method == method {
+			out = append(out, req)
+		}
+	}
+	return out
+}
+
+func (s *fakeTus) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	id := strings.TrimPrefix(r.URL.Path, tusUploadPath)
+	w.Header().Set("Tus-Resumable", tusResumableVersion)
+
+	s.mu.Lock()
+	s.requests = append(s.requests, recordedRequest{method: r.Method, header: r.Header.Clone()})
+	s.mu.Unlock()
+
+	switch r.Method {
+	case http.MethodPost:
+		if s.unsupported {
+			w.WriteHeader(http.StatusNotFound)
+			return
+		}
+		length, _ := strconv.ParseInt(r.Header.Get("Upload-Length"), 10, 64)
+		s.mu.Lock()
+		s.counter++
+		s.creates++
+		newID := fmt.Sprintf("up-%d", s.counter)
+		s.uploads[newID] = &fakeUpload{size: length}
+		s.mu.Unlock()
+		w.Header().Set("Location", tusUploadPath+newID)
+		w.WriteHeader(http.StatusCreated)
+
+	case http.MethodHead:
+		s.mu.Lock()
+		u, ok := s.uploads[id]
+		s.mu.Unlock()
+		if !ok {
+			w.WriteHeader(http.StatusNotFound)
+			return
+		}
+		w.Header().Set("Upload-Offset", strconv.FormatInt(u.offset, 10))
+		w.Header().Set("Upload-Length", strconv.FormatInt(u.size, 10))
+		w.WriteHeader(http.StatusOK)
+
+	case http.MethodPatch:
+		s.mu.Lock()
+		u, ok := s.uploads[id]
+		s.mu.Unlock()
+		if !ok {
+			w.WriteHeader(http.StatusNotFound)
+			return
+		}
+		n, _ := io.Copy(io.Discard, r.Body)
+		s.mu.Lock()
+		u.offset += n
+		s.lastPatchBytes = n
+		s.patchSizes = append(s.patchSizes, n)
+		complete := u.offset >= u.size
+		failNow := complete && s.failFinalize > 0
+		if failNow {
+			s.failFinalize--
+		}
+		offset := u.offset
+		s.mu.Unlock()
+
+		w.Header().Set("Upload-Offset", strconv.FormatInt(offset, 10))
+		if failNow {
+			// Bytes are stored but the (simulated) completion hook failed.
+			w.WriteHeader(http.StatusBadGateway)
+			return
+		}
+		w.WriteHeader(http.StatusNoContent)
+
+	case http.MethodDelete:
+		s.mu.Lock()
+		delete(s.uploads, id)
+		s.mu.Unlock()
+		w.WriteHeader(http.StatusNoContent)
+
+	default:
+		w.WriteHeader(http.StatusMethodNotAllowed)
+	}
+}
+
+// withRecording switches into a fresh temp working directory containing a
+// recording at data/recordings/<fileName>. The working directory is restored on
+// cleanup. Tests using this helper must not run in parallel.
+func withRecording(t *testing.T, fileName string, payload []byte) {
+	t.Helper()
+	dir := t.TempDir()
+	old, err := os.Getwd()
+	if err != nil {
+		t.Fatalf("getwd: %v", err)
+	}
+	if err := os.Chdir(dir); err != nil {
+		t.Fatalf("chdir: %v", err)
+	}
+	t.Cleanup(func() { _ = os.Chdir(old) })
+
+	if err := os.MkdirAll("data/recordings", 0o755); err != nil {
+		t.Fatalf("mkdir recordings: %v", err)
+	}
+	if err := os.WriteFile(filepath.Join("data/recordings", fileName), payload, 0o644); err != nil {
+		t.Fatalf("write recording: %v", err)
+	}
+}
+
+func testVault(uri string) models.KStorage {
+	return models.KStorage{
+		URI:             uri,
+		AccessKey:       "ak",
+		SecretAccessKey: "sk",
+		Provider:        "gcp",
+		Directory:       "dir",
+	}
+}
+
+func TestUploadVaultResumable_HappyPath(t *testing.T) {
+	srv := newFakeTus()
+	ts := httptest.NewServer(srv)
+	defer ts.Close()
+
+	fileName := "1564859471_6-474162_oprit_577-283-727-375_1153_27.mp4"
+	payload := bytes.Repeat([]byte("x"), 4096)
+	withRecording(t, fileName, payload)
+
+	uploaded, responded, supported, _, err := uploadVaultResumable(testVault(ts.URL), "pk", "dev", fileName, "test", "primary")
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if !uploaded || !responded || !supported {
+		t.Fatalf("uploaded/responded/supported = %v/%v/%v, want all true", uploaded, responded, supported)
+	}
+	if got := srv.totalBytes(); got != int64(len(payload)) {
+		t.Fatalf("server received %d bytes, want %d", got, len(payload))
+	}
+	if _, err := os.Stat(tusSidecarPath(fileName, "primary")); !os.IsNotExist(err) {
+		t.Fatalf("expected sidecar to be removed after success, stat err = %v", err)
+	}
+}
+
+func TestUploadVaultResumable_Chunked(t *testing.T) {
+	srv := newFakeTus()
+	ts := httptest.NewServer(srv)
+	defer ts.Close()
+
+	fileName := "1564859471_6-474162_oprit_577-283-727-375_1153_27.mp4"
+	// 10 KiB payload uploaded in 4 KiB chunks => 3 PATCH requests (4096+4096+2048).
+	payload := bytes.Repeat([]byte("c"), 10240)
+	withRecording(t, fileName, payload)
+	t.Setenv("AGENT_TUS_CHUNK_SIZE_BYTES", "4096")
+
+	uploaded, _, supported, _, err := uploadVaultResumable(testVault(ts.URL), "pk", "dev", fileName, "test", "primary")
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if !uploaded || !supported {
+		t.Fatalf("expected chunked upload success, got uploaded=%v supported=%v", uploaded, supported)
+	}
+	if got := srv.totalBytes(); got != int64(len(payload)) {
+		t.Fatalf("server received %d bytes, want %d", got, len(payload))
+	}
+	count, sizes := srv.patchCounts()
+	if count != 3 {
+		t.Fatalf("expected 3 chunked PATCH requests, got %d (sizes=%v)", count, sizes)
+	}
+	want := []int64{4096, 4096, 2048}
+	for i, w := range want {
+		if sizes[i] != w {
+			t.Fatalf("chunk %d size = %d, want %d (sizes=%v)", i, sizes[i], w, sizes)
+		}
+	}
+	if _, err := os.Stat(tusSidecarPath(fileName, "primary")); !os.IsNotExist(err) {
+		t.Fatalf("expected sidecar removed after success, stat err = %v", err)
+	}
+}
+
+func TestUploadVaultResumable_ChunkingDisabled(t *testing.T) {
+	srv := newFakeTus()
+	ts := httptest.NewServer(srv)
+	defer ts.Close()
+
+	fileName := "1564859471_6-474162_oprit_577-283-727-375_1153_27.mp4"
+	payload := bytes.Repeat([]byte("d"), 10240)
+	withRecording(t, fileName, payload)
+	// 0 disables chunking: the whole file should go out in a single PATCH.
+	t.Setenv("AGENT_TUS_CHUNK_SIZE_BYTES", "0")
+
+	uploaded, _, supported, _, err := uploadVaultResumable(testVault(ts.URL), "pk", "dev", fileName, "test", "primary")
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if !uploaded || !supported {
+		t.Fatalf("expected success, got uploaded=%v supported=%v", uploaded, supported)
+	}
+	count, sizes := srv.patchCounts()
+	if count != 1 {
+		t.Fatalf("expected a single PATCH when chunking is disabled, got %d (sizes=%v)", count, sizes)
+	}
+	if sizes[0] != int64(len(payload)) {
+		t.Fatalf("single PATCH size = %d, want %d", sizes[0], len(payload))
+	}
+}
+
+func TestTusChunkSize(t *testing.T) {
+	cases := []struct {
+		name string
+		env  string
+		set  bool
+		want int64
+	}{
+		{name: "default when unset", set: false, want: tusDefaultChunkSize},
+		{name: "default on invalid", env: "notanumber", set: true, want: tusDefaultChunkSize},
+		{name: "explicit value", env: "65536", set: true, want: 65536},
+		{name: "zero disables", env: "0", set: true, want: 0},
+		{name: "negative disables", env: "-5", set: true, want: 0},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			if tc.set {
+				t.Setenv("AGENT_TUS_CHUNK_SIZE_BYTES", tc.env)
+			} else {
+				t.Setenv("AGENT_TUS_CHUNK_SIZE_BYTES", "")
+			}
+			if got := tusChunkSize(); got != tc.want {
+				t.Fatalf("tusChunkSize() = %d, want %d", got, tc.want)
+			}
+		})
+	}
+}
+
+func TestUploadVaultResumable_Unsupported(t *testing.T) {
+	srv := newFakeTus()
+	srv.unsupported = true
+	ts := httptest.NewServer(srv)
+	defer ts.Close()
+
+	fileName := "f.mp4"
+	withRecording(t, fileName, []byte("hello"))
+
+	uploaded, _, supported, _, _ := uploadVaultResumable(testVault(ts.URL), "pk", "dev", fileName, "test", "primary")
+	if uploaded {
+		t.Fatal("expected uploaded=false against a vault without a tus endpoint")
+	}
+	if supported {
+		t.Fatal("expected supported=false so the caller falls back to the legacy upload")
+	}
+}
+
+func TestUploadVaultResumable_FinalizeRetry(t *testing.T) {
+	srv := newFakeTus()
+	srv.failFinalize = 1
+	ts := httptest.NewServer(srv)
+	defer ts.Close()
+
+	fileName := "1564859471_6-474162_oprit_577-283-727-375_1153_27.mp4"
+	payload := bytes.Repeat([]byte("y"), 2048)
+	withRecording(t, fileName, payload)
+
+	uploaded, _, supported, _, err := uploadVaultResumable(testVault(ts.URL), "pk", "dev", fileName, "test", "primary")
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if !uploaded || !supported {
+		t.Fatalf("expected success after a failed finalize + restart, got uploaded=%v supported=%v", uploaded, supported)
+	}
+	if got := srv.createCount(); got < 2 {
+		t.Fatalf("expected at least 2 create requests (restart after failed finalize), got %d", got)
+	}
+}
+
+func TestUploadVaultResumable_ResumeFromSidecar(t *testing.T) {
+	srv := newFakeTus()
+	ts := httptest.NewServer(srv)
+	defer ts.Close()
+
+	fileName := "1564859471_6-474162_oprit_577-283-727-375_1153_27.mp4"
+	total := 8192
+	half := 4096
+	payload := bytes.Repeat([]byte("z"), total)
+	withRecording(t, fileName, payload)
+
+	// Simulate a previous run that uploaded half the file before being interrupted.
+	id := srv.seed(int64(total), int64(half))
+	baseURL := strings.TrimRight(ts.URL, "/") + tusUploadPath
+	saveTusResumeState(tusSidecarPath(fileName, "primary"), tusResumeState{
+		UploadURL: strings.TrimRight(baseURL, "/") + "/" + id,
+		VaultURI:  baseURL,
+		Size:      int64(total),
+	})
+
+	uploaded, _, supported, _, err := uploadVaultResumable(testVault(ts.URL), "pk", "dev", fileName, "test", "primary")
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if !uploaded || !supported {
+		t.Fatalf("expected resume success, got uploaded=%v supported=%v", uploaded, supported)
+	}
+	if got := srv.lastPatch(); got != int64(total-half) {
+		t.Fatalf("resume should only send the remaining %d bytes, sent %d", total-half, got)
+	}
+	if srv.createCount() != 0 {
+		t.Fatalf("resume should not create a new upload, got %d creates", srv.createCount())
+	}
+}
+
+func testHubConfig(hubURI string) *models.Config {
+	return &models.Config{
+		Key:           "device-key",
+		HubURI:        hubURI,
+		HubKey:        "hubpub",
+		HubPrivateKey: "hubpriv",
+		S3:            &models.S3{Region: "eu-west"},
+	}
+}
+
+// decodeTusMetadata parses a tus Upload-Metadata header value ("key b64,key b64")
+// back into a map of decoded key/value pairs.
+func decodeTusMetadata(meta string) map[string]string {
+	out := map[string]string{}
+	if meta == "" {
+		return out
+	}
+	for _, pair := range strings.Split(meta, ",") {
+		parts := strings.SplitN(strings.TrimSpace(pair), " ", 2)
+		if parts[0] == "" {
+			continue
+		}
+		val := ""
+		if len(parts) == 2 {
+			if b, err := base64.StdEncoding.DecodeString(parts[1]); err == nil {
+				val = string(b)
+			}
+		}
+		out[parts[0]] = val
+	}
+	return out
+}
+
+func TestUploadHubResumable_HappyPath(t *testing.T) {
+	srv := newFakeTus()
+	ts := httptest.NewServer(srv)
+	defer ts.Close()
+
+	fileName := "1564859471_6-474162_oprit_577-283-727-375_1153_27.mp4"
+	payload := bytes.Repeat([]byte("h"), 4096)
+	withRecording(t, fileName, payload)
+
+	uploaded, _, supported, _, err := uploadHubResumable(testHubConfig(ts.URL), fileName, "test", "hub")
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if !uploaded || !supported {
+		t.Fatalf("uploaded/supported = %v/%v, want both true", uploaded, supported)
+	}
+	if got := srv.totalBytes(); got != int64(len(payload)) {
+		t.Fatalf("server received %d bytes, want %d", got, len(payload))
+	}
+
+	// The Hub auth headers must be present on every request type (POST/HEAD/PATCH),
+	// because Kerberos Hub validates them on each proxied request. Conversely the
+	// vault credentials/routing are injected by Kerberos Hub on the agent's behalf
+	// and must never be sent by the agent on the hub path.
+	for _, method := range []string{http.MethodPost, http.MethodHead, http.MethodPatch} {
+		reqs := srv.requestsForMethod(method)
+		if len(reqs) == 0 {
+			t.Fatalf("expected at least one %s request", method)
+		}
+		for _, req := range reqs {
+			if got := req.header.Get("X-Kerberos-Hub-PublicKey"); got != "hubpub" {
+				t.Errorf("%s: X-Kerberos-Hub-PublicKey = %q, want %q", method, got, "hubpub")
+			}
+			if got := req.header.Get("X-Kerberos-Hub-PrivateKey"); got != "hubpriv" {
+				t.Errorf("%s: X-Kerberos-Hub-PrivateKey = %q, want %q", method, got, "hubpriv")
+			}
+			if got := req.header.Get("X-Kerberos-Hub-Region"); got != "eu-west" {
+				t.Errorf("%s: X-Kerberos-Hub-Region = %q, want %q", method, got, "eu-west")
+			}
+			if got := req.header.Get("X-Kerberos-Storage-Device"); got != "device-key" {
+				t.Errorf("%s: X-Kerberos-Storage-Device = %q, want %q", method, got, "device-key")
+			}
+			for _, h := range []string{
+				"X-Kerberos-Storage-AccessKey",
+				"X-Kerberos-Storage-SecretAccessKey",
+				"X-Kerberos-Storage-CloudKey",
+				"X-Kerberos-Storage-Provider",
+				"X-Kerberos-Storage-Directory",
+			} {
+				if got := req.header.Get(h); got != "" {
+					t.Errorf("%s: %s should be empty on the hub path, got %q", method, h, got)
+				}
+			}
+		}
+	}
+
+	// The creation request carries the upload metadata; on the hub path it must
+	// omit directory/provider/cloudkey (Hub resolves those) but include
+	// filename/device/capture. The filename header is also set on create.
+	posts := srv.requestsForMethod(http.MethodPost)
+	if got := posts[0].header.Get("X-Kerberos-Storage-FileName"); got != fileName {
+		t.Errorf("POST X-Kerberos-Storage-FileName = %q, want %q", got, fileName)
+	}
+	meta := decodeTusMetadata(posts[0].header.Get("Upload-Metadata"))
+	for _, omitted := range []string{"directory", "provider", "cloudkey"} {
+		if _, ok := meta[omitted]; ok {
+			t.Errorf("hub metadata must omit %q, got %v", omitted, meta)
+		}
+	}
+	if meta["filename"] != fileName {
+		t.Errorf("hub metadata filename = %q, want %q", meta["filename"], fileName)
+	}
+	if meta["device"] != "device-key" {
+		t.Errorf("hub metadata device = %q, want %q", meta["device"], "device-key")
+	}
+	if meta["capture"] != "IPCamera" {
+		t.Errorf("hub metadata capture = %q, want %q", meta["capture"], "IPCamera")
+	}
+}
+
+func TestUploadHubResumable_Unsupported(t *testing.T) {
+	srv := newFakeTus()
+	srv.unsupported = true
+	ts := httptest.NewServer(srv)
+	defer ts.Close()
+
+	fileName := "f.mp4"
+	withRecording(t, fileName, []byte("hello"))
+
+	uploaded, _, supported, _, _ := uploadHubResumable(testHubConfig(ts.URL), fileName, "test", "hub")
+	if uploaded {
+		t.Fatal("expected uploaded=false against a hub without a tus endpoint")
+	}
+	if supported {
+		t.Fatal("expected supported=false so the caller falls back to the legacy upload")
+	}
+}
+
+func TestEncodeTusMetadata(t *testing.T) {
+	got := encodeTusMetadata(map[string]string{
+		"b":     "2",
+		"a":     "1",
+		"empty": "",
+	})
+	// keys sorted, empty values skipped, values base64-encoded.
+	want := "a MQ==,b Mg=="
+	if got != want {
+		t.Fatalf("encodeTusMetadata = %q, want %q", got, want)
+	}
+}
+
+func TestResolveTusLocation(t *testing.T) {
+	cases := []struct {
+		name     string
+		base     string
+		location string
+		want     string
+	}{
+		{
+			name:     "absolute path location",
+			base:     "http://host/storage/tus/",
+			location: "/storage/tus/abc",
+			want:     "http://host/storage/tus/abc",
+		},
+		{
+			name:     "absolute url keeps configured host",
+			base:     "http://host/storage/tus/",
+			location: "http://internal:8080/storage/tus/xyz",
+			want:     "http://host/storage/tus/xyz",
+		},
+		{
+			name:     "relative id",
+			base:     "http://host/api/storage/tus/",
+			location: "abc",
+			want:     "http://host/api/storage/tus/abc",
+		},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			if got := resolveTusLocation(tc.base, tc.location); got != tc.want {
+				t.Fatalf("resolveTusLocation(%q, %q) = %q, want %q", tc.base, tc.location, got, tc.want)
+			}
+		})
+	}
+}
+
+func TestTusResumeStateRoundTrip(t *testing.T) {
+	dir := t.TempDir()
+	old, _ := os.Getwd()
+	if err := os.Chdir(dir); err != nil {
+		t.Fatalf("chdir: %v", err)
+	}
+	defer os.Chdir(old)
+
+	path := tusSidecarPath("file.mp4", "primary")
+	state := tusResumeState{UploadURL: "http://host/storage/tus/abc", VaultURI: "http://host/storage/tus/", Size: 123}
+	saveTusResumeState(path, state)
+
+	if got := loadTusResumeState(path, state.VaultURI); got != state.UploadURL {
+		t.Fatalf("loadTusResumeState = %q, want %q", got, state.UploadURL)
+	}
+	// A mismatched vault URI must not be reused.
+	if got := loadTusResumeState(path, "http://other/storage/tus/"); got != "" {
+		t.Fatalf("loadTusResumeState with mismatched vault = %q, want empty", got)
+	}
+}

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,23 +32,15 @@ const MacEpochOffset uint64 = 2082844800
 // resulting in ~3 second fragments (assuming a typical GOP interval).
 const FragmentDurationMs = 3000
 
-// MinNormalGOPMs is the maximum spacing between two consecutive IDRs that
-// we still consider an anomalous "loop/restart seam". When two keyframes
-// arrive closer than this, we treat the second one as an upstream
-// restart/loop-seam and force a fresh fragment so the seam IDR cannot end
-// up as a mid-fragment sync sample. The check only runs when the current
-// fragment has not yet reached FragmentDurationMs.
-//
-// This must be set well below the smallest plausible *legitimate* GOP
-// length. Typical IP cameras use GOP intervals of 1000-2000 ms, and the
-// arrival timing of consecutive IDRs can jitter by a few hundred ms due to
-// network/RTSP buffering. A threshold close to 1 s (e.g. 950) caused
-// false positives on cameras with ~1 s GOPs (warnings like
-// "gap=800 ms / 300 ms / 200 ms" while the stream itself was healthy).
-// 400 ms is comfortably below any realistic GOP yet still catches the
-// virtual-rtsp / ffmpeg loop-seam pattern (seam IDRs typically arrive
-// 100-200 ms after the prior IDR).
-const MinNormalGOPMs = 400
+// 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
@@ -74,8 +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)
-	LastKeyframeGapMs       uint64            // Gap (ms) between the previous two consecutive keyframes
 	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
@@ -94,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.
@@ -253,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 {
@@ -304,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++
@@ -327,37 +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.
-		//
-		// We only flag this as a seam when it is a *sudden* anomaly: the
-		// previous keyframe gap must have been healthy (>= MinNormalGOPMs).
-		// This avoids false positives on cameras that legitimately emit
-		// short-interval IDRs (short GOP, motion-triggered recovery IDRs,
-		// all-intra streams) where every keyframe would otherwise be flagged
-		// in a cascade, producing many tiny fragments and log spam.
-		if trackID == uint32(mp4.VideoTrack) && mp4.Start &&
-			mp4.LastKeyframeRawPTS > 0 && pts > mp4.LastKeyframeRawPTS {
-			gap := pts - mp4.LastKeyframeRawPTS
-			if !shouldFlush && gap < MinNormalGOPMs &&
-				(mp4.LastKeyframeGapMs == 0 || mp4.LastKeyframeGapMs >= MinNormalGOPMs) {
-				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", gap, elapsed))
-				shouldFlush = true
-			}
-			mp4.LastKeyframeGapMs = gap
-		}
-		if trackID == uint32(mp4.VideoTrack) {
-			mp4.LastKeyframeRawPTS = pts
-		}
-
 		if shouldFlush {
 			// Write the previous segment to the file
 			if mp4.Start {
@@ -444,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
@@ -462,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
@@ -505,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))
 
@@ -639,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())
@@ -665,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

@@ -1,7 +1,6 @@
 package video
 
 import (
-	"fmt"
 	"os"
 	"testing"
 
@@ -9,20 +8,39 @@ import (
 	"github.com/kerberos-io/agent/machinery/src/models"
 )
 
-// TestMP4LoopSeamIsolation reproduces the loop-seam pattern from the
-// failing virtual-rtsp recordings: ~1s GOPs, but at the source-MP4
-// loop boundary an IDR arrives prematurely (~200-870ms after the
-// previous IDR). Without the fix this seam IDR ends up bunched into
-// the same fragment as the prior GOP's IDR which trips macOS
-// VideoToolbox (kVTVideoDecoderBadDataErr / -12909). The fix forces
-// a fragment flush whenever two IDRs arrive closer than MinNormalGOPMs.
-func TestMP4LoopSeamIsolation(t *testing.T) {
-	tmpFile := "/tmp/test_loop_seam.mp4"
-	defer os.Remove(tmpFile)
+// 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, [][]byte{sps}, [][]byte{pps}, nil, 30)
+	mp4Video := NewMP4(tmpFile.Name(), [][]byte{sps}, [][]byte{pps}, nil, 60)
 	mp4Video.SetWidth(1920)
 	mp4Video.SetHeight(1080)
 	v := mp4Video.AddVideoTrack("H264")
@@ -40,33 +58,77 @@ func TestMP4LoopSeamIsolation(t *testing.T) {
 	}
 
 	frameDur := uint64(33)
+	normalGOPms := uint64(gopFrames) * frameDur
 	pts := uint64(0)
-	emit := func(n int, gopLen int) {
-		for f := 0; f < n; f++ {
-			isKey := (f % gopLen) == 0
-			mp4Video.AddSampleToTrack(v, isKey, mk(isKey), pts)
-			pts += frameDur
+	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)
+	}
 
-	// 17 seconds of normal content (last "good" IDR at sec 17).
-	emit(17*30, 30)
-	// Seam: IDR arrives ~150ms after previous (vs normal ~1000ms).
-	// This matches the realistic virtual-rtsp / ffmpeg `-stream_loop`
-	// loop boundary, where the new clip's first IDR is emitted shortly
-	// after the previous clip's final IDR.
-	pts -= 820
-	emit(13*30, 30)
+	// 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, _ := os.Open(tmpFile)
+	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 {
@@ -79,24 +141,54 @@ func TestMP4LoopSeamIsolation(t *testing.T) {
 				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)
 					}
 				}
-				fmt.Printf("frag %d tfdt=%d samples_dur=%d keys@%v\n",
-					fragIdx, tfdt, offset, keys)
+				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 < MinNormalGOPMs {
-						t.Errorf("frag %d (tfdt=%d): two IDRs only %d ms apart "+
-							"in same fragment (< %d) - seam was not isolated",
-							fragIdx, tfdt, gap, MinNormalGOPMs)
+					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/aac_transcoder_stub.go

@@ -96,7 +96,7 @@ func NewAACTranscoder() (*AACTranscoder, error) {
 				buffered := t.outBuf.Len()
 				t.outMu.Unlock()
 				if buffered <= 8192 || buffered%16000 == 0 {
-					log.Log.Info("webrtc.aac_transcoder: ffmpeg produced PCMU bytes, buffered=" + strconv.Itoa(buffered))
+					log.Log.Debug("webrtc.aac_transcoder: ffmpeg produced PCMU bytes, buffered=" + strconv.Itoa(buffered))
 				}
 			}
 			if readErr != nil {
@@ -129,14 +129,14 @@ func (t *AACTranscoder) Transcode(adtsData []byte) ([]byte, error) {
 		return nil, err
 	}
 	if len(adtsData) <= 512 || len(adtsData)%1024 == 0 {
-		log.Log.Info("webrtc.aac_transcoder: wrote AAC bytes to ffmpeg, input=" + strconv.Itoa(len(adtsData)))
+		log.Log.Debug("webrtc.aac_transcoder: wrote AAC bytes to ffmpeg, input=" + strconv.Itoa(len(adtsData)))
 	}
 
 	deadline := time.Now().Add(75 * time.Millisecond)
 	for {
 		data := t.readAvailable()
 		if len(data) > 0 {
-			log.Log.Info("webrtc.aac_transcoder: returning PCMU bytes=" + strconv.Itoa(len(data)))
+			log.Log.Debug("webrtc.aac_transcoder: returning PCMU bytes=" + strconv.Itoa(len(data)))
 			return data, nil
 		}
 
@@ -144,7 +144,7 @@ func (t *AACTranscoder) Transcode(adtsData []byte) ([]byte, error) {
 			if stderr := t.stderrString(); stderr != "" {
 				log.Log.Warning("webrtc.aac_transcoder: no output before deadline, ffmpeg stderr: " + stderr)
 			} else {
-				log.Log.Info("webrtc.aac_transcoder: no PCMU output before deadline")
+				log.Log.Debug("webrtc.aac_transcoder: no PCMU output before deadline")
 			}
 			return nil, nil
 		}

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
@@ -956,7 +988,7 @@ func processAudioPacket(pkt packets.Packet, state *streamState, audioBroadcaster
 		if len(pcmu) == 0 {
 			state.aacNoOutput++
 			if state.aacNoOutput <= 5 || state.aacNoOutput%100 == 0 {
-				log.Log.Info(fmt.Sprintf("webrtc.main.processAudioPacket(): AAC packet produced no PCMU output yet (aac_packets=%d, no_output=%d, input_bytes=%d)", state.aacPacketsSeen, state.aacNoOutput, len(pkt.Data)))
+				log.Log.Debug(fmt.Sprintf("webrtc.main.processAudioPacket(): AAC packet produced no PCMU output yet (aac_packets=%d, no_output=%d, input_bytes=%d)", state.aacPacketsSeen, state.aacNoOutput, len(pkt.Data)))
 			}
 			return // decoder still buffering
 		}
@@ -972,7 +1004,7 @@ func processAudioPacket(pkt packets.Packet, state *streamState, audioBroadcaster
 		state.lastAudioSample.Duration = sampleDuration(pkt, state.lastAudioSample.PacketTimestamp, 20*time.Millisecond)
 		state.audioSamplesSent++
 		if state.audioSamplesSent <= 5 || state.audioSamplesSent%100 == 0 {
-			log.Log.Info(fmt.Sprintf("webrtc.main.processAudioPacket(): queueing audio sample (samples=%d, codec=%s, bytes=%d, duration_ms=%d, peers=%d)", state.audioSamplesSent, pkt.Codec, len(state.lastAudioSample.Data), state.lastAudioSample.Duration.Milliseconds(), audioBroadcaster.PeerCount()))
+			log.Log.Debug(fmt.Sprintf("webrtc.main.processAudioPacket(): queueing audio sample (samples=%d, codec=%s, bytes=%d, duration_ms=%d, peers=%d)", state.audioSamplesSent, pkt.Codec, len(state.lastAudioSample.Data), state.lastAudioSample.Duration.Milliseconds(), audioBroadcaster.PeerCount()))
 		}
 		audioBroadcaster.WriteSample(*state.lastAudioSample)
 	}

machinery/swag.sh

@@ -1,2 +1,2 @@
 #!/bin/bash
-swag init -g ./src/routers/http/Server.go
+swag init -g ./src/routers/http/server.go