Merge pull request #340 from kerberos-io/fix/hub-failed-error-messages

fix/hub-failed-error-messages

.devcontainer/devcontainer.json

@@ -18,7 +18,7 @@
 	},
 	"forwardPorts": [
 		3000,
-		8080
+		8082
 	],
 	"postCreateCommand": "cd ui && yarn install && yarn build && cd ../machinery && go mod download && bash ./verify-moq-devcontainer.sh"
 }

.github/workflows/pr-build.yml

@@ -10,7 +10,7 @@ jobs:
   build:
     runs-on: ${{ matrix.runner }}
     permissions:
-      contents: write
+      contents: read
     strategy:
       matrix:
         include:
@@ -21,11 +21,6 @@ jobs:
     steps:
       - name: Checkout
         uses: actions/checkout@v4
-      - name: Login to DockerHub
-        uses: docker/login-action@v3
-        with:
-          username: ${{ secrets.DOCKER_USERNAME }}
-          password: ${{ secrets.DOCKER_PASSWORD }}
       - name: Run Build
         run: |
           docker build --build-arg VERSION=$(echo "${{ github.event.pull_request.head.sha }}" | cut -c1-7) -t ${{ matrix.architecture }} .

.vscode/launch.json

@@ -14,7 +14,7 @@
                 "-action",
                 "run",
                 "-port",
-                "8080"
+                "8082"
             ],
             "envFile": "${workspaceFolder}/machinery/.env.local",
             "buildFlags": "--tags dynamic,moq",

README.md

@@ -365,11 +365,13 @@ See [RTSPS and TLS certificates](README-RTSPS-TLS.md) for the complete Bosch UI,
 | `AGENT_KERBEROSVAULT_SECRET_KEY`            | The secret key of a Kerberos Vault account.                                                     | ""                             |
 | `AGENT_KERBEROSVAULT_PROVIDER`              | A Kerberos Vault provider you have created (optional).                                          | ""                             |
 | `AGENT_KERBEROSVAULT_DIRECTORY`             | The directory, in the Kerberos vault, where the recordings will be stored.                      | ""                             |
+| `AGENT_KERBEROSVAULT_CUSTOM_HEADERS`        | JSON object of custom upload headers that Vault includes as integration metadata.               | `{"site_id":"site-1"}`         |
 | `AGENT_KERBEROSVAULT_SECONDARY_URI`         | The Kerberos Vault API url.                                                                     | "https://vault.domain.com/api" |
 | `AGENT_KERBEROSVAULT_SECONDARY_ACCESS_KEY`  | The access key of a secondary Kerberos Vault account.                                           | ""                             |
 | `AGENT_KERBEROSVAULT_SECONDARY_SECRET_KEY`  | The secret key of a secondary Kerberos Vault account.                                           | ""                             |
 | `AGENT_KERBEROSVAULT_SECONDARY_PROVIDER`    | A secondary Kerberos Vault provider you have created (optional).                                | ""                             |
 | `AGENT_KERBEROSVAULT_SECONDARY_DIRECTORY`   | The directory, in the secondary Kerberos vault, where the recordings will be stored.            | ""                             |
+| `AGENT_KERBEROSVAULT_SECONDARY_CUSTOM_HEADERS` | JSON object of custom upload headers for the secondary Vault.                                | `{"site_id":"site-1"}`         |
 | `AGENT_DROPBOX_ACCESS_TOKEN`                | The Access Token from your Dropbox app, that is used to leverage the Dropbox SDK.               | ""                             |
 | `AGENT_DROPBOX_DIRECTORY`                   | The directory, in Dropbox, where the recordings will be stored.                                 | ""                             |
 | `AGENT_ENCRYPTION`                          | Enable 'true' or disable 'false' end-to-end encryption for MQTT messages.                       | "false"                        |
@@ -408,6 +410,94 @@ Smaller chunks provide more frequent resumable checkpoints but create more HTTP
 requests. Larger chunks reduce request overhead but require more data to be
 retransmitted when a request fails.
 
+### Custom metadata headers for Kerberos Vault
+
+Use `AGENT_KERBEROSVAULT_CUSTOM_HEADERS` to attach your own metadata (for
+example a site, line, or equipment identifier) to every recording uploaded to
+Kerberos Vault. The value is a JSON object whose keys are header names and whose
+values are strings. The Agent sends these as HTTP headers on every upload, and
+Vault adds them to the `metadata` object of its integration messages (Kafka,
+MQTT, SQS, RabbitMQ, etc.):
+
+```json
+{
+  "timestamp": "2026-09-24T15:06:45Z",
+  "key": "agent1/1790262405_3-425_agent1_0-0-0-0_-1_10000.mp4",
+  "device": "agent1",
+  "metadata": {
+    "site_id": "site-1",
+    "line_id": "line-2",
+    "equipment_id": "equipment-3"
+  }
+}
+```
+
+The value must be valid JSON on a single line, so wrap it in single quotes in a
+shell:
+
+```bash
+docker run -p 80:80 --name mycamera \
+  -e AGENT_CLOUD=kstorage \
+  -e AGENT_KERBEROSVAULT_URI=https://vault.domain.com/api \
+  -e AGENT_KERBEROSVAULT_ACCESS_KEY=xxx \
+  -e AGENT_KERBEROSVAULT_SECRET_KEY=xxx \
+  -e AGENT_KERBEROSVAULT_DIRECTORY=agent1 \
+  -e AGENT_KERBEROSVAULT_CUSTOM_HEADERS='{"site_id":"site-1","line_id":"line-2","equipment_id":"equipment-3"}' \
+  -d --restart=always kerberos/agent:latest
+```
+
+In `docker compose` or an `.env` file:
+
+```yaml
+services:
+  agent:
+    image: kerberos/agent:latest
+    environment:
+      AGENT_CLOUD: kstorage
+      AGENT_KERBEROSVAULT_CUSTOM_HEADERS: '{"site_id":"site-1","line_id":"line-2","equipment_id":"equipment-3"}'
+```
+
+```dotenv
+AGENT_KERBEROSVAULT_CUSTOM_HEADERS={"site_id":"site-1","line_id":"line-2","equipment_id":"equipment-3"}
+```
+
+In Kubernetes, quote the value so YAML keeps it as a string:
+
+```yaml
+env:
+  - name: AGENT_KERBEROSVAULT_CUSTOM_HEADERS
+    value: '{"site_id":"site-1","line_id":"line-2","equipment_id":"equipment-3"}'
+```
+
+Use `AGENT_KERBEROSVAULT_SECONDARY_CUSTOM_HEADERS` for the secondary Vault. In
+ConfigMap deployments you can set shared defaults with the `GLOBAL_` prefix
+(for example `GLOBAL_AGENT_KERBEROSVAULT_CUSTOM_HEADERS`) and override them per
+agent with `AGENT_KERBEROSVAULT_CUSTOM_HEADERS`. You can also set the same JSON
+under **Settings → Persistence → Custom upload headers** in the Agent UI, which
+stores it as `kstorage.custom_headers` in `config.json`.
+
+The rules for this value are:
+
+- It must be a JSON object with string values. Numbers, `null`, arrays, and
+  nested objects are rejected.
+- It can contain up to 32 headers, with names up to 128 characters and values
+  up to 4096 bytes.
+- Header names must be valid HTTP tokens (letters, digits, and ``!#$%&'*+-.^_`|~``).
+  Keys are matched case-insensitively, so `site_id` and `SITE_ID` cannot both
+  be used.
+- Reserved names are rejected: anything starting with `X-Kerberos-`,
+  `Authorization`, `Cookie`, `Host`, `Content-Type`, `Content-Length`,
+  `Trace-Id`, tus protocol headers (`Tus-Resumable`, `Upload-*`), and
+  hop-by-hop headers.
+- Values cannot contain line breaks, tabs, or other control characters.
+
+If the value is invalid, the Agent logs an error and does not upload to that
+Vault until the configuration is fixed. The recording stays queued, so it will
+upload once the value is corrected. Header names reach Vault unchanged, but
+proxies may normalise their casing. If Vault redirects an upload to another
+host, the Agent removes the custom headers and Vault credentials before
+following the redirect.
+
 
 ## Encryption
 
@@ -512,7 +602,10 @@ The `web` is a **React** project which is the main entry point for an end user t
     cd ui
     yarn start
 
-This will start a webserver and launches the web app on port `3000`.
+This starts the web app on port `3000`. In development, the UI connects to the
+Agent API on port `8082`. Override that port when needed:
+
+    REACT_APP_AGENT_API_PORT=9090 yarn start
 
 ![login-agent](./assets/img/agent-login.gif)
 
@@ -528,9 +621,14 @@ You can simply run the `machinery` using following commands.
 
     git clone https://github.com/kerberos-io/agent
     cd machinery
-    go run main.go -action run -port 80
+    go run main.go -action run -port 8082
 
-This will launch the Kerberos Agent and run a webserver on port `80`. You can change the port by your own preference. We strongly support the usage of [Goland](https://www.jetbrains.com/go/) or [Visual Studio Code](https://code.visualstudio.com/), as it comes with all the debugging and linting features built in.
+This will launch the Kerberos Agent and run a webserver on port `8082`, matching
+the UI development default. You can change the port together with
+`REACT_APP_AGENT_API_PORT`. We strongly support the usage of
+[Goland](https://www.jetbrains.com/go/) or
+[Visual Studio Code](https://code.visualstudio.com/), as it comes with all the
+debugging and linting features built in.
 
 ![VSCode desktop](./assets/img/vscode-desktop.png)
 

machinery/docs/docs.go

@@ -1296,6 +1296,9 @@ const docTemplate = `{
                     "description": "old way, remove this",
                     "type": "string"
                 },
+                "custom_headers": {
+                    "type": "string"
+                },
                 "directory": {
                     "type": "string"
                 },

machinery/docs/swagger.json

@@ -1289,6 +1289,9 @@
                     "description": "old way, remove this",
                     "type": "string"
                 },
+                "custom_headers": {
+                    "type": "string"
+                },
                 "directory": {
                     "type": "string"
                 },

machinery/go.mod

@@ -25,7 +25,7 @@ require (
 	github.com/gorilla/websocket v1.5.3
 	github.com/kellydunn/golang-geo v0.7.0
 	github.com/kerberos-io/joy4 v1.0.64
-	github.com/kerberos-io/onvif v1.2.2
+	github.com/kerberos-io/onvif v1.2.3
 	github.com/minio/minio-go/v6 v6.0.57
 	github.com/moq-dev/moq-go v0.5.7
 	github.com/nfnt/resize v0.0.0-20180221191011-83c6a9932646

machinery/go.sum

@@ -776,8 +776,8 @@ github.com/kellydunn/golang-geo v0.7.0 h1:A5j0/BvNgGwY6Yb6inXQxzYwlPHc6WVZR+Mrar
 github.com/kellydunn/golang-geo v0.7.0/go.mod h1:YYlQPJ+DPEzrHx8kT3oPHC/NjyvCCXE+IuKGKdrjrcU=
 github.com/kerberos-io/joy4 v1.0.64 h1:gTUSotHSOhp9mNqEecgq88tQHvpj7TjmrvPUsPm0idg=
 github.com/kerberos-io/joy4 v1.0.64/go.mod h1:nZp4AjvKvTOXRrmDyAIOw+Da+JA5OcSo/JundGfOlFU=
-github.com/kerberos-io/onvif v1.2.2 h1:QnxITps7xvAVD2abWRsa3+p9QexjKESQQldUFMx/mYA=
-github.com/kerberos-io/onvif v1.2.2/go.mod h1:XSgEQXmEDjUQTbdXvsaRJt6Az8YPGj7L+j5iXKEGijU=
+github.com/kerberos-io/onvif v1.2.3 h1:RDnDvLBd92VCgFOrlbGavzCVN9lQA/qs9novdNFIAOU=
+github.com/kerberos-io/onvif v1.2.3/go.mod h1:XSgEQXmEDjUQTbdXvsaRJt6Az8YPGj7L+j5iXKEGijU=
 github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8=
 github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
 github.com/klauspost/asmfmt v1.3.2/go.mod h1:AG8TuvYojzulgDAMCnYn50l/5QV3Bs/tp6j0HLHbNSE=

machinery/src/capture/gortsplib.go

@@ -32,6 +32,7 @@ import (
 	"github.com/bluenviron/gortsplib/v5/pkg/format/rtplpcm"
 	"github.com/bluenviron/gortsplib/v5/pkg/format/rtpmpeg4audio"
 	"github.com/bluenviron/gortsplib/v5/pkg/format/rtpsimpleaudio"
+	"github.com/bluenviron/mediacommon/pkg/bits"
 	"github.com/bluenviron/mediacommon/pkg/codecs/h264"
 	"github.com/bluenviron/mediacommon/pkg/codecs/h265"
 	"github.com/bluenviron/mediacommon/pkg/codecs/mpeg4audio"
@@ -159,13 +160,10 @@ type Golibrtsp struct {
 }
 
 // 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.
+// packet on a single read goroutine, so prolonged queue lock contention or
+// process starvation can delay socket reads. Queue consumers process packets
+// after releasing the lock; a slow recorder therefore falls behind its cursor
+// instead of directly blocking WritePacket.
 type streamHealth struct {
 	mu          sync.Mutex
 	windowStart time.Time
@@ -178,6 +176,152 @@ type streamHealth struct {
 	decodeErrs  int64
 }
 
+type videoReferenceState struct {
+	haveSequence       bool
+	lastSequence       uint16
+	valid              bool
+	droppedAccessUnits uint64
+}
+
+type h264FrameNumTracker struct {
+	sps          *h264.SPS
+	haveFrameNum bool
+	lastFrameNum uint32
+}
+
+const maxH264SliceHeaderBytes = 64
+
+func (t *h264FrameNumTracker) setSPS(sps *h264.SPS) {
+	if sps == nil {
+		return
+	}
+	if t.sps == nil ||
+		t.sps.Log2MaxFrameNumMinus4 != sps.Log2MaxFrameNumMinus4 ||
+		t.sps.SeparateColourPlaneFlag != sps.SeparateColourPlaneFlag ||
+		t.sps.FrameMbsOnlyFlag != sps.FrameMbsOnlyFlag {
+		t.haveFrameNum = false
+	}
+	copy := *sps
+	t.sps = &copy
+}
+
+func (t *h264FrameNumTracker) observe(au [][]byte) (uint32, uint32, bool, bool) {
+	if t.sps == nil || !t.sps.FrameMbsOnlyFlag {
+		return 0, 0, false, false
+	}
+
+	for _, nalu := range au {
+		if len(nalu) == 0 {
+			continue
+		}
+		naluType := h264.NALUType(nalu[0] & 0x1F)
+		if naluType != h264.NALUTypeIDR && naluType != h264.NALUTypeNonIDR {
+			continue
+		}
+
+		frameNum, err := h264SliceFrameNum(nalu, t.sps)
+		if err != nil {
+			return 0, 0, false, false
+		}
+		if naluType == h264.NALUTypeIDR {
+			t.haveFrameNum = true
+			t.lastFrameNum = frameNum
+			return frameNum, frameNum, false, true
+		}
+
+		// Non-reference pictures do not advance the reference-picture frame_num
+		// sequence and therefore cannot prove that a reference picture is missing.
+		if (nalu[0]>>5)&0x03 == 0 {
+			return frameNum, frameNum, false, true
+		}
+		if !t.haveFrameNum {
+			t.haveFrameNum = true
+			t.lastFrameNum = frameNum
+			return frameNum, frameNum, false, true
+		}
+
+		frameNumBits := t.sps.Log2MaxFrameNumMinus4 + 4
+		if frameNumBits > 16 {
+			return 0, 0, false, false
+		}
+		maxFrameNum := uint32(1) << frameNumBits
+		expected := (t.lastFrameNum + 1) & (maxFrameNum - 1)
+		// When the SPS permits frame_num gaps, the decoder is required to
+		// synthesize non-existing pictures and the discontinuity is not evidence
+		// of transport loss. RTP/depacketizer errors still invalidate references.
+		gap := frameNum != expected && !t.sps.GapsInFrameNumValueAllowedFlag
+		t.lastFrameNum = frameNum
+		return expected, frameNum, gap, true
+	}
+
+	return 0, 0, false, false
+}
+
+func h264SliceFrameNum(nalu []byte, sps *h264.SPS) (uint32, error) {
+	if len(nalu) < 2 || sps == nil {
+		return 0, errors.New("H264 slice or SPS is missing")
+	}
+	frameNumBits := sps.Log2MaxFrameNumMinus4 + 4
+	if frameNumBits > 16 {
+		return 0, fmt.Errorf("invalid H264 frame_num width: %d", frameNumBits)
+	}
+	header := nalu[1:]
+	if len(header) > maxH264SliceHeaderBytes {
+		header = header[:maxH264SliceHeaderBytes]
+	}
+	header = h264.EmulationPreventionRemove(header)
+	pos := 0
+
+	for range 3 {
+		if _, err := bits.ReadGolombUnsigned(header, &pos); err != nil {
+			return 0, fmt.Errorf("read H264 slice header: %w", err)
+		}
+	}
+	if sps.SeparateColourPlaneFlag {
+		if _, err := bits.ReadBits(header, &pos, 2); err != nil {
+			return 0, fmt.Errorf("read H264 colour plane: %w", err)
+		}
+	}
+	frameNum, err := bits.ReadBits(header, &pos, int(frameNumBits))
+	if err != nil {
+		return 0, fmt.Errorf("read H264 frame_num: %w", err)
+	}
+	return uint32(frameNum), nil
+}
+
+func (s *videoReferenceState) observeSequence(sequence uint16) (uint16, bool) {
+	expected := sequence
+	gap := false
+	if s.haveSequence {
+		expected = s.lastSequence + 1
+		gap = sequence != expected
+		if gap {
+			s.valid = false
+		}
+	}
+	s.haveSequence = true
+	s.lastSequence = sequence
+	return expected, gap
+}
+
+func (s *videoReferenceState) invalidate() {
+	s.valid = false
+}
+
+func (s *videoReferenceState) accept(randomAccess bool) (bool, uint64) {
+	if randomAccess {
+		dropped := s.droppedAccessUnits
+		s.droppedAccessUnits = 0
+		s.valid = true
+		return true, dropped
+	}
+	if !s.valid {
+		s.droppedAccessUnits++
+		return false, 0
+	}
+	return true, 0
+}
+
 const (
 	streamHealthWindow    = 10 * time.Second
 	streamHealthWriteWarn = 150 * time.Millisecond
@@ -190,9 +334,9 @@ func newStreamHealth() *streamHealth {
 }
 
 // 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.
+// previous frame and how long WritePacket spent acquiring and updating the
+// queue. 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
@@ -264,9 +408,9 @@ func (h *streamHealth) observePacket(streamType string, writeDur time.Duration)
 	}
 }
 
-// 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.
+// observeLost is invoked by gortsplib when RTP sequence numbers skip. With the
+// configured TCP transport, the sender or an upstream RTSP component omitted
+// those packets; sender-side pressure is one possible cause.
 func (h *streamHealth) observeLost(streamType string, lost uint64) {
 	if h == nil {
 		return
@@ -887,6 +1031,14 @@ func (g *Golibrtsp) Start(ctx context.Context, streamType string, queue *packets
 	// called when a video RTP packet arrives for H264
 	var filteredAU [][]byte
 	if g.VideoH264Media != nil && g.VideoH264Forma != nil {
+		referenceState := &videoReferenceState{}
+		frameNumTracker := &h264FrameNumTracker{}
+		if len(g.VideoH264Forma.SPS) > 0 {
+			var sps h264.SPS
+			if sps.Unmarshal(g.VideoH264Forma.SPS) == nil {
+				frameNumTracker.setSPS(&sps)
+			}
+		}
 
 		// Extracts DTS from the bitstream to support B-frame H264 streams.
 		// Created once per stream (tracks reorder state across access units).
@@ -902,6 +1054,16 @@ func (g *Golibrtsp) Start(ctx context.Context, streamType string, queue *packets
 			default:
 			}
 
+			if expected, gap := referenceState.observeSequence(rtppkt.SequenceNumber); gap {
+				log.WithFields(log.Fields{
+					"component":                "capture",
+					"event":                    "video_rtp_sequence_gap",
+					"expected_sequence_number": expected,
+					"received_sequence_number": rtppkt.SequenceNumber,
+					"stream":                   streamType,
+				}).Warn("RTP sequence gap invalidated video reference chain")
+			}
+
 			if len(rtppkt.Payload) > 0 {
 
 				// decode timestamps — validate each call separately
@@ -917,6 +1079,9 @@ func (g *Golibrtsp) Start(ctx context.Context, streamType string, queue *packets
 				// this is a keyframe.
 				au, errDecode := g.VideoH264Decoder.Decode(rtppkt)
 				if errDecode != nil {
+					if errDecode != rtph264.ErrMorePacketsNeeded {
+						referenceState.invalidate()
+					}
 					if errDecode != rtph264.ErrNonStartingPacketAndNoPrevious && errDecode != rtph264.ErrMorePacketsNeeded {
 						log.Error("capture.golibrtsp.Start(): " + errDecode.Error())
 					}
@@ -961,6 +1126,7 @@ func (g *Golibrtsp) Start(ctx context.Context, streamType string, queue *packets
 						if errSPS == nil {
 							// Debug SPS information
 							g.debugSPSInfo(&sps, streamType)
+							frameNumTracker.setSPS(&sps)
 
 							// Get width
 							g.Streams[g.VideoH264Index].Width = sps.Width()
@@ -1022,6 +1188,39 @@ func (g *Golibrtsp) Start(ctx context.Context, streamType string, queue *packets
 					return
 				}
 
+				if expected, received, gap, ok := frameNumTracker.observe(au); ok && gap {
+					referenceState.invalidate()
+					log.WithFields(log.Fields{
+						"component":          "capture",
+						"event":              "h264_frame_num_gap",
+						"expected_frame_num": expected,
+						"received_frame_num": received,
+						"stream":             streamType,
+					}).Warn("H264 frame_num gap invalidated video reference chain")
+				}
+
+				accepted, recoveredDropped := referenceState.accept(idrPresent)
+				if !accepted {
+					if referenceState.droppedAccessUnits == 1 {
+						log.WithFields(log.Fields{
+							"codec":     "H264",
+							"component": "capture",
+							"event":     "dependent_access_units_dropped",
+							"stream":    streamType,
+						}).Warn("Dropping dependent video frames until the next IDR")
+					}
+					return
+				}
+				if recoveredDropped > 0 {
+					log.WithFields(log.Fields{
+						"codec":                "H264",
+						"component":            "capture",
+						"dropped_access_units": recoveredDropped,
+						"event":                "video_reference_chain_recovered",
+						"stream":               streamType,
+					}).Info("Video reference chain recovered at IDR")
+				}
+
 				if idrPresent {
 					log.WithFields(log.Fields{
 						"component":  "capture",
@@ -1165,6 +1364,7 @@ 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 {
+		referenceState := &videoReferenceState{}
 
 		// Extracts DTS from the bitstream to support B-frame H265 streams.
 		// Created once per stream (tracks reorder state across access units).
@@ -1180,6 +1380,16 @@ func (g *Golibrtsp) Start(ctx context.Context, streamType string, queue *packets
 			default:
 			}
 
+			if expected, gap := referenceState.observeSequence(rtppkt.SequenceNumber); gap {
+				log.WithFields(log.Fields{
+					"component":                "capture",
+					"event":                    "video_rtp_sequence_gap",
+					"expected_sequence_number": expected,
+					"received_sequence_number": rtppkt.SequenceNumber,
+					"stream":                   streamType,
+				}).Warn("RTP sequence gap invalidated video reference chain")
+			}
+
 			if len(rtppkt.Payload) > 0 {
 
 				// decode timestamps — validate each call separately
@@ -1195,6 +1405,9 @@ func (g *Golibrtsp) Start(ctx context.Context, streamType string, queue *packets
 				// this is a keyframe.
 				au, errDecode := g.VideoH265Decoder.Decode(rtppkt)
 				if errDecode != nil {
+					if errDecode != rtph265.ErrMorePacketsNeeded {
+						referenceState.invalidate()
+					}
 					if errDecode != rtph265.ErrNonStartingPacketAndNoPrevious && errDecode != rtph265.ErrMorePacketsNeeded {
 						log.Error("capture.golibrtsp.Start(): " + errDecode.Error())
 					}
@@ -1244,6 +1457,28 @@ func (g *Golibrtsp) Start(ctx context.Context, streamType string, queue *packets
 					return
 				}
 
+				accepted, recoveredDropped := referenceState.accept(isRandomAccess)
+				if !accepted {
+					if referenceState.droppedAccessUnits == 1 {
+						log.WithFields(log.Fields{
+							"codec":     "H265",
+							"component": "capture",
+							"event":     "dependent_access_units_dropped",
+							"stream":    streamType,
+						}).Warn("Dropping dependent video frames until the next random-access frame")
+					}
+					return
+				}
+				if recoveredDropped > 0 {
+					log.WithFields(log.Fields{
+						"codec":                "H265",
+						"component":            "capture",
+						"dropped_access_units": recoveredDropped,
+						"event":                "video_reference_chain_recovered",
+						"stream":               streamType,
+					}).Info("Video reference chain recovered at random-access frame")
+				}
+
 				// add VPS, SPS and PPS before random access access unit
 				if isRandomAccess {
 					au = append([][]byte{

machinery/src/capture/gortsplib_test.go

@@ -11,6 +11,8 @@ import (
 	"path/filepath"
 	"strings"
 	"testing"
+
+	"github.com/bluenviron/mediacommon/pkg/codecs/h264"
 )
 
 func TestPreRecordingGOPCount(t *testing.T) {
@@ -134,3 +136,114 @@ func TestRTSPSTLSConfig(t *testing.T) {
 		}
 	})
 }
+
+func TestVideoReferenceStateDropsDependentFramesUntilRandomAccess(t *testing.T) {
+	state := &videoReferenceState{}
+
+	accepted, recovered := state.accept(false)
+	if accepted || recovered != 0 || state.droppedAccessUnits != 1 {
+		t.Fatalf("initial dependent frame = (accepted=%t, recovered=%d, dropped=%d), want (false, 0, 1)",
+			accepted, recovered, state.droppedAccessUnits)
+	}
+
+	accepted, recovered = state.accept(true)
+	if !accepted || recovered != 1 || !state.valid {
+		t.Fatalf("random-access recovery = (accepted=%t, recovered=%d, valid=%t), want (true, 1, true)",
+			accepted, recovered, state.valid)
+	}
+
+	if expected, gap := state.observeSequence(100); gap || expected != 100 {
+		t.Fatalf("first sequence observation = (expected=%d, gap=%t), want (100, false)", expected, gap)
+	}
+	if expected, gap := state.observeSequence(102); !gap || expected != 101 || state.valid {
+		t.Fatalf("sequence gap = (expected=%d, gap=%t, valid=%t), want (101, true, false)",
+			expected, gap, state.valid)
+	}
+
+	accepted, recovered = state.accept(false)
+	if accepted || recovered != 0 || state.droppedAccessUnits != 1 {
+		t.Fatalf("post-loss dependent frame = (accepted=%t, recovered=%d, dropped=%d), want (false, 0, 1)",
+			accepted, recovered, state.droppedAccessUnits)
+	}
+
+	accepted, recovered = state.accept(true)
+	if !accepted || recovered != 1 || !state.valid {
+		t.Fatalf("post-loss recovery = (accepted=%t, recovered=%d, valid=%t), want (true, 1, true)",
+			accepted, recovered, state.valid)
+	}
+}
+
+func TestVideoReferenceStateAcceptsSequenceWraparound(t *testing.T) {
+	state := &videoReferenceState{valid: true}
+
+	if _, gap := state.observeSequence(^uint16(0)); gap {
+		t.Fatal("first sequence observation reported a gap")
+	}
+	if expected, gap := state.observeSequence(0); gap || expected != 0 || !state.valid {
+		t.Fatalf("sequence wraparound = (expected=%d, gap=%t, valid=%t), want (0, false, true)",
+			expected, gap, state.valid)
+	}
+}
+
+func TestH264FrameNumTrackerDetectsMissingIDR(t *testing.T) {
+	tracker := &h264FrameNumTracker{}
+	tracker.setSPS(&h264.SPS{
+		Log2MaxFrameNumMinus4: 1,
+		FrameMbsOnlyFlag:      true,
+	})
+
+	if _, _, gap, ok := tracker.observe([][]byte{testH264Slice(h264.NALUTypeIDR, 0)}); !ok || gap {
+		t.Fatalf("initial IDR observation = (ok=%t, gap=%t), want (true, false)", ok, gap)
+	}
+	for frameNum := uint32(1); frameNum <= 24; frameNum++ {
+		if expected, received, gap, ok := tracker.observe([][]byte{testH264Slice(h264.NALUTypeNonIDR, frameNum)}); !ok || gap {
+			t.Fatalf("frame_num %d observation = (expected=%d, received=%d, ok=%t, gap=%t), want no gap",
+				frameNum, expected, received, ok, gap)
+		}
+	}
+
+	expected, received, gap, ok := tracker.observe([][]byte{testH264Slice(h264.NALUTypeNonIDR, 1)})
+	if !ok || !gap || expected != 25 || received != 1 {
+		t.Fatalf("missing IDR transition = (expected=%d, received=%d, ok=%t, gap=%t), want (25, 1, true, true)",
+			expected, received, ok, gap)
+	}
+}
+
+func TestH264FrameNumTrackerResetsAtIDR(t *testing.T) {
+	tracker := &h264FrameNumTracker{}
+	tracker.setSPS(&h264.SPS{
+		Log2MaxFrameNumMinus4: 1,
+		FrameMbsOnlyFlag:      true,
+	})
+
+	tracker.observe([][]byte{testH264Slice(h264.NALUTypeIDR, 0)})
+	tracker.observe([][]byte{testH264Slice(h264.NALUTypeNonIDR, 1)})
+	if expected, received, gap, ok := tracker.observe([][]byte{testH264Slice(h264.NALUTypeIDR, 0)}); !ok || gap {
+		t.Fatalf("IDR reset = (expected=%d, received=%d, ok=%t, gap=%t), want no gap",
+			expected, received, ok, gap)
+	}
+	if expected, received, gap, ok := tracker.observe([][]byte{testH264Slice(h264.NALUTypeNonIDR, 1)}); !ok || gap || expected != 1 || received != 1 {
+		t.Fatalf("post-IDR frame = (expected=%d, received=%d, ok=%t, gap=%t), want (1, 1, true, false)",
+			expected, received, ok, gap)
+	}
+}
+
+func TestH264FrameNumTrackerHonorsPermittedGaps(t *testing.T) {
+	tracker := &h264FrameNumTracker{}
+	tracker.setSPS(&h264.SPS{
+		Log2MaxFrameNumMinus4:          1,
+		FrameMbsOnlyFlag:               true,
+		GapsInFrameNumValueAllowedFlag: true,
+	})
+
+	tracker.observe([][]byte{testH264Slice(h264.NALUTypeIDR, 0)})
+	expected, received, gap, ok := tracker.observe([][]byte{testH264Slice(h264.NALUTypeNonIDR, 2)})
+	if !ok || gap || expected != 1 || received != 2 {
+		t.Fatalf("permitted frame_num gap = (expected=%d, received=%d, ok=%t, gap=%t), want (1, 2, true, false)",
+			expected, received, ok, gap)
+	}
+}
+
+func testH264Slice(naluType h264.NALUType, frameNum uint32) []byte {
+	return []byte{0x60 | byte(naluType), 0xE0 | byte(frameNum&0x1F)}
+}

machinery/src/capture/main.go

@@ -1070,6 +1070,10 @@ func Base64Image(captureDevice *Capture, communication *models.Communication, co
 					count++
 					continue
 				}
+			} else {
+				// A stopped run closes this queue permanently. Retrying its
+				// cursor would spin on EOF even after the HTTP client leaves.
+				break
 			}
 		} else {
 			break
@@ -1113,6 +1117,10 @@ func JpegImage(captureDevice *Capture, communication *models.Communication) imag
 				} else {
 					break
 				}
+			} else {
+				// A stopped run closes this queue permanently. Retrying its
+				// cursor would spin on EOF even after the HTTP client leaves.
+				break
 			}
 		} else {
 			break

machinery/src/capture/snapshot_test.go

@@ -0,0 +1,68 @@
+package capture
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/kerberos-io/agent/machinery/src/models"
+	"github.com/kerberos-io/agent/machinery/src/packets"
+)
+
+func TestSnapshotReturnsAfterQueueClosure(t *testing.T) {
+	for _, stream := range []string{"main", "sub"} {
+		for _, format := range []string{"jpeg", "base64"} {
+			for _, alreadyClosed := range []bool{true, false} {
+				name := stream + "/" + format + "/closed_during_read"
+				if alreadyClosed {
+					name = stream + "/" + format + "/already_closed"
+				}
+				t.Run(name, func(t *testing.T) {
+					communication := &models.Communication{}
+					run := models.NewAgentRun(context.Background(), communication, false)
+					queue := packets.NewQueue()
+					captureDevice := &Capture{}
+					if stream == "main" {
+						captureDevice.SetMainClient("rtsp://unused.invalid/main")
+						run.SetMainQueue(queue)
+					} else {
+						captureDevice.SetSubClient("rtsp://unused.invalid/sub")
+						run.SetSubQueue(queue)
+					}
+					if err := run.Activate(); err != nil {
+						t.Fatal(err)
+					}
+					defer queue.Close()
+					if alreadyClosed {
+						queue.Close()
+					}
+					done := make(chan bool, 1)
+					go func() {
+						if format == "jpeg" {
+							image := JpegImage(captureDevice, communication)
+							done <- image.Rect.Empty()
+						} else {
+							done <- Base64Image(captureDevice, communication, &models.Configuration{}) == ""
+						}
+					}()
+					if !alreadyClosed {
+						select {
+						case <-done:
+							t.Fatal("snapshot returned before a frame or queue closure")
+						case <-time.After(10 * time.Millisecond):
+						}
+						queue.Close()
+					}
+					select {
+					case empty := <-done:
+						if !empty {
+							t.Fatal("closed empty queue produced a snapshot")
+						}
+					case <-time.After(time.Second):
+						t.Fatal("snapshot did not return after its packet queue closed")
+					}
+				})
+			}
+		}
+	}
+}

machinery/src/cloud/cloud.go

@@ -541,13 +541,23 @@ func (state *heartbeatONVIFState) fetchEvents(device *goonvif.Device) ([]byte, b
 		}
 	}
 
-	if !initialStateSubscriptionCreated && state.loopPullPoint == "" {
+	// Fall back to querying the relay outputs and digital inputs directly when
+	// no subscription could be created, and also when the subscriptions
+	// produced no I/O events: some devices accept the subscription but reject
+	// the pulls, or never publish the initial I/O state.
+	noSubscription := !initialStateSubscriptionCreated && state.loopPullPoint == ""
+	if noSubscription || heartbeatEventsListEmpty(onvifEventsList) {
 		return state.fallbackEvents(device), operationFailed
 	}
 
 	return onvifEventsList, operationFailed
 }
 
+func heartbeatEventsListEmpty(eventsList []byte) bool {
+	trimmed := strings.TrimSpace(string(eventsList))
+	return trimmed == "" || trimmed == "[]" || trimmed == "null"
+}
+
 func getHeartbeatONVIFPayload(camera models.IPCamera, state *heartbeatONVIFState) heartbeatONVIFPayload {
 	if state == nil {
 		state = newHeartbeatONVIFState()
@@ -1444,13 +1454,19 @@ func HandleRealtimeProcessing(processingCursor *packets.QueueCursor, configurati
 // @Summary Will verify the hub connectivity.
 // @Description Will verify the hub connectivity.
 // @Success 200 {object} models.APIResponse
-func VerifyHub(c *gin.Context) {
+func VerifyHub(c *gin.Context, configuredHubURI string) {
 
 	var config models.Config
 	err := c.BindJSON(&config)
 
 	if err == nil {
-		hubURI := config.HubURI
+		hubURI, err := configuredHubVerificationURI(config.HubURI, configuredHubURI)
+		if err != nil {
+			c.JSON(http.StatusBadRequest, models.APIResponse{
+				Data: "cloud.VerifyHub(): " + err.Error(),
+			})
+			return
+		}
 		publicKey := config.HubKey
 		privateKey := config.HubPrivateKey
 
@@ -1463,9 +1479,9 @@ func VerifyHub(c *gin.Context) {
 				tr := &http.Transport{
 					TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
 				}
-				client = &http.Client{Transport: tr}
+				client = &http.Client{Transport: tr, CheckRedirect: rejectHubVerificationRedirect}
 			} else {
-				client = &http.Client{}
+				client = &http.Client{CheckRedirect: rejectHubVerificationRedirect}
 			}
 
 			resp, err := client.Do(req)
@@ -1514,7 +1530,7 @@ func VerifyHub(c *gin.Context) {
 // @Summary Will verify the persistence.
 // @Description Will verify the persistence.
 // @Success 200 {object} models.APIResponse
-func VerifyPersistence(c *gin.Context, configDirectory string) {
+func VerifyPersistence(c *gin.Context, configDirectory, configuredHubURI string) {
 
 	var config models.Config
 	err := c.BindJSON(&config)
@@ -1527,6 +1543,7 @@ func VerifyPersistence(c *gin.Context, configDirectory string) {
 			if config.HubURI == "" ||
 				config.HubKey == "" ||
 				config.HubPrivateKey == "" ||
+				config.S3 == nil ||
 				config.S3.Region == "" {
 				msg := "cloud.VerifyPersistence(kerberoshub): Kerberos Hub not properly configured."
 				log.Error(msg)
@@ -1534,25 +1551,24 @@ func VerifyPersistence(c *gin.Context, configDirectory string) {
 					Data: msg,
 				})
 			} else {
-
-				// Open test-480p.mp4
-				file, err := os.Open(configDirectory + "/data/test-480p.mp4")
+				hubURI, err := configuredHubVerificationURI(config.HubURI, configuredHubURI)
 				if err != nil {
-					msg := "cloud.VerifyPersistence(kerberoshub): error reading test-480p.mp4: " + err.Error()
+					msg := "cloud.VerifyPersistence(kerberoshub): " + err.Error()
 					log.Error(msg)
-					c.JSON(400, models.APIResponse{
+					c.JSON(http.StatusBadRequest, models.APIResponse{
 						Data: msg,
 					})
+					return
 				}
-				defer file.Close()
 
-				req, err := http.NewRequest("POST", config.HubURI+"/storage/upload", file)
+				req, err := http.NewRequest(http.MethodHead, hubURI+"/storage/upload", nil)
 				if err != nil {
-					msg := "cloud.VerifyPersistence(kerberoshub): error reading Kerberos Hub HEAD request, " + config.HubURI + "/storage: " + err.Error()
+					msg := "cloud.VerifyPersistence(kerberoshub): error creating Kerberos Hub upload authorization request: " + err.Error()
 					log.Error(msg)
 					c.JSON(400, models.APIResponse{
 						Data: msg,
 					})
+					return
 				}
 
 				timestamp := time.Now().Unix()
@@ -1570,9 +1586,9 @@ func VerifyPersistence(c *gin.Context, configDirectory string) {
 					tr := &http.Transport{
 						TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
 					}
-					client = &http.Client{Transport: tr}
+					client = &http.Client{Transport: tr, CheckRedirect: rejectHubVerificationRedirect}
 				} else {
-					client = &http.Client{}
+					client = &http.Client{CheckRedirect: rejectHubVerificationRedirect}
 				}
 
 				resp, err := client.Do(req)
@@ -1588,14 +1604,17 @@ func VerifyPersistence(c *gin.Context, configDirectory string) {
 							Data: msg,
 						})
 					} else {
-						msg := "cloud.VerifyPersistence(kerberoshub): upload not allowed using the configured credentials"
+						msg := hubPersistenceVerificationError(resp.StatusCode)
 						log.Error(msg)
 						c.JSON(400, models.APIResponse{
 							Data: msg,
 						})
 					}
 				} else {
-					msg := "cloud.VerifyPersistence(kerberoshub): Error creating Kerberos Hub request"
+					msg := "cloud.VerifyPersistence(kerberoshub): Kerberos Hub upload authorization request failed"
+					if err != nil {
+						msg += ": " + err.Error()
+					}
 					log.Error(msg)
 					c.JSON(400, models.APIResponse{
 						Data: msg,
@@ -1764,6 +1783,35 @@ func VerifyPersistence(c *gin.Context, configDirectory string) {
 	}
 }
 
+func configuredHubVerificationURI(requested, configured string) (string, error) {
+	requested = strings.TrimRight(strings.TrimSpace(requested), "/")
+	configured = strings.TrimRight(strings.TrimSpace(configured), "/")
+	if configured == "" {
+		return "", errors.New("save the Hub URI before verifying it")
+	}
+	if requested != configured {
+		return "", errors.New("the Hub URI does not match the saved configuration; save it before verifying")
+	}
+	return configured, nil
+}
+
+func rejectHubVerificationRedirect(_ *http.Request, _ []*http.Request) error {
+	return http.ErrUseLastResponse
+}
+
+func hubPersistenceVerificationError(statusCode int) string {
+	reason := "Kerberos Hub rejected upload authorization"
+	switch statusCode {
+	case http.StatusBadRequest:
+		reason += ": the credentials or device/project assignment are invalid, or the storage limit was reached"
+	case http.StatusUnauthorized:
+		reason += ": the account has no active subscription or the subscription expired"
+	case http.StatusForbidden:
+		reason += ": the device is muted"
+	}
+	return "cloud.VerifyPersistence(kerberoshub): " + reason + " (HTTP " + strconv.Itoa(statusCode) + ")"
+}
+
 // VerifySecondaryPersistence godoc
 // @Router /api/persistence/secondary/verify [post]
 // @ID verify-secondary-persistence

machinery/src/cloud/cloud_test.go

@@ -1,22 +1,222 @@
 package cloud
 
 import (
+	"bytes"
 	"encoding/json"
 	"errors"
 	"fmt"
 	"io"
 	"net/http"
+	"net/http/httptest"
 	"strings"
 	"testing"
 	"time"
 
+	"github.com/gin-gonic/gin"
 	"github.com/kerberos-io/agent/machinery/src/models"
 	agentonvif "github.com/kerberos-io/agent/machinery/src/onvif"
 	goonvif "github.com/kerberos-io/onvif"
 	goonvifdevice "github.com/kerberos-io/onvif/device"
 	goonvifptz "github.com/kerberos-io/onvif/ptz"
+	goonvifxsd "github.com/kerberos-io/onvif/xsd/onvif"
 )
 
+func TestVerifyPersistenceKerberosHubUsesAuthorizationHead(t *testing.T) {
+	var requestMethod string
+	var requestBody []byte
+	var requestHeaders http.Header
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		requestMethod = r.Method
+		requestHeaders = r.Header.Clone()
+		requestBody, _ = io.ReadAll(r.Body)
+		w.WriteHeader(http.StatusOK)
+	}))
+	defer server.Close()
+
+	config := models.Config{
+		Cloud:         "kerberoshub",
+		Key:           "camera-1",
+		Name:          "front-door",
+		HubURI:        server.URL,
+		HubKey:        "project-public",
+		HubPrivateKey: "project-private",
+		S3:            &models.S3{Region: "eu"},
+	}
+	payload, err := json.Marshal(config)
+	if err != nil {
+		t.Fatalf("json.Marshal() error = %v", err)
+	}
+
+	gin.SetMode(gin.TestMode)
+	recorder := httptest.NewRecorder()
+	ctx, _ := gin.CreateTestContext(recorder)
+	ctx.Request = httptest.NewRequest(http.MethodPost, "/api/persistence/verify", bytes.NewReader(payload))
+	ctx.Request.Header.Set("Content-Type", "application/json")
+
+	VerifyPersistence(ctx, t.TempDir(), server.URL)
+
+	if recorder.Code != http.StatusOK {
+		t.Fatalf("status = %d, body = %s", recorder.Code, recorder.Body.String())
+	}
+	if requestMethod != http.MethodHead {
+		t.Fatalf("Hub request method = %q, want HEAD", requestMethod)
+	}
+	if len(requestBody) != 0 {
+		t.Fatalf("Hub request body has %d bytes, want none", len(requestBody))
+	}
+	for header, want := range map[string]string{
+		"X-Kerberos-Storage-Device":  "camera-1",
+		"X-Kerberos-Hub-PublicKey":   "project-public",
+		"X-Kerberos-Hub-PrivateKey":  "project-private",
+		"X-Kerberos-Hub-Region":      "eu",
+		"X-Kerberos-Storage-Capture": "IPCamera",
+	} {
+		if got := requestHeaders.Get(header); got != want {
+			t.Errorf("%s = %q, want %q", header, got, want)
+		}
+	}
+}
+
+func TestVerifyPersistenceKerberosHubRejectsCrossHostRedirect(t *testing.T) {
+	var redirectedHeaders http.Header
+	target := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		redirectedHeaders = r.Header.Clone()
+		w.WriteHeader(http.StatusOK)
+	}))
+	defer target.Close()
+
+	var originalHeaders http.Header
+	hub := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		originalHeaders = r.Header.Clone()
+		http.Redirect(w, r, target.URL+"/storage/upload", http.StatusTemporaryRedirect)
+	}))
+	defer hub.Close()
+
+	config := models.Config{
+		Cloud:         "kerberoshub",
+		Key:           "camera-1",
+		Name:          "front-door",
+		HubURI:        hub.URL,
+		HubKey:        "project-public",
+		HubPrivateKey: "project-private",
+		S3:            &models.S3{Region: "eu"},
+	}
+	payload, err := json.Marshal(config)
+	if err != nil {
+		t.Fatalf("json.Marshal() error = %v", err)
+	}
+
+	gin.SetMode(gin.TestMode)
+	recorder := httptest.NewRecorder()
+	ctx, _ := gin.CreateTestContext(recorder)
+	ctx.Request = httptest.NewRequest(http.MethodPost, "/api/persistence/verify", bytes.NewReader(payload))
+	ctx.Request.Header.Set("Content-Type", "application/json")
+
+	VerifyPersistence(ctx, t.TempDir(), hub.URL)
+
+	if recorder.Code != http.StatusBadRequest {
+		t.Fatalf("status = %d, want %d; body = %s", recorder.Code, http.StatusBadRequest, recorder.Body.String())
+	}
+	for _, header := range []string{"X-Kerberos-Hub-PublicKey", "X-Kerberos-Hub-PrivateKey"} {
+		if originalHeaders.Get(header) == "" {
+			t.Errorf("%s was not sent to the configured Hub", header)
+		}
+		if redirectedHeaders != nil {
+			t.Errorf("redirect target received a request with %s=%q", header, redirectedHeaders.Get(header))
+		}
+	}
+}
+
+func TestVerifyPersistenceKerberosHubRejectsUnsavedURI(t *testing.T) {
+	requestCount := 0
+	untrusted := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		requestCount++
+		w.WriteHeader(http.StatusOK)
+	}))
+	defer untrusted.Close()
+
+	config := models.Config{
+		Cloud:         "kerberoshub",
+		Key:           "camera-1",
+		Name:          "front-door",
+		HubURI:        untrusted.URL,
+		HubKey:        "project-public",
+		HubPrivateKey: "project-private",
+		S3:            &models.S3{Region: "eu"},
+	}
+	payload, err := json.Marshal(config)
+	if err != nil {
+		t.Fatalf("json.Marshal() error = %v", err)
+	}
+
+	gin.SetMode(gin.TestMode)
+	recorder := httptest.NewRecorder()
+	ctx, _ := gin.CreateTestContext(recorder)
+	ctx.Request = httptest.NewRequest(http.MethodPost, "/api/persistence/verify", bytes.NewReader(payload))
+	ctx.Request.Header.Set("Content-Type", "application/json")
+
+	VerifyPersistence(ctx, t.TempDir(), "https://saved-hub.example.com")
+
+	if recorder.Code != http.StatusBadRequest {
+		t.Fatalf("status = %d, want %d; body = %s", recorder.Code, http.StatusBadRequest, recorder.Body.String())
+	}
+	if requestCount != 0 {
+		t.Fatalf("untrusted Hub received %d requests, want none", requestCount)
+	}
+}
+
+func TestVerifyHubRejectsUnsavedURI(t *testing.T) {
+	requestCount := 0
+	untrusted := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		requestCount++
+		w.WriteHeader(http.StatusOK)
+	}))
+	defer untrusted.Close()
+
+	config := models.Config{
+		HubURI:        untrusted.URL,
+		HubKey:        "project-public",
+		HubPrivateKey: "project-private",
+	}
+	payload, err := json.Marshal(config)
+	if err != nil {
+		t.Fatalf("json.Marshal() error = %v", err)
+	}
+
+	gin.SetMode(gin.TestMode)
+	recorder := httptest.NewRecorder()
+	ctx, _ := gin.CreateTestContext(recorder)
+	ctx.Request = httptest.NewRequest(http.MethodPost, "/api/hub/verify", bytes.NewReader(payload))
+	ctx.Request.Header.Set("Content-Type", "application/json")
+
+	VerifyHub(ctx, "https://saved-hub.example.com")
+
+	if recorder.Code != http.StatusBadRequest {
+		t.Fatalf("status = %d, want %d; body = %s", recorder.Code, http.StatusBadRequest, recorder.Body.String())
+	}
+	if requestCount != 0 {
+		t.Fatalf("untrusted Hub received %d requests, want none", requestCount)
+	}
+}
+
+func TestHubPersistenceVerificationErrorExplainsStatus(t *testing.T) {
+	for _, test := range []struct {
+		status int
+		want   string
+	}{
+		{status: http.StatusBadRequest, want: "credentials or device/project assignment"},
+		{status: http.StatusUnauthorized, want: "no active subscription"},
+		{status: http.StatusForbidden, want: "device is muted"},
+		{status: http.StatusBadGateway, want: "HTTP 502"},
+	} {
+		t.Run(http.StatusText(test.status), func(t *testing.T) {
+			if got := hubPersistenceVerificationError(test.status); !strings.Contains(got, test.want) {
+				t.Fatalf("hubPersistenceVerificationError(%d) = %q, want it to contain %q", test.status, got, test.want)
+			}
+		})
+	}
+}
+
 func TestHeartbeatFailureLogOmitsHubResponseBody(t *testing.T) {
 	response := &http.Response{
 		StatusCode: http.StatusBadRequest,
@@ -590,3 +790,60 @@ func countString(values []string, want string) int {
 	}
 	return count
 }
+
+// Some devices accept the subscription but reject the pulls, or never publish
+// the initial I/O state. The relay outputs and digital inputs must then be
+// queried directly instead of reporting no I/O at all.
+func TestHeartbeatONVIFPayloadFallsBackToDirectIOWhenSubscriptionsReturnNoEvents(t *testing.T) {
+	restoreHeartbeatONVIFStubs(t)
+
+	device := newTestONVIFDevice()
+	camera := models.IPCamera{
+		ONVIFXAddr:    "http://camera/onvif",
+		ONVIFUsername: "operator",
+		ONVIFPassword: "secret",
+	}
+
+	state := newHeartbeatONVIFState()
+	state.cameraConfiguration = camera
+	state.cameraKey = heartbeatONVIFCameraKey(camera)
+	state.device = device
+	state.staticLoaded = true
+	state.staticPayload.enabled = "true"
+
+	heartbeatCreatePullPointSubscription = func(*goonvif.Device) (string, error) {
+		return "pull-point", nil
+	}
+	heartbeatGetEventMessages = func(*goonvif.Device, string) ([]agentonvif.ONVIFEvents, error) {
+		return nil, errors.New("pull rejected")
+	}
+	heartbeatUnsubscribePullPoint = func(*goonvif.Device, string) error {
+		return nil
+	}
+	var relayCalls, inputCalls int
+	heartbeatGetRelayOutputs = func(*goonvif.Device) (goonvifdevice.GetRelayOutputsResponse, error) {
+		relayCalls++
+		return goonvifdevice.GetRelayOutputsResponse{RelayOutputs: []goonvifxsd.RelayOutput{{DeviceEntity: goonvifxsd.DeviceEntity{Token: "relay0"}}}}, nil
+	}
+	heartbeatGetDigitalInputs = func(*goonvif.Device) (goonvifdevice.GetDigitalInputsResponse, error) {
+		inputCalls++
+		return goonvifdevice.GetDigitalInputsResponse{DigitalInputs: []goonvifxsd.DigitalInput{{Token: "di0"}}}, nil
+	}
+
+	payload := getHeartbeatONVIFPayload(camera, state)
+
+	if relayCalls != 1 || inputCalls != 1 {
+		t.Fatalf("direct I/O queries: relay=%d inputs=%d, want 1 each", relayCalls, inputCalls)
+	}
+	var events []agentonvif.ONVIFEvents
+	if err := json.Unmarshal(payload.eventsList, &events); err != nil {
+		t.Fatalf("payload.eventsList is not valid JSON: %v", err)
+	}
+	got := map[string]string{}
+	for _, event := range events {
+		got[event.Key] = event.Type
+	}
+	if got["relay0"] != "output" || got["di0"] != "input" || len(got) != 2 {
+		t.Fatalf("payload.eventsList = %s, want relay0 output and di0 input", payload.eventsList)
+	}
+}

machinery/src/cloud/custom_headers.go

@@ -0,0 +1,195 @@
+package cloud
+
+import (
+	"encoding/base64"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"net/http"
+	"sort"
+	"strings"
+)
+
+const (
+	customHeadersManifestHeader = "X-Kerberos-Storage-Custom-Headers"
+	maxCustomHeaders            = 32
+	maxCustomHeaderNameLength   = 128
+	maxCustomHeaderValueLength  = 4096
+)
+
+var disallowedCustomHeaders = map[string]struct{}{
+	"authorization":       {},
+	"connection":          {},
+	"content-length":      {},
+	"content-type":        {},
+	"cookie":              {},
+	"expect":              {},
+	"host":                {},
+	"keep-alive":          {},
+	"proxy-authorization": {},
+	"proxy-connection":    {},
+	"te":                  {},
+	"trailer":             {},
+	"trace-id":            {},
+	"transfer-encoding":   {},
+	"tus-resumable":       {},
+	"upgrade":             {},
+}
+
+func parseCustomVaultHeaders(value string) (map[string]string, error) {
+	if strings.TrimSpace(value) == "" {
+		return nil, nil
+	}
+
+	// Decode into pointers so JSON null values can be told apart from strings.
+	var rawHeaders map[string]*string
+	if err := json.Unmarshal([]byte(value), &rawHeaders); err != nil {
+		return nil, fmt.Errorf("custom Vault headers must be a JSON object of string values: %w", err)
+	}
+	if rawHeaders == nil {
+		return nil, errors.New("custom Vault headers must be a JSON object of string values")
+	}
+	if len(rawHeaders) > maxCustomHeaders {
+		return nil, fmt.Errorf("custom Vault headers exceed the limit of %d", maxCustomHeaders)
+	}
+	headers := make(map[string]string, len(rawHeaders))
+	seen := make(map[string]struct{}, len(rawHeaders))
+	for name, rawValue := range rawHeaders {
+		if rawValue == nil {
+			return nil, fmt.Errorf("custom Vault header %q must be a string, not null", name)
+		}
+		value := *rawValue
+		headers[name] = value
+		if err := validateCustomVaultHeader(name, value); err != nil {
+			return nil, err
+		}
+		lowerName := strings.ToLower(name)
+		if _, duplicate := seen[lowerName]; duplicate {
+			return nil, fmt.Errorf("custom Vault header %q is duplicated with different casing", name)
+		}
+		seen[lowerName] = struct{}{}
+	}
+	return headers, nil
+}
+
+func validateCustomVaultHeader(name, value string) error {
+	if name == "" || len(name) > maxCustomHeaderNameLength || !isHTTPToken(name) {
+		return fmt.Errorf("invalid custom Vault header name %q", name)
+	}
+	lowerName := strings.ToLower(name)
+	if strings.HasPrefix(lowerName, "x-kerberos-") {
+		return fmt.Errorf("custom Vault header %q uses the reserved X-Kerberos namespace", name)
+	}
+	if strings.HasPrefix(lowerName, "upload-") {
+		return fmt.Errorf("custom Vault header %q uses the reserved tus Upload namespace", name)
+	}
+	if _, disallowed := disallowedCustomHeaders[lowerName]; disallowed {
+		return fmt.Errorf("custom Vault header %q is reserved", name)
+	}
+	if len(value) > maxCustomHeaderValueLength {
+		return fmt.Errorf("custom Vault header %q exceeds the %d byte value limit", name, maxCustomHeaderValueLength)
+	}
+	for _, character := range value {
+		if character < 0x20 || character == 0x7f {
+			return fmt.Errorf("custom Vault header %q contains an invalid value", name)
+		}
+	}
+	return nil
+}
+
+func isHTTPToken(value string) bool {
+	if value == "" {
+		return false
+	}
+	for _, character := range value {
+		if (character >= 'a' && character <= 'z') ||
+			(character >= 'A' && character <= 'Z') ||
+			(character >= '0' && character <= '9') ||
+			strings.ContainsRune("!#$%&'*+-.^_`|~", character) {
+			continue
+		}
+		return false
+	}
+	return true
+}
+
+func setCustomVaultHeaders(headers http.Header, customHeaders map[string]string) error {
+	if len(customHeaders) == 0 {
+		return nil
+	}
+
+	names := make([]string, 0, len(customHeaders))
+	for name, value := range customHeaders {
+		if err := validateCustomVaultHeader(name, value); err != nil {
+			return err
+		}
+		names = append(names, name)
+		// Assign directly so the configured name is not canonicalised by Set.
+		headers[name] = []string{value}
+	}
+	sort.Strings(names)
+	manifest, err := json.Marshal(names)
+	if err != nil {
+		return err
+	}
+	headers.Set(customHeadersManifestHeader, base64.RawURLEncoding.EncodeToString(manifest))
+	return nil
+}
+
+const maxVaultRedirects = 10
+
+// stripVaultHeadersOnCrossHostRedirect removes Vault credentials and the
+// customer-supplied metadata (custom headers, their manifest and the tus
+// Upload-Metadata that embeds them) before following a redirect to another host.
+func stripVaultHeadersOnCrossHostRedirect(req *http.Request, via []*http.Request) error {
+	if len(via) >= maxVaultRedirects {
+		return fmt.Errorf("stopped after %d redirects", maxVaultRedirects)
+	}
+	if len(via) == 0 || req.URL.Host == via[0].URL.Host {
+		return nil
+	}
+
+	customNames := customVaultHeaderNamesFromManifest(req.Header)
+	for name := range req.Header {
+		canonicalName := http.CanonicalHeaderKey(name)
+		_, custom := customNames[strings.ToLower(name)]
+		if custom || strings.HasPrefix(canonicalName, "X-Kerberos-") || canonicalName == "Upload-Metadata" {
+			delete(req.Header, name)
+		}
+	}
+	return nil
+}
+
+func customVaultHeaderNamesFromManifest(headers http.Header) map[string]struct{} {
+	encoded := headers.Get(customHeadersManifestHeader)
+	if encoded == "" {
+		return nil
+	}
+	manifest, err := base64.RawURLEncoding.DecodeString(encoded)
+	if err != nil {
+		return nil
+	}
+	var names []string
+	if err := json.Unmarshal(manifest, &names); err != nil {
+		return nil
+	}
+	lowerNames := make(map[string]struct{}, len(names))
+	for _, name := range names {
+		lowerNames[strings.ToLower(name)] = struct{}{}
+	}
+	return lowerNames
+}
+
+func encodeCustomVaultHeaders(customHeaders map[string]string) (string, error) {
+	if len(customHeaders) == 0 {
+		return "", nil
+	}
+	encoded, err := json.Marshal(customHeaders)
+	if err != nil {
+		return "", err
+	}
+	if len(encoded) == 0 {
+		return "", errors.New("custom Vault headers could not be encoded")
+	}
+	return string(encoded), nil
+}

machinery/src/cloud/custom_headers_test.go

@@ -0,0 +1,100 @@
+package cloud
+
+import (
+	"encoding/base64"
+	"encoding/json"
+	"net/http"
+	"reflect"
+	"testing"
+)
+
+func TestCustomVaultHeadersAreValidatedAndAdded(t *testing.T) {
+	customHeaders, err := parseCustomVaultHeaders(`{"site_id":"site-1","line_id":"line-2"}`)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	headers := http.Header{}
+	if err := setCustomVaultHeaders(headers, customHeaders); err != nil {
+		t.Fatal(err)
+	}
+	if !reflect.DeepEqual(headers["site_id"], []string{"site-1"}) || !reflect.DeepEqual(headers["line_id"], []string{"line-2"}) {
+		t.Fatalf("custom header names were not preserved: %#v", headers)
+	}
+	if _, canonicalised := headers["Site_id"]; canonicalised {
+		t.Fatalf("custom header name was canonicalised: %#v", headers)
+	}
+
+	encodedManifest, err := base64.RawURLEncoding.DecodeString(headers.Get(customHeadersManifestHeader))
+	if err != nil {
+		t.Fatal(err)
+	}
+	var names []string
+	if err := json.Unmarshal(encodedManifest, &names); err != nil {
+		t.Fatal(err)
+	}
+	if !reflect.DeepEqual(names, []string{"line_id", "site_id"}) {
+		t.Fatalf("manifest = %#v", names)
+	}
+}
+
+func TestCustomVaultHeadersRejectReservedAndInvalidValues(t *testing.T) {
+	for _, value := range []string{
+		`{"X-Kerberos-Storage-AccessKey":"secret"}`,
+		`{"Authorization":"secret"}`,
+		`{"site_id":"one\nInjected: value"}`,
+		`{"bad header":"value"}`,
+		`{"site_id":1}`,
+		`{"Content-Type":"text/plain"}`,
+		`{"site_id":"one","SITE_ID":"two"}`,
+		`{"site_id":null}`,
+		`{"site_id":"one\ttwo"}`,
+		`{"Upload-Checksum":"sha1 abc"}`,
+		`{"upload-concat":"final;/files/a"}`,
+		`{"Upload-Defer-Length":"1"}`,
+		`null`,
+	} {
+		if _, err := parseCustomVaultHeaders(value); err == nil {
+			t.Fatalf("parseCustomVaultHeaders(%q) succeeded", value)
+		}
+	}
+}
+
+func TestVaultRedirectStripsCustomHeadersOnHostChange(t *testing.T) {
+	newRedirect := func(t *testing.T, host string) (*http.Request, []*http.Request) {
+		t.Helper()
+		original, err := http.NewRequest(http.MethodPost, "https://vault.example.com/api/storage", nil)
+		if err != nil {
+			t.Fatal(err)
+		}
+		redirected, err := http.NewRequest(http.MethodPost, "https://"+host+"/api/storage", nil)
+		if err != nil {
+			t.Fatal(err)
+		}
+		redirected.Header.Set("X-Kerberos-Storage-AccessKey", "secret")
+		redirected.Header.Set("Upload-Metadata", "custom_headers abc")
+		redirected.Header.Set("Tus-Resumable", "1.0.0")
+		if err := setCustomVaultHeaders(redirected.Header, map[string]string{"site_id": "site-1", "Line-Id": "line-2"}); err != nil {
+			t.Fatal(err)
+		}
+		return redirected, []*http.Request{original}
+	}
+
+	crossHost, via := newRedirect(t, "other.example.com")
+	if err := stripVaultHeadersOnCrossHostRedirect(crossHost, via); err != nil {
+		t.Fatal(err)
+	}
+	if len(crossHost.Header) != 1 || crossHost.Header.Get("Tus-Resumable") != "1.0.0" {
+		t.Fatalf("cross-host redirect headers = %#v", crossHost.Header)
+	}
+
+	sameHost, via := newRedirect(t, "vault.example.com")
+	if err := stripVaultHeadersOnCrossHostRedirect(sameHost, via); err != nil {
+		t.Fatal(err)
+	}
+	if !reflect.DeepEqual(sameHost.Header["site_id"], []string{"site-1"}) ||
+		sameHost.Header.Get("X-Kerberos-Storage-AccessKey") != "secret" ||
+		sameHost.Header.Get(customHeadersManifestHeader) == "" {
+		t.Fatalf("same-host redirect headers = %#v", sameHost.Header)
+	}
+}

machinery/src/cloud/kerberos_vault.go

@@ -178,6 +178,10 @@ func sendToVault(vault models.KStorage, publicKey, deviceKey, fileName, label, s
 // 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) {
+	customHeaders, err := parseCustomVaultHeaders(vault.CustomHeaders)
+	if err != nil {
+		return false, false, "", err
+	}
 	fullname := "data/recordings/" + fileName
 
 	file, err := os.Open(fullname)
@@ -203,6 +207,9 @@ func uploadVaultLegacy(vault models.KStorage, publicKey, deviceKey, fileName, la
 	}
 	req.Header.Set("Content-Type", "video/mp4")
 	setVaultHeaders(req.Header, vault, publicKey, deviceKey, fileName)
+	if err := setCustomVaultHeaders(req.Header, customHeaders); err != nil {
+		return false, false, "", err
+	}
 	setQueuedRecordingMetadataHeaders(req.Header, fileName)
 
 	client := newVaultHTTPClient(0)
@@ -267,7 +274,7 @@ func newVaultHTTPClient(timeout time.Duration) *http.Client {
 		transport.TLSClientConfig.InsecureSkipVerify = true
 	}
 
-	client := &http.Client{Transport: transport}
+	client := &http.Client{Transport: transport, CheckRedirect: stripVaultHeadersOnCrossHostRedirect}
 	if timeout > 0 {
 		client.Timeout = timeout
 	}

machinery/src/cloud/tus_client.go

@@ -141,19 +141,6 @@ func runTusUpload(baseURL, metadata, fileName, label, slot string, setHeaders tu
 	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)
@@ -311,6 +298,10 @@ func runTusUpload(baseURL, metadata, fileName, label, slot string, setHeaders tu
 // 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) {
+	customHeaders, err := parseCustomVaultHeaders(vault.CustomHeaders)
+	if err != nil {
+		return false, false, true, "", err
+	}
 	baseURL := strings.TrimRight(vault.URI, "/") + tusUploadPath
 	metadataValues := map[string]string{
 		"filename":  fileName,
@@ -321,10 +312,22 @@ func uploadVaultResumable(vault models.KStorage, publicKey, deviceKey, fileName,
 		"cloudkey":  publicKey,
 		"fps":       queuedRecordingFPS(fileName),
 	}
+	if encodedHeaders, err := encodeCustomVaultHeaders(customHeaders); err != nil {
+		return false, false, true, "", err
+	} else if encodedHeaders != "" {
+		metadataValues["custom_headers"] = encodedHeaders
+	}
+	customRequestHeaders := http.Header{}
+	if err := setCustomVaultHeaders(customRequestHeaders, customHeaders); err != nil {
+		return false, false, true, "", err
+	}
 	addRecordingTusMetadata(metadataValues, fileName)
 	metadata := encodeTusMetadata(metadataValues)
 	setHeaders := func(h http.Header, fn string) {
 		setVaultTusHeaders(h, vault, publicKey, deviceKey, fn)
+		for name, values := range customRequestHeaders {
+			h[name] = append([]string(nil), values...)
+		}
 	}
 	return runTusUpload(baseURL, metadata, fileName, label, slot, setHeaders)
 }

machinery/src/cloud/tus_client_test.go

@@ -3,6 +3,7 @@ package cloud
 import (
 	"bytes"
 	"encoding/base64"
+	"encoding/json"
 	"fmt"
 	"io"
 	"net"
@@ -304,7 +305,9 @@ func TestUploadVaultResumable_HappyPath(t *testing.T) {
 	withRecording(t, fileName, payload)
 	withQueuedRecordingFPS(t, fileName, `{"filename":"recording.mp4","device_key":"device-key","timestamp":1785934709414,"duration":20452,"fps":29.97}`)
 
-	uploaded, responded, supported, _, err := uploadVaultResumable(testVault(ts.URL), "pk", "dev", fileName, "test", "primary")
+	vault := testVault(ts.URL)
+	vault.CustomHeaders = `{"site_id":"site-1","equipment_id":"equipment-3"}`
+	uploaded, responded, supported, _, err := uploadVaultResumable(vault, "pk", "dev", fileName, "test", "primary")
 	if err != nil {
 		t.Fatalf("unexpected error: %v", err)
 	}
@@ -328,6 +331,23 @@ func TestUploadVaultResumable_HappyPath(t *testing.T) {
 	if got := metadata["timestamp"]; got != "1785934709414" {
 		t.Fatalf("POST metadata timestamp = %q, want %q", got, "1785934709414")
 	}
+	var customHeaders map[string]string
+	if err := json.Unmarshal([]byte(metadata["custom_headers"]), &customHeaders); err != nil {
+		t.Fatal(err)
+	}
+	if customHeaders["site_id"] != "site-1" || customHeaders["equipment_id"] != "equipment-3" {
+		t.Fatalf("POST custom header metadata = %#v", customHeaders)
+	}
+	for _, method := range []string{http.MethodPost, http.MethodHead, http.MethodPatch} {
+		for _, request := range srv.requestsForMethod(method) {
+			if request.header.Get("site_id") != "site-1" || request.header.Get("equipment_id") != "equipment-3" {
+				t.Fatalf("%s custom headers = %#v", method, request.header)
+			}
+			if request.header.Get(customHeadersManifestHeader) == "" {
+				t.Fatalf("%s is missing the custom header manifest", method)
+			}
+		}
+	}
 }
 
 func TestQueuedRecordingFPSValidation(t *testing.T) {

machinery/src/config/main.go

@@ -612,6 +612,9 @@ func applyAgentEnvVars(configuration *models.Configuration, prefix string, apply
 			case "AGENT_KERBEROSVAULT_DIRECTORY":
 				configuration.Config.KStorage.Directory = value
 				break
+			case "AGENT_KERBEROSVAULT_CUSTOM_HEADERS":
+				configuration.Config.KStorage.CustomHeaders = value
+				break
 
 			/* Retry policy and timeout */
 			case "AGENT_KERBEROSVAULT_MAX_RETRIES":
@@ -643,6 +646,9 @@ func applyAgentEnvVars(configuration *models.Configuration, prefix string, apply
 			case "AGENT_KERBEROSVAULT_SECONDARY_DIRECTORY":
 				configuration.Config.KStorageSecondary.Directory = value
 				break
+			case "AGENT_KERBEROSVAULT_SECONDARY_CUSTOM_HEADERS":
+				configuration.Config.KStorageSecondary.CustomHeaders = value
+				break
 
 			/* When storing in dropbox */
 			case "AGENT_DROPBOX_ACCESS_TOKEN":

machinery/src/config/main_test.go

@@ -88,6 +88,22 @@ func TestApplyAgentEnvVarsPixelChangeThresholdDefault(t *testing.T) {
 	}
 }
 
+func TestApplyAgentEnvVarsVaultCustomHeaders(t *testing.T) {
+	t.Setenv("AGENT_KERBEROSVAULT_CUSTOM_HEADERS", `{"site_id":"site-1"}`)
+	t.Setenv("AGENT_KERBEROSVAULT_SECONDARY_CUSTOM_HEADERS", `{"site_id":"site-2"}`)
+	configuration := &models.Configuration{}
+	initConfigPointers(&configuration.Config)
+
+	applyAgentEnvVars(configuration, "", false)
+
+	if got := configuration.Config.KStorage.CustomHeaders; got != `{"site_id":"site-1"}` {
+		t.Fatalf("primary custom headers = %q", got)
+	}
+	if got := configuration.Config.KStorageSecondary.CustomHeaders; got != `{"site_id":"site-2"}` {
+		t.Fatalf("secondary custom headers = %q", got)
+	}
+}
+
 func intPointer(value int) *int {
 	return &value
 }

machinery/src/models/config.go

@@ -187,6 +187,7 @@ type KStorage struct {
 	SecretAccessKey string `json:"secret_access_key,omitempty" bson:"secret_access_key,omitempty"`
 	Provider        string `json:"provider,omitempty" bson:"provider,omitempty"`
 	Directory       string `json:"directory,omitempty" bson:"directory,omitempty"`
+	CustomHeaders   string `json:"custom_headers,omitempty" bson:"custom_headers,omitempty"`
 	MaxRetries      int    `json:"max_retries,omitempty" bson:"max_retries,omitempty"`
 	Timeout         int    `json:"timeout,omitempty" bson:"timeout,omitempty"`
 }

machinery/src/onvif/main.go

@@ -5,7 +5,9 @@ import (
 	"encoding/json"
 	"encoding/xml"
 	"errors"
+	"fmt"
 	"io"
+	"math"
 	"net/http"
 	"strconv"
 	"strings"
@@ -130,7 +132,7 @@ func HandleONVIFActions(configuration *models.Configuration, communication *mode
 
 					if onvifAction.Action == "absolute-move" {
 
-						// We will move the camera to zero position.
+						// Move the camera to the saved absolute position.
 						x := ptzAction.X
 						y := ptzAction.Y
 						z := ptzAction.Z
@@ -141,42 +143,13 @@ func HandleONVIFActions(configuration *models.Configuration, communication *mode
 						// Log functions
 						log.Debug("onvif.HandleONVIFActions(): functions: " + strings.Join(functions, ", "))
 
-						// Check if we need to use absolute or continuous move
-						/*canAbsoluteMove := false
-						canContinuousMove := false
-
-						if len(functions) > 0 {
-							for _, function := range functions {
-								if function == "AbsolutePanTiltMove" || function == "AbsoluteZoomMove" {
-									canAbsoluteMove = true
-								} else if function == "ContinuousPanTiltMove" || function == "ContinuousZoomMove" {
-									canContinuousMove = true
-								}
-							}
-						}*/
-
-						// Ideally we should be able to use the AbsolutePanTiltMove function, but it looks like
-						// the current detection through GetPTZFuntionsFromDevice is not working properly. Therefore we will fallback
-						// on the ContinuousPanTiltMove function which is more compatible with more cameras.
-						err = AbsolutePanTiltMoveFake(device, configurations, token, x, y, z)
+						err = MoveToPosition(device, configurations, token, x, y, z)
 						if err != nil {
-							log.Debug("onvif.HandleONVIFActions() - AbsolutePanTitleMoveFake: " + err.Error())
+							log.Error("onvif.HandleONVIFActions() - MoveToPosition: " + err.Error())
 						} else {
-							log.Info("onvif.HandleONVIFActions() - AbsolutePanTitleMoveFake: successfully moved camera.")
+							log.Info("onvif.HandleONVIFActions() - MoveToPosition: successfully moved camera.")
 						}
 
-						/*if canAbsoluteMove {
-							err = AbsolutePanTiltMove(device, configurations, token, x, y, z)
-							if err != nil {
-								log.Error("HandleONVIFActions (AbsolutePanTitleMove): " + err.Error())
-							}
-						} else if canContinuousMove {
-							err = AbsolutePanTiltMoveFake(device, configurations, token, x, y, z)
-							if err != nil {
-								log.Error("HandleONVIFActions (AbsolutePanTitleMoveFake): " + err.Error())
-							}
-						}*/
-
 					} else if onvifAction.Action == "preset" {
 
 						// Execute the preset
@@ -471,16 +444,20 @@ func handleONVIFResponse(operation string, response *http.Response, requestErr e
 }
 
 func AbsolutePanTiltMove(device *onvif.Device, configuration ptz.GetConfigurationsResponse, token xsdonvif.ReferenceToken, pan float64, tilt float64, zoom float64) error {
+	if len(configuration.PTZConfiguration) == 0 {
+		return errors.New("camera returned no PTZ configuration")
+	}
+	ptzConfiguration := configuration.PTZConfiguration[0]
 
 	absolutePantiltVector := xsdonvif.Vector2D{
 		X:     pan,
 		Y:     tilt,
-		Space: configuration.PTZConfiguration[0].DefaultAbsolutePantTiltPositionSpace,
+		Space: ptzConfiguration.DefaultAbsolutePantTiltPositionSpace,
 	}
 
 	absoluteZoomVector := xsdonvif.Vector1D{
 		X:     zoom,
-		Space: configuration.PTZConfiguration[0].DefaultAbsoluteZoomPositionSpace,
+		Space: ptzConfiguration.DefaultAbsoluteZoomPositionSpace,
 	}
 
 	resp, err := device.CallMethod(ptz.AbsoluteMove{
@@ -493,240 +470,325 @@ func AbsolutePanTiltMove(device *onvif.Device, configuration ptz.GetConfiguratio
 	return handleONVIFResponse("absolute_move", resp, err)
 }
 
-// This function will simulate the AbsolutePanTiltMove function.
-// However the AboslutePanTiltMove function is not working on all cameras.
-// So we'll use the ContinuousMove function to simulate the AbsolutePanTiltMove function using the position polling.
-func AbsolutePanTiltMoveFake(device *onvif.Device, configuration ptz.GetConfigurationsResponse, token xsdonvif.ReferenceToken, pan float64, tilt float64, zoom float64) error {
-	position, err := GetPosition(device, token)
-	if position.PanTilt.X >= pan-0.01 && position.PanTilt.X <= pan+0.01 && position.PanTilt.Y >= tilt-0.01 && position.PanTilt.Y <= tilt+0.01 && position.Zoom.X >= zoom-0.01 && position.Zoom.X <= zoom+0.01 {
-		log.Debug("AbsolutePanTiltMoveFake: already at position")
-	} else {
-
-		// The speed of panning, the higher the faster we'll pan the camera
-		// value is a range between 0 and 1.
-		speed := 0.6
-		wait := 100 * time.Millisecond
-
-		// We'll move quickly to the position (might be inaccurate)
-		err = ZoomOutCompletely(device, configuration, token)
-		err = PanUntilPosition(device, configuration, token, pan, zoom, speed, wait)
-		err = TiltUntilPosition(device, configuration, token, tilt, zoom, speed, wait)
-
-		// Now we'll move a bit slower to make sure we are ok (will be more accurate)
-		speed = 0.1
-		wait = 200 * time.Millisecond
-
-		err = PanUntilPosition(device, configuration, token, pan, zoom, speed, wait)
-		err = TiltUntilPosition(device, configuration, token, tilt, zoom, speed, wait)
-		err = ZoomUntilPosition(device, configuration, token, zoom, speed, wait)
-
-		return err
-	}
-	return err
+type ptzMoveClient interface {
+	GetPosition() (xsdonvif.PTZVector, error)
+	AbsoluteMove(ptzTarget) error
+	ContinuousMove(ptzVelocity) error
+	Stop() error
 }
 
-func ZoomOutCompletely(device *onvif.Device, configuration ptz.GetConfigurationsResponse, token xsdonvif.ReferenceToken) error {
-	// Zoom out completely!!!
-	zoomOut := xsdonvif.Vector1D{
-		X:     -1,
-		Space: configuration.PTZConfiguration[0].DefaultContinuousZoomVelocitySpace,
+type ptzTarget struct {
+	Pan  float64
+	Tilt float64
+	Zoom float64
+}
+
+type ptzVelocity struct {
+	Pan  float64
+	Tilt float64
+	Zoom float64
+}
+
+type ptzMoveSupport struct {
+	Absolute   bool
+	Continuous bool
+}
+
+type ptzMoveOptions struct {
+	PanTolerance      float64
+	TiltTolerance     float64
+	ZoomTolerance     float64
+	NativeTimeout     time.Duration
+	ContinuousTimeout time.Duration
+	NoProgressTimeout time.Duration
+	PollInterval      time.Duration
+	ProgressEpsilon   float64
+	VelocityGain      float64
+	MinVelocity       float64
+	MaxVelocity       float64
+	VelocityChange    float64
+}
+
+var defaultPTZMoveOptions = ptzMoveOptions{
+	PanTolerance:      0.01,
+	TiltTolerance:     0.005,
+	ZoomTolerance:     0.005,
+	NativeTimeout:     8 * time.Second,
+	ContinuousTimeout: 15 * time.Second,
+	NoProgressTimeout: 3 * time.Second,
+	PollInterval:      150 * time.Millisecond,
+	ProgressEpsilon:   0.001,
+	VelocityGain:      1.5,
+	MinVelocity:       0.08,
+	MaxVelocity:       0.7,
+	VelocityChange:    0.05,
+}
+
+type devicePTZMoveClient struct {
+	device        *onvif.Device
+	configuration xsdonvif.PTZConfiguration
+	token         xsdonvif.ReferenceToken
+}
+
+func (client *devicePTZMoveClient) GetPosition() (xsdonvif.PTZVector, error) {
+	return GetPosition(client.device, client.token)
+}
+
+func (client *devicePTZMoveClient) AbsoluteMove(target ptzTarget) error {
+	configuration := ptz.GetConfigurationsResponse{
+		PTZConfiguration: []xsdonvif.PTZConfiguration{client.configuration},
 	}
-	_, err := device.CallMethod(ptz.ContinuousMove{
-		ProfileToken: &token,
-		Velocity: xsdonvif.PTZSpeedZoom{
-			Zoom: zoomOut,
+	return AbsolutePanTiltMove(client.device, configuration, client.token, target.Pan, target.Tilt, target.Zoom)
+}
+
+func (client *devicePTZMoveClient) ContinuousMove(velocity ptzVelocity) error {
+	panTilt := xsdonvif.Vector2D{
+		X:     velocity.Pan,
+		Y:     velocity.Tilt,
+		Space: client.configuration.DefaultContinuousPanTiltVelocitySpace,
+	}
+	zoom := xsdonvif.Vector1D{
+		X:     velocity.Zoom,
+		Space: client.configuration.DefaultContinuousZoomVelocitySpace,
+	}
+	resp, err := client.device.CallMethod(ptz.ContinuousMove{
+		ProfileToken: &client.token,
+		Velocity: ptz.Speed{
+			PanTilt: &panTilt,
+			Zoom:    &zoom,
 		},
 	})
-	if err != nil {
-		log.Error("ZoomOutCompletely: " + err.Error())
-	}
+	return handleONVIFResponse("continuous_move_to_position", resp, err)
+}
 
-	for {
-		position, _ := GetPosition(device, token)
-		if position.Zoom.X == 0 {
-			break
-		}
-		time.Sleep(250 * time.Millisecond)
-	}
-
-	_, err = device.CallMethod(ptz.Stop{
-		ProfileToken: token,
+func (client *devicePTZMoveClient) Stop() error {
+	resp, err := client.device.CallMethod(ptz.Stop{
+		ProfileToken: client.token,
+		PanTilt:      true,
 		Zoom:         true,
 	})
+	return handleONVIFResponse("move_to_position_stop", resp, err)
+}
+
+func MoveToPosition(device *onvif.Device, configuration ptz.GetConfigurationsResponse, token xsdonvif.ReferenceToken, pan float64, tilt float64, zoom float64) error {
+	if len(configuration.PTZConfiguration) == 0 {
+		return errors.New("camera returned no PTZ configuration")
+	}
+	ptzConfiguration := configuration.PTZConfiguration[0]
+	client := &devicePTZMoveClient{
+		device:        device,
+		configuration: ptzConfiguration,
+		token:         token,
+	}
+	support := ptzMoveSupport{
+		Absolute: ptzConfiguration.DefaultAbsolutePantTiltPositionSpace != nil &&
+			ptzConfiguration.DefaultAbsoluteZoomPositionSpace != nil,
+		Continuous: ptzConfiguration.DefaultContinuousPanTiltVelocitySpace != nil &&
+			ptzConfiguration.DefaultContinuousZoomVelocitySpace != nil,
+	}
+	return moveToPosition(client, support, ptzTarget{Pan: pan, Tilt: tilt, Zoom: zoom}, defaultPTZMoveOptions)
+}
+
+// AbsolutePanTiltMoveFake preserves the continuous-move compatibility path for callers that explicitly request it.
+func AbsolutePanTiltMoveFake(device *onvif.Device, configuration ptz.GetConfigurationsResponse, token xsdonvif.ReferenceToken, pan float64, tilt float64, zoom float64) error {
+	if len(configuration.PTZConfiguration) == 0 {
+		return errors.New("camera returned no PTZ configuration")
+	}
+	ptzConfiguration := configuration.PTZConfiguration[0]
+	client := &devicePTZMoveClient{
+		device:        device,
+		configuration: ptzConfiguration,
+		token:         token,
+	}
+	support := ptzMoveSupport{
+		Continuous: ptzConfiguration.DefaultContinuousPanTiltVelocitySpace != nil &&
+			ptzConfiguration.DefaultContinuousZoomVelocitySpace != nil,
+	}
+	return moveToPosition(client, support, ptzTarget{Pan: pan, Tilt: tilt, Zoom: zoom}, defaultPTZMoveOptions)
+}
+
+func moveToPosition(client ptzMoveClient, support ptzMoveSupport, target ptzTarget, options ptzMoveOptions) error {
+	if err := validatePTZTarget(target); err != nil {
+		return err
+	}
+
+	position, err := client.GetPosition()
 	if err != nil {
-		log.Error("ZoomOutCompletely: " + err.Error())
+		return fmt.Errorf("read initial PTZ position: %w", err)
 	}
-	return err
+	_, _, reached, err := ptzTargetError(position, target, options)
+	if err != nil {
+		return err
+	}
+	if reached {
+		return nil
+	}
+
+	var nativeErr error
+	if support.Absolute {
+		if err := client.AbsoluteMove(target); err != nil {
+			nativeErr = fmt.Errorf("native absolute move: %w", err)
+		} else if err := waitForPTZTarget(client, target, options.NativeTimeout, options); err != nil {
+			nativeErr = fmt.Errorf("verify native absolute move: %w", err)
+		} else {
+			return nil
+		}
+
+		if err := client.Stop(); err != nil {
+			nativeErr = errors.Join(nativeErr, fmt.Errorf("stop native absolute move: %w", err))
+		}
+		log.WithError(nativeErr).Warn("Native ONVIF absolute move failed; using continuous-move fallback")
+	}
+
+	if !support.Continuous {
+		if nativeErr != nil {
+			return errors.Join(nativeErr, errors.New("camera does not advertise complete continuous PTZ support"))
+		}
+		return errors.New("camera does not advertise a supported absolute or continuous PTZ movement method")
+	}
+
+	if err := moveContinuouslyToPTZTarget(client, target, options); err != nil {
+		if nativeErr != nil {
+			return errors.Join(nativeErr, fmt.Errorf("continuous-move fallback: %w", err))
+		}
+		return fmt.Errorf("continuous move: %w", err)
+	}
+	return nil
 }
 
-func PanUntilPosition(device *onvif.Device, configuration ptz.GetConfigurationsResponse, token xsdonvif.ReferenceToken, pan float64, zoom float64, speed float64, wait time.Duration) error {
-	position, err := GetPosition(device, token)
-
-	if position.PanTilt.X >= pan-0.01 && position.PanTilt.X <= pan+0.01 {
-
-	} else {
-
-		// We'll need to determine if we need to move CW or CCW.
-		// Check the current position and compare it with the desired position.
-		directionX := speed
-		if position.PanTilt.X > pan {
-			directionX = speed * -1
-		}
-
-		panTiltVector := xsdonvif.Vector2D{
-			X:     directionX,
-			Y:     0,
-			Space: configuration.PTZConfiguration[0].DefaultContinuousPanTiltVelocitySpace,
-		}
-		resp, err := device.CallMethod(ptz.ContinuousMove{
-			ProfileToken: &token,
-			Velocity: xsdonvif.PTZSpeedPanTilt{
-				PanTilt: panTiltVector,
-			},
-		})
-		err = handleONVIFResponse("continuous_pan", resp, err)
-
-		// While moving we'll check if we reached the desired position.
-		// or if we overshot the desired position.
-
-		// Break after 3seconds
-		now := time.Now()
-		for {
-			position, _ := GetPosition(device, token)
-			if position.PanTilt.X == -1 || position.PanTilt.X == 1 || (directionX > 0 && position.PanTilt.X >= pan) || (directionX < 0 && position.PanTilt.X <= pan) || (position.PanTilt.X >= pan-0.01 && position.PanTilt.X <= pan+0.01) {
-				break
-			}
-			if time.Since(now) > 3*time.Second {
-				break
-			}
-			time.Sleep(wait)
-		}
-
-		_, err = device.CallMethod(ptz.Stop{
-			ProfileToken: token,
-			PanTilt:      true,
-			Zoom:         true,
-		})
+func waitForPTZTarget(client ptzMoveClient, target ptzTarget, timeout time.Duration, options ptzMoveOptions) error {
+	deadline := time.Now().Add(timeout)
+	lastProgress := time.Now()
+	bestDistance := math.Inf(1)
 
+	for {
+		position, err := client.GetPosition()
 		if err != nil {
-			log.Error("ContinuousPanTiltMove (Pan): " + err.Error())
+			return fmt.Errorf("read PTZ position: %w", err)
 		}
+		_, distance, reached, err := ptzTargetError(position, target, options)
+		if err != nil {
+			return err
+		}
+		if reached {
+			return nil
+		}
+
+		now := time.Now()
+		if bestDistance-distance >= options.ProgressEpsilon {
+			bestDistance = distance
+			lastProgress = now
+		}
+		if !now.Before(deadline) {
+			return fmt.Errorf("target not reached within %s", timeout)
+		}
+		if now.Sub(lastProgress) >= options.NoProgressTimeout {
+			return fmt.Errorf("PTZ position made no progress for %s", options.NoProgressTimeout)
+		}
+		time.Sleep(options.PollInterval)
 	}
-	return err
 }
 
-func TiltUntilPosition(device *onvif.Device, configuration ptz.GetConfigurationsResponse, token xsdonvif.ReferenceToken, tilt float64, zoom float64, speed float64, wait time.Duration) error {
-	position, err := GetPosition(device, token)
+func moveContinuouslyToPTZTarget(client ptzMoveClient, target ptzTarget, options ptzMoveOptions) (err error) {
+	defer func() {
+		if stopErr := client.Stop(); stopErr != nil {
+			err = errors.Join(err, fmt.Errorf("stop continuous move: %w", stopErr))
+		}
+	}()
 
-	if position.PanTilt.Y >= tilt-0.005 && position.PanTilt.Y <= tilt+0.005 {
+	deadline := time.Now().Add(options.ContinuousTimeout)
+	lastProgress := time.Now()
+	bestDistance := math.Inf(1)
+	var lastVelocity ptzVelocity
+	haveVelocity := false
 
-	} else {
-
-		// We'll need to determine if we need to move CW or CCW.
-		// Check the current position and compare it with the desired position.
-		directionY := speed
-		if position.PanTilt.Y > tilt {
-			directionY = speed * -1
+	for {
+		position, positionErr := client.GetPosition()
+		if positionErr != nil {
+			return fmt.Errorf("read PTZ position: %w", positionErr)
+		}
+		delta, distance, reached, targetErr := ptzTargetError(position, target, options)
+		if targetErr != nil {
+			return targetErr
+		}
+		if reached {
+			return nil
 		}
 
-		panTiltVector := xsdonvif.Vector2D{
-			X:     0,
-			Y:     directionY,
-			Space: configuration.PTZConfiguration[0].DefaultContinuousPanTiltVelocitySpace,
-		}
-
-		velocity := xsdonvif.PTZSpeedPanTilt{
-			PanTilt: panTiltVector,
-		}
-
-		resp, err := device.CallMethod(ptz.ContinuousMove{
-			ProfileToken: &token,
-			Velocity:     velocity,
-		})
-		err = handleONVIFResponse("continuous_tilt", resp, err)
-
-		// While moving we'll check if we reached the desired position.
-		// or if we overshot the desired position.
-
-		// Break after 3seconds
 		now := time.Now()
-		for {
-			position, _ := GetPosition(device, token)
-			if position.PanTilt.Y == -1 || position.PanTilt.Y == 1 || (directionY > 0 && position.PanTilt.Y >= tilt) || (directionY < 0 && position.PanTilt.Y <= tilt) || (position.PanTilt.Y >= tilt-0.005 && position.PanTilt.Y <= tilt+0.005) {
-				break
-			}
-			if time.Since(now) > 3*time.Second {
-				break
-			}
-			time.Sleep(wait)
+		if bestDistance-distance >= options.ProgressEpsilon {
+			bestDistance = distance
+			lastProgress = now
+		}
+		if !now.Before(deadline) {
+			return fmt.Errorf("target not reached within %s", options.ContinuousTimeout)
+		}
+		if now.Sub(lastProgress) >= options.NoProgressTimeout {
+			return fmt.Errorf("PTZ position made no progress for %s", options.NoProgressTimeout)
 		}
 
-		_, err = device.CallMethod(ptz.Stop{
-			ProfileToken: token,
-			PanTilt:      true,
-			Zoom:         true,
-		})
-
-		if err != nil {
-			log.Error("ContinuousPanTiltMove (Tilt): " + err.Error())
+		velocity := ptzVelocity{
+			Pan:  proportionalPTZVelocity(delta.Pan, options.PanTolerance, options),
+			Tilt: proportionalPTZVelocity(delta.Tilt, options.TiltTolerance, options),
+			Zoom: proportionalPTZVelocity(delta.Zoom, options.ZoomTolerance, options),
 		}
+		if !haveVelocity || ptzVelocityChanged(lastVelocity, velocity, options.VelocityChange) {
+			if moveErr := client.ContinuousMove(velocity); moveErr != nil {
+				return fmt.Errorf("set continuous PTZ velocity: %w", moveErr)
+			}
+			lastVelocity = velocity
+			haveVelocity = true
+		}
+		time.Sleep(options.PollInterval)
 	}
-	return err
 }
 
-func ZoomUntilPosition(device *onvif.Device, configuration ptz.GetConfigurationsResponse, token xsdonvif.ReferenceToken, zoom float64, speed float64, wait time.Duration) error {
-	position, err := GetPosition(device, token)
+func ptzTargetError(position xsdonvif.PTZVector, target ptzTarget, options ptzMoveOptions) (ptzVelocity, float64, bool, error) {
+	if position.PanTilt == nil || position.Zoom == nil {
+		return ptzVelocity{}, 0, false, errors.New("camera returned an incomplete PTZ position")
+	}
+	delta := ptzVelocity{
+		Pan:  target.Pan - position.PanTilt.X,
+		Tilt: target.Tilt - position.PanTilt.Y,
+		Zoom: target.Zoom - position.Zoom.X,
+	}
+	reached := math.Abs(delta.Pan) <= options.PanTolerance &&
+		math.Abs(delta.Tilt) <= options.TiltTolerance &&
+		math.Abs(delta.Zoom) <= options.ZoomTolerance
+	distance := math.Abs(delta.Pan) + math.Abs(delta.Tilt) + math.Abs(delta.Zoom)
+	return delta, distance, reached, nil
+}
 
-	if position.Zoom.X >= zoom-0.005 && position.Zoom.X <= zoom+0.005 {
+func proportionalPTZVelocity(delta float64, tolerance float64, options ptzMoveOptions) float64 {
+	if math.Abs(delta) <= tolerance {
+		return 0
+	}
+	speed := math.Abs(delta) * options.VelocityGain
+	speed = math.Max(options.MinVelocity, math.Min(speed, options.MaxVelocity))
+	return math.Copysign(speed, delta)
+}
 
-	} else {
+func ptzVelocityChanged(previous ptzVelocity, next ptzVelocity, threshold float64) bool {
+	return math.Abs(previous.Pan-next.Pan) >= threshold ||
+		math.Abs(previous.Tilt-next.Tilt) >= threshold ||
+		math.Abs(previous.Zoom-next.Zoom) >= threshold
+}
 
-		// We'll need to determine if we need to move CW or CCW.
-		// Check the current position and compare it with the desired position.
-		directionZ := speed
-		if position.Zoom.X > zoom {
-			directionZ = speed * -1
-		}
-
-		zoomVector := xsdonvif.Vector1D{
-			X:     directionZ,
-			Space: configuration.PTZConfiguration[0].DefaultContinuousZoomVelocitySpace,
-		}
-		resp, err := device.CallMethod(ptz.ContinuousMove{
-			ProfileToken: &token,
-			Velocity: xsdonvif.PTZSpeedZoom{
-				Zoom: zoomVector,
-			},
-		})
-		err = handleONVIFResponse("continuous_zoom_to_position", resp, err)
-
-		// While moving we'll check if we reached the desired position.
-		// or if we overshot the desired position.
-
-		// Break after 3seconds
-		now := time.Now()
-		for {
-			position, _ := GetPosition(device, token)
-			if position.Zoom.X == -1 || position.Zoom.X == 1 || (directionZ > 0 && position.Zoom.X >= zoom) || (directionZ < 0 && position.Zoom.X <= zoom) || (position.Zoom.X >= zoom-0.005 && position.Zoom.X <= zoom+0.005) {
-				break
-			}
-			if time.Since(now) > 3*time.Second {
-				break
-			}
-			time.Sleep(wait)
-		}
-
-		_, err = device.CallMethod(ptz.Stop{
-			ProfileToken: token,
-			PanTilt:      true,
-			Zoom:         true,
-		})
-
-		if err != nil {
-			log.Error("ContinuousPanTiltMove (Zoom): " + err.Error())
+func validatePTZTarget(target ptzTarget) error {
+	values := []struct {
+		name  string
+		value float64
+	}{
+		{name: "pan", value: target.Pan},
+		{name: "tilt", value: target.Tilt},
+		{name: "zoom", value: target.Zoom},
+	}
+	for _, value := range values {
+		if math.IsNaN(value.value) || math.IsInf(value.value, 0) {
+			return fmt.Errorf("invalid PTZ %s target", value.name)
 		}
 	}
-	return err
+	return nil
 }
 
 func ContinuousPanTilt(device *onvif.Device, configuration ptz.GetConfigurationsResponse, token xsdonvif.ReferenceToken, pan float64, tilt float64) error {
@@ -895,6 +957,10 @@ func GetPTZFunctionsFromDevice(configurations ptz.GetConfigurationsResponse) ([]
 	canZoom := false
 	canPanTilt := false
 
+	if len(configurations.PTZConfiguration) == 0 {
+		return functions, canZoom, canPanTilt
+	}
+
 	if configurations.PTZConfiguration[0].DefaultAbsolutePantTiltPositionSpace != nil {
 		functions = append(functions, "AbsolutePanTiltMove")
 		canPanTilt = true
@@ -1042,6 +1108,23 @@ func CreatePullPointSubscription(dev *onvif.Device) (string, error) {
 	return pullPointAdress, err
 }
 
+const (
+	wsAddressingNamespace = "http://www.w3.org/2005/08/addressing"
+	pullMessagesAction    = "http://www.onvif.org/ver10/events/wsdl/PullPointSubscription/PullMessagesRequest"
+	unsubscribeAction     = "http://docs.oasis-open.org/wsn/bw-2/SubscriptionManager/UnsubscribeRequest"
+)
+
+// subscriptionAddressingHeader builds the WS-Addressing Action and To headers
+// for a request sent to a pull-point subscription. Some devices identify the
+// subscription from wsa:To rather than the request URL, and reject requests
+// without it as NotAuthorized.
+func subscriptionAddressingHeader(subscriptionAddress string, action string) string {
+	var address bytes.Buffer
+	_ = xml.EscapeText(&address, []byte(subscriptionAddress))
+	return `<wsa:Action xmlns:wsa="` + wsAddressingNamespace + `">` + action + `</wsa:Action>` +
+		`<wsa:To xmlns:wsa="` + wsAddressingNamespace + `">` + address.String() + `</wsa:To>`
+}
+
 func UnsubscribePullPoint(dev *onvif.Device, pullPointAddress string) error {
 
 	// Unsubscribe from the device
@@ -1055,7 +1138,7 @@ func UnsubscribePullPoint(dev *onvif.Device, pullPointAddress string) error {
 		return err
 	}
 
-	res, err := dev.SendSoap(pullPointAddress, string(requestBody))
+	res, err := dev.SendSoapWithHeader(pullPointAddress, string(requestBody), subscriptionAddressingHeader(pullPointAddress, unsubscribeAction))
 	if err != nil {
 		log.WithError(err).WithFields(log.Fields{
 			"component": "onvif",
@@ -1137,7 +1220,7 @@ func GetEventMessages(dev *onvif.Device, pullPointAddress string) ([]ONVIFEvents
 				log.Error("onvif.main.GetEventMessages(pullMessages): " + err.Error())
 				return eventsArray, err
 			}
-			res, err := dev.SendSoap(string(subscriptionURI), string(requestBody))
+			res, err := dev.SendSoapWithHeader(string(subscriptionURI), string(requestBody), subscriptionAddressingHeader(subscriptionURI, pullMessagesAction))
 			if err != nil {
 				log.Error("onvif.main.GetEventMessages(pullMessages): " + err.Error())
 				return eventsArray, err

machinery/src/onvif/move_test.go

@@ -0,0 +1,264 @@
+package onvif
+
+import (
+	"encoding/xml"
+	"errors"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/kerberos-io/onvif/ptz"
+	xsdonvif "github.com/kerberos-io/onvif/xsd/onvif"
+)
+
+type fakePTZMoveClient struct {
+	current              ptzTarget
+	target               ptzTarget
+	lastVelocity         ptzVelocity
+	absoluteErr          error
+	continuousErr        error
+	stopErr              error
+	ignoreAbsolute       bool
+	stalled              bool
+	incompletePosition   bool
+	positionErr          error
+	positionErrAtCall    int
+	positionCalls        int
+	absoluteCalls        int
+	continuousVelocities []ptzVelocity
+	stopCalls            int
+	moving               bool
+}
+
+func (client *fakePTZMoveClient) GetPosition() (xsdonvif.PTZVector, error) {
+	client.positionCalls++
+	if client.positionErr != nil && client.positionCalls == client.positionErrAtCall {
+		return xsdonvif.PTZVector{}, client.positionErr
+	}
+	if client.moving && !client.stalled {
+		client.current.Pan = moveFakeAxis(client.current.Pan, client.target.Pan, client.lastVelocity.Pan)
+		client.current.Tilt = moveFakeAxis(client.current.Tilt, client.target.Tilt, client.lastVelocity.Tilt)
+		client.current.Zoom = moveFakeAxis(client.current.Zoom, client.target.Zoom, client.lastVelocity.Zoom)
+	}
+
+	panTilt := &xsdonvif.Vector2D{X: client.current.Pan, Y: client.current.Tilt}
+	if client.incompletePosition {
+		return xsdonvif.PTZVector{PanTilt: panTilt}, nil
+	}
+	return xsdonvif.PTZVector{
+		PanTilt: panTilt,
+		Zoom:    &xsdonvif.Vector1D{X: client.current.Zoom},
+	}, nil
+}
+
+func (client *fakePTZMoveClient) AbsoluteMove(target ptzTarget) error {
+	client.absoluteCalls++
+	client.target = target
+	if client.absoluteErr != nil {
+		return client.absoluteErr
+	}
+	if !client.ignoreAbsolute {
+		client.current = target
+	}
+	return nil
+}
+
+func (client *fakePTZMoveClient) ContinuousMove(velocity ptzVelocity) error {
+	if client.continuousErr != nil {
+		return client.continuousErr
+	}
+	client.lastVelocity = velocity
+	client.continuousVelocities = append(client.continuousVelocities, velocity)
+	client.moving = true
+	return nil
+}
+
+func (client *fakePTZMoveClient) Stop() error {
+	client.stopCalls++
+	client.moving = false
+	return client.stopErr
+}
+
+func moveFakeAxis(current float64, target float64, velocity float64) float64 {
+	step := velocity * 0.5
+	if step == 0 {
+		return current
+	}
+	next := current + step
+	if (velocity > 0 && next >= target) || (velocity < 0 && next <= target) {
+		return target
+	}
+	return next
+}
+
+func testPTZMoveOptions() ptzMoveOptions {
+	options := defaultPTZMoveOptions
+	options.NativeTimeout = 50 * time.Millisecond
+	options.ContinuousTimeout = 100 * time.Millisecond
+	options.NoProgressTimeout = 10 * time.Millisecond
+	options.PollInterval = time.Millisecond
+	options.ProgressEpsilon = 0.0001
+	options.VelocityChange = 0.01
+	return options
+}
+
+func TestMoveToPositionUsesVerifiedNativeMove(t *testing.T) {
+	target := ptzTarget{Pan: 0.5, Tilt: -0.25, Zoom: 0.4}
+	client := &fakePTZMoveClient{}
+
+	err := moveToPosition(client, ptzMoveSupport{Absolute: true, Continuous: true}, target, testPTZMoveOptions())
+	if err != nil {
+		t.Fatalf("moveToPosition() error = %v", err)
+	}
+	if client.absoluteCalls != 1 {
+		t.Fatalf("absolute calls = %d, want 1", client.absoluteCalls)
+	}
+	if len(client.continuousVelocities) != 0 {
+		t.Fatalf("continuous calls = %d, want 0", len(client.continuousVelocities))
+	}
+}
+
+func TestMoveToPositionFallsBackToSimultaneousContinuousMove(t *testing.T) {
+	target := ptzTarget{Pan: 0.5, Tilt: -0.25, Zoom: 0.4}
+	client := &fakePTZMoveClient{
+		target:      target,
+		absoluteErr: errors.New("absolute move rejected"),
+	}
+
+	err := moveToPosition(client, ptzMoveSupport{Absolute: true, Continuous: true}, target, testPTZMoveOptions())
+	if err != nil {
+		t.Fatalf("moveToPosition() error = %v", err)
+	}
+	if len(client.continuousVelocities) == 0 {
+		t.Fatal("continuous fallback was not used")
+	}
+	first := client.continuousVelocities[0]
+	if first.Pan == 0 || first.Tilt == 0 || first.Zoom == 0 {
+		t.Fatalf("first fallback velocity = %+v, want all axes moving", first)
+	}
+	if client.stopCalls != 2 {
+		t.Fatalf("stop calls = %d, want native cleanup and fallback cleanup", client.stopCalls)
+	}
+}
+
+func TestMoveToPositionFallsBackWhenNativeMoveMakesNoProgress(t *testing.T) {
+	target := ptzTarget{Pan: 0.3, Tilt: 0.2, Zoom: 0.1}
+	client := &fakePTZMoveClient{
+		target:         target,
+		ignoreAbsolute: true,
+	}
+	options := testPTZMoveOptions()
+	options.NoProgressTimeout = 3 * time.Millisecond
+
+	err := moveToPosition(client, ptzMoveSupport{Absolute: true, Continuous: true}, target, options)
+	if err != nil {
+		t.Fatalf("moveToPosition() error = %v", err)
+	}
+	if len(client.continuousVelocities) == 0 {
+		t.Fatal("continuous fallback was not used")
+	}
+}
+
+func TestMoveToPositionStopsAfterContinuousStatusError(t *testing.T) {
+	target := ptzTarget{Pan: 0.3, Tilt: 0.2, Zoom: 0.1}
+	client := &fakePTZMoveClient{
+		target:            target,
+		positionErr:       errors.New("status unavailable"),
+		positionErrAtCall: 3,
+	}
+
+	err := moveToPosition(client, ptzMoveSupport{Continuous: true}, target, testPTZMoveOptions())
+	if err == nil || !strings.Contains(err.Error(), "status unavailable") {
+		t.Fatalf("moveToPosition() error = %v, want status error", err)
+	}
+	if client.stopCalls != 1 {
+		t.Fatalf("stop calls = %d, want 1", client.stopCalls)
+	}
+}
+
+func TestMoveToPositionStopsWhenContinuousMoveIsRejected(t *testing.T) {
+	target := ptzTarget{Pan: 0.3, Tilt: 0.2, Zoom: 0.1}
+	client := &fakePTZMoveClient{
+		target:        target,
+		continuousErr: errors.New("continuous move rejected"),
+	}
+
+	err := moveToPosition(client, ptzMoveSupport{Continuous: true}, target, testPTZMoveOptions())
+	if err == nil || !strings.Contains(err.Error(), "continuous move rejected") {
+		t.Fatalf("moveToPosition() error = %v, want movement error", err)
+	}
+	if client.stopCalls != 1 {
+		t.Fatalf("stop calls = %d, want 1", client.stopCalls)
+	}
+}
+
+func TestMoveToPositionReturnsNoProgressErrorAndStops(t *testing.T) {
+	target := ptzTarget{Pan: 0.3, Tilt: 0.2, Zoom: 0.1}
+	client := &fakePTZMoveClient{
+		target:  target,
+		stalled: true,
+	}
+	options := testPTZMoveOptions()
+	options.NoProgressTimeout = 3 * time.Millisecond
+
+	err := moveToPosition(client, ptzMoveSupport{Continuous: true}, target, options)
+	if err == nil || !strings.Contains(err.Error(), "made no progress") {
+		t.Fatalf("moveToPosition() error = %v, want no-progress error", err)
+	}
+	if client.stopCalls != 1 {
+		t.Fatalf("stop calls = %d, want 1", client.stopCalls)
+	}
+}
+
+func TestMoveToPositionReturnsDeadlineErrorAndStops(t *testing.T) {
+	target := ptzTarget{Pan: 100, Tilt: 100, Zoom: 100}
+	client := &fakePTZMoveClient{target: target}
+	options := testPTZMoveOptions()
+	options.ContinuousTimeout = 3 * time.Millisecond
+	options.NoProgressTimeout = 50 * time.Millisecond
+
+	err := moveToPosition(client, ptzMoveSupport{Continuous: true}, target, options)
+	if err == nil || !strings.Contains(err.Error(), "target not reached within") {
+		t.Fatalf("moveToPosition() error = %v, want deadline error", err)
+	}
+	if client.stopCalls != 1 {
+		t.Fatalf("stop calls = %d, want 1", client.stopCalls)
+	}
+}
+
+func TestMoveToPositionRejectsIncompletePosition(t *testing.T) {
+	client := &fakePTZMoveClient{incompletePosition: true}
+
+	err := moveToPosition(client, ptzMoveSupport{Continuous: true}, ptzTarget{Pan: 0.1}, testPTZMoveOptions())
+	if err == nil || !strings.Contains(err.Error(), "incomplete PTZ position") {
+		t.Fatalf("moveToPosition() error = %v, want incomplete-position error", err)
+	}
+}
+
+func TestCombinedContinuousMoveVelocityIncludesEveryAxis(t *testing.T) {
+	request := ptz.ContinuousMove{
+		Velocity: ptz.Speed{
+			PanTilt: &xsdonvif.Vector2D{X: 0.5, Y: -0.25},
+			Zoom:    &xsdonvif.Vector1D{X: 0.2},
+		},
+	}
+
+	encoded, err := xml.Marshal(request)
+	if err != nil {
+		t.Fatalf("xml.Marshal() error = %v", err)
+	}
+	xmlBody := string(encoded)
+	if !strings.Contains(xmlBody, `PanTilt x="0.5" y="-0.25"`) {
+		t.Fatalf("ContinuousMove XML = %s, want pan and tilt", xmlBody)
+	}
+	if !strings.Contains(xmlBody, `Zoom x="0.2"`) {
+		t.Fatalf("ContinuousMove XML = %s, want zoom", xmlBody)
+	}
+}
+
+func TestGetPTZFunctionsHandlesMissingConfiguration(t *testing.T) {
+	functions, canZoom, canPanTilt := GetPTZFunctionsFromDevice(ptz.GetConfigurationsResponse{})
+	if len(functions) != 0 || canZoom || canPanTilt {
+		t.Fatalf("GetPTZFunctionsFromDevice() = %v, %t, %t; want empty capabilities", functions, canZoom, canPanTilt)
+	}
+}

machinery/src/onvif/subscription_test.go

@@ -0,0 +1,88 @@
+package onvif
+
+import (
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"sync"
+	"testing"
+
+	"github.com/kerberos-io/onvif"
+)
+
+// newSubscriptionTestDevice starts a fake ONVIF device and returns a connected
+// device plus the request bodies it received on the subscription endpoint.
+func newSubscriptionTestDevice(t *testing.T, response string) (*onvif.Device, string, func() []string) {
+	t.Helper()
+
+	var mu sync.Mutex
+	var subscriptionBodies []string
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		body, _ := io.ReadAll(r.Body)
+		if r.URL.Path == "/onvif/subscription" {
+			mu.Lock()
+			subscriptionBodies = append(subscriptionBodies, string(body))
+			mu.Unlock()
+			_, _ = w.Write([]byte(response))
+			return
+		}
+		_, _ = w.Write([]byte(`<Envelope><Body/></Envelope>`))
+	}))
+	t.Cleanup(srv.Close)
+
+	dev, err := onvif.NewDevice(onvif.DeviceParams{Xaddr: strings.TrimPrefix(srv.URL, "http://")})
+	if err != nil {
+		t.Fatalf("NewDevice() error = %v", err)
+	}
+	subscriptionAddress := srv.URL + "/onvif/subscription?subscription=7&idx=2"
+	return dev, subscriptionAddress, func() []string {
+		mu.Lock()
+		defer mu.Unlock()
+		return append([]string(nil), subscriptionBodies...)
+	}
+}
+
+func assertSubscriptionAddressing(t *testing.T, body string, action string) {
+	t.Helper()
+
+	for _, want := range []string{
+		`<wsa:Action xmlns:wsa="http://www.w3.org/2005/08/addressing">` + action + `</wsa:Action>`,
+		`<wsa:To xmlns:wsa="http://www.w3.org/2005/08/addressing">`,
+		`/onvif/subscription?subscription=7&amp;idx=2</wsa:To>`,
+	} {
+		if !strings.Contains(body, want) {
+			t.Fatalf("request body does not contain %q:\n%s", want, body)
+		}
+	}
+}
+
+// Some devices identify a pull-point subscription by wsa:To and reject
+// PullMessages without it as NotAuthorized.
+func TestGetEventMessagesSendsSubscriptionAddressingHeaders(t *testing.T) {
+	dev, subscriptionAddress, bodies := newSubscriptionTestDevice(t, `<Envelope><Body><PullMessagesResponse/></Body></Envelope>`)
+
+	if _, err := GetEventMessages(dev, subscriptionAddress); err != nil {
+		t.Fatalf("GetEventMessages() error = %v", err)
+	}
+
+	received := bodies()
+	if len(received) != 1 {
+		t.Fatalf("subscription requests = %d, want 1", len(received))
+	}
+	assertSubscriptionAddressing(t, received[0], pullMessagesAction)
+}
+
+func TestUnsubscribePullPointSendsSubscriptionAddressingHeaders(t *testing.T) {
+	dev, subscriptionAddress, bodies := newSubscriptionTestDevice(t, `<Envelope><Body><UnsubscribeResponse/></Body></Envelope>`)
+
+	if err := UnsubscribePullPoint(dev, subscriptionAddress); err != nil {
+		t.Fatalf("UnsubscribePullPoint() error = %v", err)
+	}
+
+	received := bodies()
+	if len(received) != 1 {
+		t.Fatalf("subscription requests = %d, want 1", len(received))
+	}
+	assertSubscriptionAddressing(t, received[0], unsubscribeAction)
+}

machinery/src/packets/queue.go

@@ -40,6 +40,9 @@ func NewQueue() *Queue {
 }
 
 func (self *Queue) SetMaxGopCount(n int) {
+	if n < 1 {
+		n = 1
+	}
 	self.lock.Lock()
 	self.maxgopcount = n
 	self.lock.Unlock()
@@ -97,20 +100,31 @@ func (self *Queue) WritePacket(pkt Packet) (err error) {
 	self.lock.Lock()
 
 	self.buf.Push(pkt)
-	if pkt.Idx == int8(self.videoidx) && pkt.IsKeyFrame {
+	isVideoKeyframe := pkt.Idx == int8(self.videoidx) && pkt.IsKeyFrame
+	if isVideoKeyframe {
 		self.curgopcount++
-	}
 
-	for self.curgopcount >= self.maxgopcount && self.buf.Count > 1 {
-		pkt := self.buf.Pop()
-		if pkt.Idx == int8(self.videoidx) && pkt.IsKeyFrame {
-			self.curgopcount--
-		}
-		if self.curgopcount < self.maxgopcount {
-			break
+		for self.curgopcount > self.maxgopcount && self.buf.Count > 0 {
+			for self.buf.Count > 0 {
+				oldest := self.buf.Pop()
+				if oldest.Idx == int8(self.videoidx) && oldest.IsKeyFrame {
+					self.curgopcount--
+					break
+				}
+			}
+
+			// Discard the dependent frames from the evicted GOP as well. Keeping
+			// them would let an overrun cursor resume on a P-frame whose reference
+			// keyframe is no longer in the queue.
+			for self.buf.Count > 0 {
+				oldest := self.buf.Get(self.buf.Head)
+				if oldest.Idx == int8(self.videoidx) && oldest.IsKeyFrame {
+					break
+				}
+				self.buf.Pop()
+			}
 		}
 	}
-	//println("shrink", self.curgopcount, self.maxgopcount, self.buf.Head, self.buf.Tail, "count", self.buf.Count, "size", self.buf.Size)
 
 	self.cond.Broadcast()
 

machinery/src/packets/queue_test.go

@@ -107,3 +107,102 @@ func TestQueueCursorReadPacketContextDoesNotAllocateWhenPacketIsAvailable(t *tes
 		t.Fatalf("ReadPacketContext() allocations = %.2f, want 0", allocations)
 	}
 }
+
+func TestQueueEvictionKeepsKeyframeAtHead(t *testing.T) {
+	queue := NewQueue()
+	if err := queue.WriteHeader([]Stream{{Index: 0, IsVideo: true}}); err != nil {
+		t.Fatal(err)
+	}
+	queue.SetMaxGopCount(2)
+
+	for _, packet := range []Packet{
+		{Idx: 0, IsVideo: true, IsKeyFrame: true, Data: []byte("IDR-0")},
+		{Idx: 1, IsAudio: true, Data: []byte("audio-0")},
+		{Idx: 0, IsVideo: true, Data: []byte("P-0")},
+		{Idx: 0, IsVideo: true, IsKeyFrame: true, Data: []byte("IDR-1")},
+		{Idx: 0, IsVideo: true, Data: []byte("P-1")},
+		{Idx: 0, IsVideo: true, IsKeyFrame: true, Data: []byte("IDR-2")},
+	} {
+		if err := queue.WritePacket(packet); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	oldest, err := queue.Oldest().ReadPacket()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !oldest.IsKeyFrame || string(oldest.Data) != "IDR-1" {
+		t.Fatalf("oldest packet = %q (keyframe=%t), want IDR-1 keyframe", oldest.Data, oldest.IsKeyFrame)
+	}
+}
+
+func TestQueueWithSingleGOPRetainsCurrentKeyframe(t *testing.T) {
+	queue := NewQueue()
+	if err := queue.WriteHeader([]Stream{{Index: 0, IsVideo: true}}); err != nil {
+		t.Fatal(err)
+	}
+	queue.SetMaxGopCount(1)
+
+	for _, packet := range []Packet{
+		{Idx: 0, IsVideo: true, IsKeyFrame: true, Data: []byte("IDR-0")},
+		{Idx: 0, IsVideo: true, Data: []byte("P-0")},
+		{Idx: 0, IsVideo: true, IsKeyFrame: true, Data: []byte("IDR-1")},
+	} {
+		if err := queue.WritePacket(packet); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	oldest, err := queue.Oldest().ReadPacket()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !oldest.IsKeyFrame || string(oldest.Data) != "IDR-1" {
+		t.Fatalf("oldest packet = %q (keyframe=%t), want IDR-1 keyframe", oldest.Data, oldest.IsKeyFrame)
+	}
+}
+
+func TestLaggingQueueCursorResumesAtKeyframe(t *testing.T) {
+	queue := NewQueue()
+	if err := queue.WriteHeader([]Stream{{Index: 0, IsVideo: true}}); err != nil {
+		t.Fatal(err)
+	}
+	queue.SetMaxGopCount(2)
+
+	for _, packet := range []Packet{
+		{Idx: 0, IsVideo: true, IsKeyFrame: true, Data: []byte("IDR-0")},
+		{Idx: 0, IsVideo: true, Data: []byte("P-0")},
+		{Idx: 0, IsVideo: true, IsKeyFrame: true, Data: []byte("IDR-1")},
+	} {
+		if err := queue.WritePacket(packet); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	cursor := queue.Oldest()
+	first, err := cursor.ReadPacket()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if string(first.Data) != "IDR-0" {
+		t.Fatalf("first packet = %q, want IDR-0", first.Data)
+	}
+
+	for _, packet := range []Packet{
+		{Idx: 0, IsVideo: true, Data: []byte("P-1")},
+		{Idx: 0, IsVideo: true, IsKeyFrame: true, Data: []byte("IDR-2")},
+	} {
+		if err := queue.WritePacket(packet); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	resumed, err := cursor.ReadPacket()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !resumed.IsKeyFrame || string(resumed.Data) != "IDR-1" {
+		t.Fatalf("resumed packet = %q (keyframe=%t), want IDR-1 keyframe", resumed.Data, resumed.IsKeyFrame)
+	}
+}

machinery/src/routers/http/routes.go

@@ -65,12 +65,12 @@ func AddRoutes(r *gin.Engine, authMiddleware *jwt.GinJWTMiddleware, configDirect
 
 			// Will verify the hub settings.
 			api.POST("/hub/verify", func(c *gin.Context) {
-				cloud.VerifyHub(c)
+				cloud.VerifyHub(c, configuration.Config.HubURI)
 			})
 
 			// Will verify the persistence settings.
 			api.POST("/persistence/verify", func(c *gin.Context) {
-				cloud.VerifyPersistence(c, configDirectory)
+				cloud.VerifyPersistence(c, configDirectory, configuration.Config.HubURI)
 			})
 
 			// Will verify the secondary persistence settings.

ui/public/locales/en/translation.json

@@ -223,6 +223,8 @@
       "kerberosvault_description_accesskey": "The access key of your Vault account.",
       "kerberosvault_secretkey": "Secret key",
       "kerberosvault_description_secretkey": "The secret key of your Vault account.",
+      "kerberosvault_customheaders": "Custom upload headers",
+      "kerberosvault_description_customheaders": "A JSON object of custom HTTP headers, for example {\"site_id\":\"site-1\",\"line_id\":\"line-2\"}. These values are included as metadata in Vault integration messages.",
       "kerberosvault_maxretries": "Max retries",
       "kerberosvault_description_maxretries": "The maximum number of retries to upload a recording.",
       "kerberosvault_timeout": "Timeout",

ui/src/config.js

@@ -1,5 +1,6 @@
 const { hostname, host, protocol } = window.location;
 const websocketprotocol = protocol === 'http:' ? 'ws:' : 'wss:';
+const devAgentPort = process.env.REACT_APP_AGENT_API_PORT || '8082';
 
 // Uncomment this when using codespaces or other special DNS names (which you can't control)
 // replace this with the DNS name of the kerberos agent server (the codespace url)
@@ -9,9 +10,9 @@ const dev = {
   ENV: 'dev',
   // Comment the below lines, when using codespaces or other special DNS names (which you can't control)
   HOSTNAME: hostname,
-  API_URL: `${protocol}//${hostname}:8080/api`,
-  URL: `${protocol}//${hostname}:8080`,
-  WS_URL: `${websocketprotocol}//${hostname}:8080/ws`,
+  API_URL: `${protocol}//${hostname}:${devAgentPort}/api`,
+  URL: `${protocol}//${hostname}:${devAgentPort}`,
+  WS_URL: `${websocketprotocol}//${hostname}:${devAgentPort}/ws`,
   MODE: window['env']['mode'],
   // Uncomment, and comment the above lines, when using codespaces or other special DNS names (which you can't control)
   // HOSTNAME: externalHost,

ui/src/pages/Settings/Settings.jsx

@@ -2542,6 +2542,28 @@ export class Settings extends React.Component {
                           )
                         }
                       />
+                      <Input
+                        noPadding
+                        label={t(
+                          'settings.persistence.kerberosvault_customheaders'
+                        )}
+                        placeholder={t(
+                          'settings.persistence.kerberosvault_description_customheaders'
+                        )}
+                        value={
+                          config.kstorage
+                            ? config.kstorage.custom_headers || ''
+                            : ''
+                        }
+                        onChange={(value) =>
+                          this.onUpdateField(
+                            'kstorage',
+                            'custom_headers',
+                            value,
+                            config.kstorage
+                          )
+                        }
+                      />
                       <Input
                         noPadding
                         label={t(
@@ -2815,6 +2837,28 @@ export class Settings extends React.Component {
                       )
                     }
                   />
+                  <Input
+                    noPadding
+                    label={t(
+                      'settings.persistence.kerberosvault_customheaders'
+                    )}
+                    placeholder={t(
+                      'settings.persistence.kerberosvault_description_customheaders'
+                    )}
+                    value={
+                      config.kstorage_secondary
+                        ? config.kstorage_secondary.custom_headers || ''
+                        : ''
+                    }
+                    onChange={(value) =>
+                      this.onUpdateField(
+                        'kstorage_secondary',
+                        'custom_headers',
+                        value,
+                        config.kstorage_secondary
+                      )
+                    }
+                  />
                 </BlockBody>
                 <BlockFooter>
                   <Button