Merge pull request #339 from kerberos-io/fix/idr-loss-recovery

fix/idr-loss-recovery

.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 }} .

README.md

@@ -347,20 +347,6 @@ See [RTSPS and TLS certificates](README-RTSPS-TLS.md) for the complete Bosch UI,
 | `AGENT_REMOTE_ACCESS_ENABLED`               | Allow encrypted Hub MQTT sessions to stream Agent logs and open an interactive shell. Enable only for trusted deployments. | "false"                        |
 | `AGENT_REALTIME_PROCESSING`                 | If `AGENT_REALTIME_PROCESSING` set to `true`, the agent will send key frames to the topic       | ""                             |
 | `AGENT_REALTIME_PROCESSING_TOPIC`           | The topic to which keyframes will be sent in base64 encoded format.                             | ""                             |
-| `AGENT_FRAME_PROCESSING_ENABLED`            | Send keyframe-aligned JPEGs to an external Frame Processor over HTTP.                           | "false"                        |
-| `AGENT_FRAME_PROCESSING_ENDPOINT`           | Full Frame Processor HTTP endpoint, including `/v1/frames`.                                     | ""                             |
-| `AGENT_FRAME_PROCESSING_TOKEN`              | Environment-only bearer token used to authenticate frame submissions; never returned by config APIs. | ""                         |
-| `AGENT_FRAME_PROCESSING_PROFILE`            | Processing profile included with each frame.                                                    | "never-trigger"                |
-| `AGENT_FRAME_PROCESSING_ALLOW_REQUESTED_FRAMES` | Allow authenticated MQTT `capture-frame` commands; frame bytes are still submitted over HTTP. | "false"                    |
-| `AGENT_FRAME_PROCESSING_STREAM`             | Source stream: `auto`, `main`, or `sub`; `auto` prefers the substream when available.           | "auto"                         |
-| `AGENT_FRAME_PROCESSING_INTERVAL_SECONDS`   | Target period between submissions; the first keyframe at or after each deadline is selected.   | "10"                           |
-| `AGENT_FRAME_PROCESSING_WIDTH`              | Output JPEG width; aspect ratio is preserved when height is `0`.                               | "640"                          |
-| `AGENT_FRAME_PROCESSING_HEIGHT`             | Output JPEG height; `0` derives it from the source aspect ratio.                               | "0"                            |
-| `AGENT_FRAME_PROCESSING_JPEG_QUALITY`       | JPEG quality from 1 through 100.                                                                | "70"                           |
-| `AGENT_FRAME_PROCESSING_REQUEST_TIMEOUT_SECONDS` | Maximum duration of one HTTP submission.                                                   | "5"                            |
-| `AGENT_FRAME_PROCESSING_FRAME_TTL_SECONDS`  | Time after capture during which the Frame Processor may accept the frame.                       | "30"                           |
-| `AGENT_FRAME_PROCESSING_MAX_FRAME_BYTES`    | Maximum encoded JPEG size; values above 16 MiB are rejected.                                    | "4194304"                      |
-| `AGENT_FRAME_PROCESSING_PERIODIC_QUEUE_CAPACITY` | Bounded latest-wins periodic frame queue capacity, from 1 through 64.                       | "1"                            |
 | `AGENT_STUN_URI`                            | When using WebRTC, you'll need to provide a STUN server.                                        | "stun:turn-fra1.kerberos.io:3478"|
 | `AGENT_FORCE_TURN`                          | Force using a TURN server, by generating relay candidates only.                                 | "false"                        |
 | `AGENT_TURN_URI`                            | When using WebRTC, you'll need to provide a TURN server.                                        | "turn:turn-fra1.kerberos.io:3478"|
@@ -379,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"                        |
@@ -422,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
 

machinery/data/config/config.json

@@ -120,21 +120,6 @@
 	"condition_uri": "",
 	"encryption": {},
 	"signing": {},
-	"frameProcessing": {
-		"enabled": "false",
-		"endpoint": "",
-		"profile": "never-trigger",
-		"allowRequestedFrames": "false",
-		"stream": "auto",
-		"intervalSeconds": 10,
-		"width": 640,
-		"height": 0,
-		"jpegQuality": 70,
-		"requestTimeoutSeconds": 5,
-		"frameTtlSeconds": 30,
-		"maxFrameBytes": 4194304,
-		"periodicQueueCapacity": 1
-	},
 	"realtimeprocessing": "false",
 	"realtimeprocessing_topic": ""
 }

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()

machinery/src/cloud/cloud_test.go

@@ -15,6 +15,7 @@ import (
 	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 TestHeartbeatFailureLogOmitsHubResponseBody(t *testing.T) {
@@ -590,3 +591,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/frameprocessing/frameprocessing.go

@@ -1,550 +0,0 @@
-package frameprocessing
-
-import (
-	"bytes"
-	"context"
-	"crypto/tls"
-	"encoding/json"
-	"errors"
-	"fmt"
-	"image"
-	"image/jpeg"
-	"io"
-	"mime/multipart"
-	"net/http"
-	"net/textproto"
-	"net/url"
-	"os"
-	"time"
-
-	mqtt "github.com/eclipse/paho.mqtt.golang"
-	"github.com/gofrs/uuid"
-	"github.com/kerberos-io/agent/machinery/src/models"
-	"github.com/kerberos-io/agent/machinery/src/packets"
-	"github.com/kerberos-io/agent/machinery/src/utils"
-	log "github.com/sirupsen/logrus"
-)
-
-const (
-	schemaVersion        = "1.0"
-	maxResponseBodyBytes = 64 << 10
-)
-
-type Decoder interface {
-	DecodePacket(packets.Packet) (image.YCbCr, error)
-}
-
-type Observer interface {
-	SetFrameProcessingConfigured(bool)
-	RecordFrameProcessingSample()
-	RecordFrameProcessingQueued(int, bool)
-	SetFrameProcessingQueueDepth(int)
-	RecordFrameProcessingSuccess(time.Time)
-	RecordFrameProcessingFailure()
-}
-
-type Metadata struct {
-	SchemaVersion     string `json:"schemaVersion"`
-	RequestID         string `json:"requestId"`
-	FrameID           string `json:"frameId"`
-	DeviceID          string `json:"deviceId"`
-	CapturedAt        int64  `json:"capturedAt"`
-	ExpiresAt         int64  `json:"expiresAt"`
-	ProcessingProfile string `json:"processingProfile"`
-	SourceStream      string `json:"sourceStream"`
-	Width             int    `json:"width"`
-	Height            int    `json:"height"`
-	TraceID           string `json:"traceId,omitempty"`
-}
-
-type Frame struct {
-	Metadata Metadata
-	JPEG     []byte
-}
-
-type Sender struct {
-	endpoint string
-	token    string
-	client   *http.Client
-}
-
-type StatusPublisher interface {
-	Publish(context.Context, models.FrameProcessingStatus) error
-}
-
-type MQTTStatusPublisher struct {
-	client        mqtt.Client
-	hubKey        string
-	configuration *models.Configuration
-	timeout       time.Duration
-}
-
-func NewMQTTStatusPublisher(client mqtt.Client, hubKey string, configuration *models.Configuration) *MQTTStatusPublisher {
-	return &MQTTStatusPublisher{
-		client: client, hubKey: hubKey, configuration: configuration, timeout: 5 * time.Second,
-	}
-}
-
-func (p *MQTTStatusPublisher) Publish(ctx context.Context, status models.FrameProcessingStatus) error {
-	if p == nil || p.client == nil || p.hubKey == "" || p.configuration == nil {
-		return errors.New("frame-processing MQTT status publisher is not configured")
-	}
-	value, err := structToMap(status)
-	if err != nil {
-		return err
-	}
-	payload, err := models.PackageMQTTMessage(p.configuration, models.Message{
-		Payload: models.Payload{
-			Version:  schemaVersion,
-			Action:   models.FrameProcessingStatusAction,
-			DeviceId: status.DeviceID,
-			Value:    value,
-		},
-	})
-	if err != nil {
-		return fmt.Errorf("package frame-processing status: %w", err)
-	}
-	token := p.client.Publish("kerberos/hub/"+p.hubKey, 1, false, payload)
-	timer := time.NewTimer(p.timeout)
-	defer timer.Stop()
-	select {
-	case <-ctx.Done():
-		return ctx.Err()
-	case <-timer.C:
-		return errors.New("frame-processing status publish timed out")
-	case <-token.Done():
-		if err := token.Error(); err != nil {
-			return fmt.Errorf("publish frame-processing status: %w", err)
-		}
-		return nil
-	}
-}
-
-func structToMap(value any) (map[string]interface{}, error) {
-	encoded, err := json.Marshal(value)
-	if err != nil {
-		return nil, fmt.Errorf("marshal value: %w", err)
-	}
-	var result map[string]interface{}
-	if err := json.Unmarshal(encoded, &result); err != nil {
-		return nil, fmt.Errorf("decode value map: %w", err)
-	}
-	return result, nil
-}
-
-func NewSender(config models.FrameProcessing) (*Sender, error) {
-	if config.Token == "" {
-		return nil, errors.New("frameProcessing.token is required")
-	}
-	endpoint, err := url.ParseRequestURI(config.Endpoint)
-	if err != nil || (endpoint.Scheme != "http" && endpoint.Scheme != "https") || endpoint.Host == "" {
-		return nil, errors.New("frameProcessing.endpoint must be an absolute HTTP or HTTPS URL")
-	}
-	transport := http.DefaultTransport.(*http.Transport).Clone()
-	if os.Getenv("AGENT_TLS_INSECURE") == "true" {
-		if transport.TLSClientConfig == nil {
-			transport.TLSClientConfig = &tls.Config{}
-		}
-		transport.TLSClientConfig.InsecureSkipVerify = true
-	}
-	return &Sender{
-		endpoint: endpoint.String(),
-		token:    config.Token,
-		client: &http.Client{
-			Transport: transport,
-			Timeout:   time.Duration(config.RequestTimeoutSeconds) * time.Second,
-		},
-	}, nil
-}
-
-func (s *Sender) Submit(ctx context.Context, frame Frame) error {
-	var body bytes.Buffer
-	writer := multipart.NewWriter(&body)
-	metadataHeader := make(textproto.MIMEHeader)
-	metadataHeader.Set("Content-Disposition", `form-data; name="metadata"`)
-	metadataHeader.Set("Content-Type", "application/json")
-	metadataPart, err := writer.CreatePart(metadataHeader)
-	if err != nil {
-		return fmt.Errorf("create metadata part: %w", err)
-	}
-	if err := json.NewEncoder(metadataPart).Encode(frame.Metadata); err != nil {
-		return fmt.Errorf("encode metadata: %w", err)
-	}
-	frameHeader := make(textproto.MIMEHeader)
-	frameHeader.Set("Content-Disposition", `form-data; name="frame"; filename="frame.jpg"`)
-	frameHeader.Set("Content-Type", "image/jpeg")
-	framePart, err := writer.CreatePart(frameHeader)
-	if err != nil {
-		return fmt.Errorf("create frame part: %w", err)
-	}
-	if _, err := framePart.Write(frame.JPEG); err != nil {
-		return fmt.Errorf("write frame part: %w", err)
-	}
-	if err := writer.Close(); err != nil {
-		return fmt.Errorf("close multipart body: %w", err)
-	}
-
-	request, err := http.NewRequestWithContext(ctx, http.MethodPost, s.endpoint, &body)
-	if err != nil {
-		return fmt.Errorf("create frame request: %w", err)
-	}
-	request.Header.Set("Content-Type", writer.FormDataContentType())
-	if s.token != "" {
-		request.Header.Set("Authorization", "Bearer "+s.token)
-	}
-	response, err := s.client.Do(request)
-	if err != nil {
-		return fmt.Errorf("submit frame: %w", err)
-	}
-	defer response.Body.Close()
-	_, readErr := io.Copy(io.Discard, io.LimitReader(response.Body, maxResponseBodyBytes))
-	if readErr != nil {
-		return fmt.Errorf("read frame response: %w", readErr)
-	}
-	if response.StatusCode < http.StatusOK || response.StatusCode >= http.StatusMultipleChoices {
-		return fmt.Errorf("frame processor returned %s", response.Status)
-	}
-	return nil
-}
-
-func Run(
-	ctx context.Context,
-	cursor *packets.QueueCursor,
-	decoder Decoder,
-	config models.FrameProcessing,
-	deviceID string,
-	stream string,
-	observer Observer,
-) error {
-	if config.Enabled != "true" {
-		return nil
-	}
-	if cursor == nil || decoder == nil {
-		return errors.New("frame processing requires a packet cursor and decoder")
-	}
-	if deviceID == "" {
-		return errors.New("frame processing requires a device ID")
-	}
-	if err := validateConfig(config); err != nil {
-		return err
-	}
-	sender, err := NewSender(config)
-	if err != nil {
-		return err
-	}
-	if observer != nil {
-		observer.SetFrameProcessingConfigured(true)
-		defer observer.SetFrameProcessingConfigured(false)
-	}
-
-	frames := make(chan Frame, config.PeriodicQueueCapacity)
-	samplerDone := make(chan error, 1)
-	go func() {
-		samplerDone <- sample(ctx, cursor, decoder, config, deviceID, stream, frames, observer)
-		close(frames)
-	}()
-
-	for {
-		select {
-		case <-ctx.Done():
-			<-samplerDone
-			return nil
-		case err := <-samplerDone:
-			return normalizeCancellation(ctx, err)
-		case frame, ok := <-frames:
-			if !ok {
-				return normalizeCancellation(ctx, <-samplerDone)
-			}
-			if observer != nil {
-				observer.SetFrameProcessingQueueDepth(len(frames))
-			}
-			if err := sender.Submit(ctx, frame); err != nil {
-				if ctx.Err() != nil {
-					return nil
-				}
-				if observer != nil {
-					observer.RecordFrameProcessingFailure()
-					observer.SetFrameProcessingQueueDepth(len(frames))
-				}
-				log.WithError(err).WithFields(log.Fields{
-					"component": "frame_processing",
-					"device_id": deviceID,
-					"event":     "frame_submission_failed",
-					"frame_id":  frame.Metadata.FrameID,
-				}).Warn("Failed to submit frame for processing")
-				continue
-			}
-			if observer != nil {
-				observer.RecordFrameProcessingSuccess(time.Now())
-				observer.SetFrameProcessingQueueDepth(len(frames))
-			}
-		}
-	}
-}
-
-func RunRequested(
-	ctx context.Context,
-	decoder Decoder,
-	config models.FrameProcessing,
-	deviceID string,
-	stream string,
-	requests <-chan models.FrameProcessingWork,
-	statusPublisher StatusPublisher,
-	observer Observer,
-) error {
-	if config.Enabled != "true" {
-		return nil
-	}
-	if decoder == nil || requests == nil {
-		return errors.New("requested frame processing requires a decoder and request channel")
-	}
-	if deviceID == "" {
-		return errors.New("requested frame processing requires a device ID")
-	}
-	if err := validateConfig(config); err != nil {
-		return err
-	}
-	sender, err := NewSender(config)
-	if err != nil {
-		return err
-	}
-
-	for {
-		select {
-		case <-ctx.Done():
-			return nil
-		case work, ok := <-requests:
-			if !ok {
-				return nil
-			}
-			request := work.Request
-			if work.Cursor == nil {
-				publishStatus(ctx, statusPublisher, request, deviceID, "", "failed", true, "capture cursor is unavailable")
-				continue
-			}
-			if request.ExpiresAt <= time.Now().UnixMilli() {
-				publishStatus(ctx, statusPublisher, request, deviceID, "", "expired", false, "capture request expired")
-				continue
-			}
-			packet, err := nextKeyframe(ctx, work.Cursor, request.ExpiresAt)
-			if err != nil {
-				if ctx.Err() != nil {
-					return nil
-				}
-				publishStatus(ctx, statusPublisher, request, deviceID, "", "expired", false, "no keyframe before request expiry")
-				continue
-			}
-			frame, err := prepareRequestedFrame(packet, decoder, config, deviceID, stream, request, time.Now())
-			if err != nil {
-				if observer != nil {
-					observer.RecordFrameProcessingFailure()
-				}
-				publishStatus(ctx, statusPublisher, request, deviceID, "", "failed", true, "failed to prepare frame")
-				continue
-			}
-			if err := sender.Submit(ctx, frame); err != nil {
-				if ctx.Err() != nil {
-					return nil
-				}
-				if observer != nil {
-					observer.RecordFrameProcessingFailure()
-				}
-				publishStatus(ctx, statusPublisher, request, deviceID, frame.Metadata.FrameID, "failed", true, "frame submission failed")
-				continue
-			}
-			if observer != nil {
-				observer.RecordFrameProcessingSuccess(time.Now())
-			}
-			publishStatus(ctx, statusPublisher, request, deviceID, frame.Metadata.FrameID, "submitted", false, "")
-		}
-	}
-}
-
-func sample(
-	ctx context.Context,
-	cursor *packets.QueueCursor,
-	decoder Decoder,
-	config models.FrameProcessing,
-	deviceID string,
-	stream string,
-	frames chan Frame,
-	observer Observer,
-) error {
-	interval := time.Duration(config.IntervalSeconds) * time.Second
-	nextDeadline := time.Now().Add(interval)
-	for {
-		packet, err := cursor.ReadPacketContext(ctx)
-		if err != nil {
-			return err
-		}
-		now := time.Now()
-		if len(packet.Data) == 0 || !packet.IsKeyFrame || now.Before(nextDeadline) {
-			continue
-		}
-		for !nextDeadline.After(now) {
-			nextDeadline = nextDeadline.Add(interval)
-		}
-		if observer != nil {
-			observer.RecordFrameProcessingSample()
-		}
-		frame, err := prepareFrame(packet, decoder, config, deviceID, stream, now)
-		if err != nil {
-			if observer != nil {
-				observer.RecordFrameProcessingFailure()
-			}
-			log.WithError(err).WithFields(log.Fields{
-				"component": "frame_processing",
-				"event":     "frame_preparation_failed",
-				"stream":    stream,
-			}).Warn("Failed to prepare frame for processing")
-			continue
-		}
-		dropped := enqueueLatest(frames, frame)
-		if observer != nil {
-			observer.RecordFrameProcessingQueued(len(frames), dropped)
-		}
-	}
-}
-
-func prepareFrame(packet packets.Packet, decoder Decoder, config models.FrameProcessing, deviceID, stream string, now time.Time) (Frame, error) {
-	frameID, err := uuid.NewV4()
-	if err != nil {
-		return Frame{}, fmt.Errorf("generate frame ID: %w", err)
-	}
-	return prepareFrameWithIdentity(packet, decoder, config, deviceID, stream, "periodic-"+frameID.String(), frameID.String(), config.Profile, "", now)
-}
-
-func prepareRequestedFrame(packet packets.Packet, decoder Decoder, config models.FrameProcessing, deviceID, stream string, request models.FrameProcessingRequest, now time.Time) (Frame, error) {
-	frameID, err := uuid.NewV4()
-	if err != nil {
-		return Frame{}, fmt.Errorf("generate frame ID: %w", err)
-	}
-	return prepareFrameWithIdentity(packet, decoder, config, deviceID, stream, request.RequestID, frameID.String(), request.ProcessingProfile, request.TraceID, now)
-}
-
-func prepareFrameWithIdentity(packet packets.Packet, decoder Decoder, config models.FrameProcessing, deviceID, stream, requestID, frameID, profile, traceID string, now time.Time) (Frame, error) {
-	decoded, err := decoder.DecodePacket(packet)
-	if err != nil {
-		return Frame{}, fmt.Errorf("decode keyframe: %w", err)
-	}
-	resized, err := utils.ResizeImage(&decoded, uint(config.Width), uint(config.Height))
-	if err != nil {
-		return Frame{}, fmt.Errorf("resize keyframe: %w", err)
-	}
-	var encoded bytes.Buffer
-	if err := jpeg.Encode(&encoded, *resized, &jpeg.Options{Quality: config.JPEGQuality}); err != nil {
-		return Frame{}, fmt.Errorf("encode keyframe: %w", err)
-	}
-	if int64(encoded.Len()) > config.MaxFrameBytes {
-		return Frame{}, fmt.Errorf("encoded keyframe exceeds frameProcessing.maxFrameBytes (%d)", config.MaxFrameBytes)
-	}
-	capturedAt := packet.CurrentTime
-	if capturedAt <= 0 {
-		capturedAt = now.UnixMilli()
-	}
-	bounds := (*resized).Bounds()
-	return Frame{
-		Metadata: Metadata{
-			SchemaVersion:     schemaVersion,
-			RequestID:         requestID,
-			FrameID:           frameID,
-			DeviceID:          deviceID,
-			CapturedAt:        capturedAt,
-			ExpiresAt:         now.Add(time.Duration(config.FrameTTLSeconds) * time.Second).UnixMilli(),
-			ProcessingProfile: profile,
-			SourceStream:      stream,
-			Width:             bounds.Dx(),
-			Height:            bounds.Dy(),
-			TraceID:           traceID,
-		},
-		JPEG: encoded.Bytes(),
-	}, nil
-}
-
-func nextKeyframe(ctx context.Context, cursor *packets.QueueCursor, expiresAt int64) (packets.Packet, error) {
-	requestContext, cancel := context.WithDeadline(ctx, time.UnixMilli(expiresAt))
-	defer cancel()
-	for {
-		packet, err := cursor.ReadPacketContext(requestContext)
-		if err != nil {
-			return packets.Packet{}, err
-		}
-		if packet.IsKeyFrame && len(packet.Data) > 0 {
-			return packet, nil
-		}
-	}
-}
-
-func publishStatus(ctx context.Context, publisher StatusPublisher, request models.FrameProcessingRequest, deviceID, frameID, status string, retryable bool, message string) {
-	if publisher == nil {
-		return
-	}
-	err := publisher.Publish(ctx, models.FrameProcessingStatus{
-		SchemaVersion: models.FrameProcessingSchemaVersion,
-		RequestID:     request.RequestID,
-		FrameID:       frameID,
-		DeviceID:      deviceID,
-		Status:        status,
-		OccurredAt:    time.Now().UnixMilli(),
-		Retryable:     retryable,
-		Message:       message,
-		TraceID:       request.TraceID,
-	})
-	if err != nil && ctx.Err() == nil {
-		log.WithError(err).WithFields(log.Fields{
-			"component":  "frame_processing",
-			"event":      "status_publish_failed",
-			"request_id": request.RequestID,
-			"status":     status,
-		}).Warn("Failed to publish frame-processing status")
-	}
-}
-
-func enqueueLatest(frames chan Frame, frame Frame) bool {
-	select {
-	case frames <- frame:
-		return false
-	default:
-	}
-	select {
-	case <-frames:
-	default:
-	}
-	frames <- frame
-	return true
-}
-
-func normalizeCancellation(ctx context.Context, err error) error {
-	if ctx.Err() != nil || errors.Is(err, context.Canceled) {
-		return nil
-	}
-	return err
-}
-
-func validateConfig(config models.FrameProcessing) error {
-	if config.IntervalSeconds <= 0 {
-		return errors.New("frameProcessing.intervalSeconds must be positive")
-	}
-	if config.Width <= 0 || config.Width > 8192 || config.Height < 0 || config.Height > 8192 {
-		return errors.New("frameProcessing dimensions must be between 0 and 8192, with a positive width")
-	}
-	if config.JPEGQuality < 1 || config.JPEGQuality > 100 {
-		return errors.New("frameProcessing.jpegQuality must be between 1 and 100")
-	}
-	if config.RequestTimeoutSeconds <= 0 || config.RequestTimeoutSeconds > 60 {
-		return errors.New("frameProcessing.requestTimeoutSeconds must be between 1 and 60")
-	}
-	if config.FrameTTLSeconds <= 0 || config.FrameTTLSeconds > 3600 {
-		return errors.New("frameProcessing.frameTtlSeconds must be between 1 and 3600")
-	}
-	if config.MaxFrameBytes <= 0 || config.MaxFrameBytes > 16<<20 {
-		return errors.New("frameProcessing.maxFrameBytes must be between 1 and 16777216")
-	}
-	if config.PeriodicQueueCapacity <= 0 || config.PeriodicQueueCapacity > 64 {
-		return errors.New("frameProcessing.periodicQueueCapacity must be between 1 and 64")
-	}
-	if config.Profile == "" {
-		return errors.New("frameProcessing.profile is required")
-	}
-	return nil
-}

machinery/src/cloud/frameprocessing/frameprocessing_test.go

@@ -1,285 +0,0 @@
-package frameprocessing
-
-import (
-	"bytes"
-	"context"
-	"encoding/json"
-	"image"
-	"image/color"
-	"io"
-	"mime/multipart"
-	"net/http"
-	"net/http/httptest"
-	"testing"
-	"time"
-
-	"github.com/kerberos-io/agent/machinery/src/models"
-	"github.com/kerberos-io/agent/machinery/src/packets"
-)
-
-type fakeDecoder struct{}
-
-func (fakeDecoder) DecodePacket(packets.Packet) (image.YCbCr, error) {
-	frame := image.NewYCbCr(image.Rect(0, 0, 4, 4), image.YCbCrSubsampleRatio420)
-	for index := range frame.Y {
-		frame.Y[index] = color.Gray{Y: 200}.Y
-	}
-	return *frame, nil
-}
-
-type fakeStatusPublisher struct {
-	statuses chan models.FrameProcessingStatus
-}
-
-func (p *fakeStatusPublisher) Publish(_ context.Context, status models.FrameProcessingStatus) error {
-	p.statuses <- status
-	return nil
-}
-
-func TestEnqueueLatestReplacesOldestFrame(t *testing.T) {
-	frames := make(chan Frame, 1)
-	frames <- Frame{Metadata: Metadata{FrameID: "old"}}
-
-	if dropped := enqueueLatest(frames, Frame{Metadata: Metadata{FrameID: "new"}}); !dropped {
-		t.Fatal("enqueueLatest() did not report dropping the stale frame")
-	}
-	if got := (<-frames).Metadata.FrameID; got != "new" {
-		t.Fatalf("queued frame = %q, want new", got)
-	}
-}
-
-func TestPrepareFrameUsesAgentCaptureTimestamp(t *testing.T) {
-	now := time.UnixMilli(2_000)
-	frame, err := prepareFrame(packets.Packet{CurrentTime: 1_500}, fakeDecoder{}, models.FrameProcessing{
-		Profile: "never-trigger", Width: 2, Height: 2, JPEGQuality: 70, FrameTTLSeconds: 30, MaxFrameBytes: 4 << 20,
-	}, "device-1", "sub", now)
-	if err != nil {
-		t.Fatal(err)
-	}
-	if frame.Metadata.CapturedAt != 1_500 || frame.Metadata.ExpiresAt != 32_000 {
-		t.Fatalf("metadata timestamps = %+v", frame.Metadata)
-	}
-	if frame.Metadata.Width != 2 || frame.Metadata.Height != 2 || len(frame.JPEG) == 0 {
-		t.Fatalf("prepared frame = %+v, bytes=%d", frame.Metadata, len(frame.JPEG))
-	}
-}
-
-func TestSenderSubmitsContractMultipartRequest(t *testing.T) {
-	var gotMetadata Metadata
-	var gotFrame []byte
-	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, request *http.Request) {
-		if got := request.Header.Get("Authorization"); got != "Bearer secret" {
-			t.Errorf("Authorization = %q", got)
-		}
-		reader, err := request.MultipartReader()
-		if err != nil {
-			t.Error(err)
-			w.WriteHeader(http.StatusBadRequest)
-			return
-		}
-		for {
-			part, err := reader.NextPart()
-			if err == io.EOF {
-				break
-			}
-			if err != nil {
-				t.Error(err)
-				return
-			}
-			switch part.FormName() {
-			case "metadata":
-				if err := json.NewDecoder(part).Decode(&gotMetadata); err != nil {
-					t.Error(err)
-				}
-			case "frame":
-				gotFrame, err = io.ReadAll(part)
-				if err != nil {
-					t.Error(err)
-				}
-			}
-		}
-		w.Header().Set("Content-Type", "application/json")
-		_, _ = w.Write([]byte(`{"decision":"no-event"}`))
-	}))
-	defer server.Close()
-
-	sender, err := NewSender(models.FrameProcessing{Endpoint: server.URL, Token: "secret", RequestTimeoutSeconds: 2})
-	if err != nil {
-		t.Fatal(err)
-	}
-	want := Frame{Metadata: Metadata{SchemaVersion: schemaVersion, FrameID: "frame-1"}, JPEG: []byte("jpeg")}
-	if err := sender.Submit(context.Background(), want); err != nil {
-		t.Fatal(err)
-	}
-	if gotMetadata.FrameID != want.Metadata.FrameID || !bytes.Equal(gotFrame, want.JPEG) {
-		t.Fatalf("submitted metadata=%+v frame=%q", gotMetadata, gotFrame)
-	}
-}
-
-func TestNewSenderRejectsRelativeEndpoint(t *testing.T) {
-	if _, err := NewSender(models.FrameProcessing{Endpoint: "/v1/frames", Token: "secret", RequestTimeoutSeconds: 1}); err == nil {
-		t.Fatal("NewSender() accepted a relative endpoint")
-	}
-}
-
-func TestValidateConfigRejectsUnboundedQueue(t *testing.T) {
-	config := models.FrameProcessing{
-		Profile: "never-trigger", IntervalSeconds: 10, Width: 640,
-		JPEGQuality: 70, RequestTimeoutSeconds: 5, FrameTTLSeconds: 30,
-		MaxFrameBytes: 4 << 20, PeriodicQueueCapacity: 65,
-	}
-	if err := validateConfig(config); err == nil {
-		t.Fatal("validateConfig() accepted an unbounded queue")
-	}
-}
-
-func TestMultipartContentTypeIsParseable(t *testing.T) {
-	var body bytes.Buffer
-	writer := multipart.NewWriter(&body)
-	if err := writer.Close(); err != nil {
-		t.Fatal(err)
-	}
-	request := httptest.NewRequest(http.MethodPost, "/", &body)
-	request.Header.Set("Content-Type", writer.FormDataContentType())
-	if _, err := request.MultipartReader(); err != nil {
-		t.Fatal(err)
-	}
-}
-
-func TestRunCancelsBlockedPacketRead(t *testing.T) {
-	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
-		w.WriteHeader(http.StatusOK)
-	}))
-	defer server.Close()
-
-	queue := packets.NewQueue()
-	defer queue.Close()
-	ctx, cancel := context.WithCancel(context.Background())
-	done := make(chan error, 1)
-	go func() {
-		done <- Run(ctx, queue.Latest(), fakeDecoder{}, models.FrameProcessing{
-			Enabled: "true", Endpoint: server.URL, Token: "secret", Profile: "never-trigger",
-			Stream: "main", IntervalSeconds: 10, Width: 640, JPEGQuality: 70,
-			RequestTimeoutSeconds: 2, FrameTTLSeconds: 30, MaxFrameBytes: 4 << 20,
-			PeriodicQueueCapacity: 1,
-		}, "device-1", "main", nil)
-	}()
-	cancel()
-	select {
-	case err := <-done:
-		if err != nil {
-			t.Fatalf("Run() error = %v", err)
-		}
-	case <-time.After(time.Second):
-		t.Fatal("Run() did not stop after cancellation")
-	}
-}
-
-func TestRunRequestedCapturesNextKeyframeAndSubmitsHTTP(t *testing.T) {
-	metadataReceived := make(chan Metadata, 1)
-	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, request *http.Request) {
-		reader, err := request.MultipartReader()
-		if err != nil {
-			t.Error(err)
-			return
-		}
-		for {
-			part, err := reader.NextPart()
-			if err == io.EOF {
-				break
-			}
-			if err != nil {
-				t.Error(err)
-				return
-			}
-			if part.FormName() == "metadata" {
-				var metadata Metadata
-				if err := json.NewDecoder(part).Decode(&metadata); err != nil {
-					t.Error(err)
-					return
-				}
-				metadataReceived <- metadata
-			}
-		}
-		w.WriteHeader(http.StatusOK)
-	}))
-	defer server.Close()
-
-	queue := packets.NewQueue()
-	defer queue.Close()
-	if err := queue.WriteHeader([]packets.Stream{{Index: 0, IsVideo: true}}); err != nil {
-		t.Fatal(err)
-	}
-	requests := make(chan models.FrameProcessingWork, 1)
-	statuses := &fakeStatusPublisher{statuses: make(chan models.FrameProcessingStatus, 4)}
-	ctx, cancel := context.WithCancel(context.Background())
-	done := make(chan error, 1)
-	config := models.FrameProcessing{
-		Enabled: "true", Endpoint: server.URL, Token: "secret", Profile: "never-trigger",
-		IntervalSeconds: 10, Width: 2, Height: 2, JPEGQuality: 70,
-		RequestTimeoutSeconds: 2, FrameTTLSeconds: 30, MaxFrameBytes: 4 << 20,
-		PeriodicQueueCapacity: 1,
-	}
-	go func() {
-		done <- RunRequested(ctx, fakeDecoder{}, config, "device-1", "sub", requests, statuses, nil)
-	}()
-	requests <- models.FrameProcessingWork{
-		Request: models.FrameProcessingRequest{
-			SchemaVersion: models.FrameProcessingSchemaVersion,
-			RequestID:     "request-1", ProcessingProfile: "always-trigger",
-			ExpiresAt: time.Now().Add(time.Second).UnixMilli(), TraceID: "trace-1",
-		},
-		Cursor: queue.LatestAtCurrentTail(),
-	}
-	queue.WritePacket(packets.Packet{Idx: 0, IsVideo: true, IsKeyFrame: true, CurrentTime: 1234, Data: []byte{1}})
-
-	metadata := <-metadataReceived
-	if metadata.RequestID != "request-1" || metadata.CapturedAt != 1234 || metadata.TraceID != "trace-1" {
-		t.Fatalf("submitted metadata = %+v", metadata)
-	}
-	status := <-statuses.statuses
-	if status.Status != "submitted" || status.FrameID == "" {
-		t.Fatalf("status = %+v", status)
-	}
-	cancel()
-	if err := <-done; err != nil {
-		t.Fatalf("RunRequested() error = %v", err)
-	}
-}
-
-func TestRunRequestedExpiresWhileWaitingForKeyframe(t *testing.T) {
-	queue := packets.NewQueue()
-	defer queue.Close()
-	requests := make(chan models.FrameProcessingWork, 1)
-	statuses := &fakeStatusPublisher{statuses: make(chan models.FrameProcessingStatus, 1)}
-	ctx, cancel := context.WithCancel(context.Background())
-	done := make(chan error, 1)
-	go func() {
-		done <- RunRequested(ctx, fakeDecoder{}, models.FrameProcessing{
-			Enabled: "true", Endpoint: "http://127.0.0.1:1/v1/frames", Token: "secret", Profile: "never-trigger",
-			IntervalSeconds: 10, Width: 2, Height: 2, JPEGQuality: 70,
-			RequestTimeoutSeconds: 1, FrameTTLSeconds: 30, MaxFrameBytes: 4 << 20,
-			PeriodicQueueCapacity: 1,
-		}, "device-1", "main", requests, statuses, nil)
-	}()
-	requests <- models.FrameProcessingWork{
-		Request: models.FrameProcessingRequest{
-			SchemaVersion: models.FrameProcessingSchemaVersion,
-			RequestID:     "request-expiring", ProcessingProfile: "never-trigger",
-			ExpiresAt: time.Now().Add(20 * time.Millisecond).UnixMilli(),
-		},
-		Cursor: queue.LatestAtCurrentTail(),
-	}
-	select {
-	case status := <-statuses.statuses:
-		if status.Status != "expired" {
-			t.Fatalf("status = %+v", status)
-		}
-	case <-time.After(time.Second):
-		t.Fatal("requested frame did not expire while waiting for a keyframe")
-	}
-	cancel()
-	if err := <-done; err != nil {
-		t.Fatalf("RunRequested() error = %v", err)
-	}
-}

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/components/kerberos.go

@@ -14,7 +14,6 @@ import (
 
 	"github.com/kerberos-io/agent/machinery/src/capture"
 	"github.com/kerberos-io/agent/machinery/src/cloud"
-	"github.com/kerberos-io/agent/machinery/src/cloud/frameprocessing"
 	"github.com/kerberos-io/agent/machinery/src/computervision"
 	configService "github.com/kerberos-io/agent/machinery/src/config"
 	"github.com/kerberos-io/agent/machinery/src/lifecycle"
@@ -559,74 +558,6 @@ func RunAgent(parent context.Context, configDirectory string, configuration *mod
 		})
 	}
 
-	// Frame Processing is the HTTP-based successor to the legacy MQTT
-	// realtime-processing output. Both remain independently configurable during
-	// the compatibility period.
-	frameProcessingConfig := configuration.Config.FrameProcessing
-	if frameProcessingConfig != nil && frameProcessingConfig.Enabled == "true" && configuration.Config.Offline != "true" {
-		selectedCursor := queue.Latest()
-		selectedRequestQueue := queue
-		selectedClient := rtspClient
-		selectedStream := "main"
-		selectionError := error(nil)
-		switch frameProcessingConfig.Stream {
-		case "auto", "":
-			if subStreamEnabled && rtspSubClient != nil && subQueue != nil {
-				selectedCursor = subQueue.Latest()
-				selectedRequestQueue = subQueue
-				selectedClient = rtspSubClient
-				selectedStream = "sub"
-			}
-		case "main":
-		case "sub":
-			if !subStreamEnabled || rtspSubClient == nil || subQueue == nil {
-				selectionError = errors.New("frameProcessing.stream is sub but no substream is available")
-			} else {
-				selectedCursor = subQueue.Latest()
-				selectedRequestQueue = subQueue
-				selectedClient = rtspSubClient
-				selectedStream = "sub"
-			}
-		default:
-			selectionError = fmt.Errorf("unsupported frameProcessing.stream %q", frameProcessingConfig.Stream)
-		}
-
-		registerTask("frame-processing", lifecycle.TaskPolicy{}, func(taskContext context.Context) error {
-			if selectionError != nil {
-				return selectionError
-			}
-			return frameprocessing.Run(
-				taskContext,
-				selectedCursor,
-				selectedClient,
-				*frameProcessingConfig,
-				configuration.Config.Key,
-				selectedStream,
-				communication,
-			)
-		})
-		if frameProcessingConfig.AllowRequestedFrames == "true" {
-			frameProcessingRequests := run.FrameProcessingRequests()
-			run.SetFrameProcessingQueue(selectedRequestQueue)
-			frameProcessingStatusPublisher := frameprocessing.NewMQTTStatusPublisher(mqttClient, config.HubKey, configuration)
-			registerTask("frame-processing-requested", lifecycle.TaskPolicy{}, func(taskContext context.Context) error {
-				if selectionError != nil {
-					return selectionError
-				}
-				return frameprocessing.RunRequested(
-					taskContext,
-					selectedClient,
-					*frameProcessingConfig,
-					configuration.Config.Key,
-					selectedStream,
-					frameProcessingRequests,
-					frameProcessingStatusPublisher,
-					communication,
-				)
-			})
-		}
-	}
-
 	// Handle Upload to cloud provider (Kerberos Hub, Kerberos Vault and others)
 	registerTask("upload", lifecycle.TaskPolicy{}, func(context.Context) error {
 		cloud.HandleUpload(configDirectory, configuration, communication)

machinery/src/config/main.go

@@ -300,9 +300,6 @@ func initConfigPointers(config *models.Config) {
 	if config.Signing == nil {
 		config.Signing = &models.Signing{}
 	}
-	if config.FrameProcessing == nil {
-		config.FrameProcessing = &models.FrameProcessing{}
-	}
 	if config.Dropbox == nil {
 		config.Dropbox = &models.Dropbox{}
 	}
@@ -318,13 +315,19 @@ func initConfigPointers(config *models.Config) {
 func applyAgentEnvVars(configuration *models.Configuration, prefix string, applyDefaults bool) {
 	environmentVariables := os.Environ()
 
+	if prefix == "" {
+		if deploymentName := os.Getenv("DEPLOYMENT_NAME"); deploymentName != "" {
+			configuration.Config.Name = deploymentName
+			if configuration.Config.FriendlyName == "" {
+				configuration.Config.FriendlyName = deploymentName
+			}
+		}
+	}
+
 	// Initialize the configuration for some new fields.
 	if configuration.Config.KStorageSecondary == nil {
 		configuration.Config.KStorageSecondary = &models.KStorage{}
 	}
-	if configuration.Config.FrameProcessing == nil {
-		configuration.Config.FrameProcessing = &models.FrameProcessing{}
-	}
 
 	for _, env := range environmentVariables {
 		fullKey := strings.SplitN(env, "=", 2)[0]
@@ -547,66 +550,6 @@ func applyAgentEnvVars(configuration *models.Configuration, prefix string, apply
 				configuration.Config.RealtimeProcessingTopic = value
 				break
 
-			/* Keyframe-aligned HTTP frame processing */
-			case "AGENT_FRAME_PROCESSING_ENABLED":
-				configuration.Config.FrameProcessing.Enabled = value
-				break
-			case "AGENT_FRAME_PROCESSING_ENDPOINT":
-				configuration.Config.FrameProcessing.Endpoint = value
-				break
-			case "AGENT_FRAME_PROCESSING_TOKEN":
-				configuration.Config.FrameProcessing.Token = value
-				break
-			case "AGENT_FRAME_PROCESSING_PROFILE":
-				configuration.Config.FrameProcessing.Profile = value
-				break
-			case "AGENT_FRAME_PROCESSING_ALLOW_REQUESTED_FRAMES":
-				configuration.Config.FrameProcessing.AllowRequestedFrames = value
-				break
-			case "AGENT_FRAME_PROCESSING_STREAM":
-				configuration.Config.FrameProcessing.Stream = value
-				break
-			case "AGENT_FRAME_PROCESSING_INTERVAL_SECONDS":
-				if interval, err := strconv.ParseInt(value, 10, 64); err == nil {
-					configuration.Config.FrameProcessing.IntervalSeconds = interval
-				}
-				break
-			case "AGENT_FRAME_PROCESSING_WIDTH":
-				if width, err := strconv.Atoi(value); err == nil {
-					configuration.Config.FrameProcessing.Width = width
-				}
-				break
-			case "AGENT_FRAME_PROCESSING_HEIGHT":
-				if height, err := strconv.Atoi(value); err == nil {
-					configuration.Config.FrameProcessing.Height = height
-				}
-				break
-			case "AGENT_FRAME_PROCESSING_JPEG_QUALITY":
-				if quality, err := strconv.Atoi(value); err == nil {
-					configuration.Config.FrameProcessing.JPEGQuality = quality
-				}
-				break
-			case "AGENT_FRAME_PROCESSING_REQUEST_TIMEOUT_SECONDS":
-				if timeout, err := strconv.ParseInt(value, 10, 64); err == nil {
-					configuration.Config.FrameProcessing.RequestTimeoutSeconds = timeout
-				}
-				break
-			case "AGENT_FRAME_PROCESSING_FRAME_TTL_SECONDS":
-				if ttl, err := strconv.ParseInt(value, 10, 64); err == nil {
-					configuration.Config.FrameProcessing.FrameTTLSeconds = ttl
-				}
-				break
-			case "AGENT_FRAME_PROCESSING_MAX_FRAME_BYTES":
-				if maxBytes, err := strconv.ParseInt(value, 10, 64); err == nil {
-					configuration.Config.FrameProcessing.MaxFrameBytes = maxBytes
-				}
-				break
-			case "AGENT_FRAME_PROCESSING_PERIODIC_QUEUE_CAPACITY":
-				if capacity, err := strconv.Atoi(value); err == nil {
-					configuration.Config.FrameProcessing.PeriodicQueueCapacity = capacity
-				}
-				break
-
 			/* WebRTC settings for live-streaming (remote) */
 			case "AGENT_STUN_URI":
 				configuration.Config.STUNURI = value
@@ -669,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":
@@ -700,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":
@@ -749,41 +698,6 @@ func applyAgentEnvVars(configuration *models.Configuration, prefix string, apply
 		configuration.Config.Capture.PixelChangeThreshold = &defaultPixelChangeThreshold
 	}
 
-	if applyDefaults {
-		frameProcessing := configuration.Config.FrameProcessing
-		if frameProcessing == nil {
-			frameProcessing = &models.FrameProcessing{}
-			configuration.Config.FrameProcessing = frameProcessing
-		}
-		if frameProcessing.Profile == "" {
-			frameProcessing.Profile = "never-trigger"
-		}
-		if frameProcessing.Stream == "" {
-			frameProcessing.Stream = "auto"
-		}
-		if frameProcessing.IntervalSeconds <= 0 {
-			frameProcessing.IntervalSeconds = 10
-		}
-		if frameProcessing.Width <= 0 {
-			frameProcessing.Width = 640
-		}
-		if frameProcessing.JPEGQuality <= 0 || frameProcessing.JPEGQuality > 100 {
-			frameProcessing.JPEGQuality = 70
-		}
-		if frameProcessing.RequestTimeoutSeconds <= 0 {
-			frameProcessing.RequestTimeoutSeconds = 5
-		}
-		if frameProcessing.FrameTTLSeconds <= 0 {
-			frameProcessing.FrameTTLSeconds = 30
-		}
-		if frameProcessing.MaxFrameBytes <= 0 {
-			frameProcessing.MaxFrameBytes = 4 << 20
-		}
-		if frameProcessing.PeriodicQueueCapacity <= 0 {
-			frameProcessing.PeriodicQueueCapacity = 1
-		}
-	}
-
 	// Signing is a new feature, so if empty we set default values. Only applied
 	// for the effective configuration (applyDefaults), not for the separate
 	// global/custom views.

machinery/src/config/main_test.go

@@ -3,6 +3,7 @@ package config
 import (
 	"context"
 	"errors"
+	"os"
 	"testing"
 	"time"
 
@@ -11,6 +12,53 @@ import (
 	"go.mongodb.org/mongo-driver/mongo"
 )
 
+func TestApplyAgentEnvVarsDeploymentName(t *testing.T) {
+	tests := []struct {
+		name           string
+		prefix         string
+		deploymentName string
+		agentName      string
+		initialName    string
+		initialDisplay string
+		wantName       string
+		wantDisplay    string
+	}{
+		{name: "deployment fallback", deploymentName: "camera-1", wantName: "camera-1", wantDisplay: "camera-1"},
+		{name: "explicit display name", deploymentName: "camera-1", agentName: "Front Door", wantName: "camera-1", wantDisplay: "Front Door"},
+		{name: "existing display name", deploymentName: "camera-1", initialDisplay: "Front Door", wantName: "camera-1", wantDisplay: "Front Door"},
+		{name: "deployment overrides bundled identity", deploymentName: "camera-1", initialName: "default", wantName: "camera-1", wantDisplay: "camera-1"},
+		{name: "no deployment", initialName: "existing", initialDisplay: "Existing Camera", wantName: "existing", wantDisplay: "Existing Camera"},
+		{name: "standalone display name", agentName: "Front Door", wantDisplay: "Front Door"},
+		{name: "global layer excludes deployment", prefix: "GLOBAL_", deploymentName: "camera-1"},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			t.Setenv("DEPLOYMENT_NAME", test.deploymentName)
+			t.Setenv(test.prefix+"AGENT_NAME", test.agentName)
+			if test.agentName == "" {
+				if err := os.Unsetenv(test.prefix + "AGENT_NAME"); err != nil {
+					t.Fatal(err)
+				}
+			}
+			configuration := &models.Configuration{Config: models.Config{
+				Name:         test.initialName,
+				FriendlyName: test.initialDisplay,
+			}}
+			initConfigPointers(&configuration.Config)
+
+			applyAgentEnvVars(configuration, test.prefix, false)
+
+			if got := configuration.Config.Name; got != test.wantName {
+				t.Errorf("Name = %q, want %q", got, test.wantName)
+			}
+			if got := configuration.Config.FriendlyName; got != test.wantDisplay {
+				t.Errorf("FriendlyName = %q, want %q", got, test.wantDisplay)
+			}
+		})
+	}
+}
+
 func TestApplyAgentEnvVarsPixelChangeThresholdDefault(t *testing.T) {
 	tests := []struct {
 		name      string
@@ -40,82 +88,26 @@ 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
 }
 
-func TestApplyAgentEnvVarsFrameProcessing(t *testing.T) {
-	t.Setenv("AGENT_FRAME_PROCESSING_ENABLED", "true")
-	t.Setenv("AGENT_FRAME_PROCESSING_ENDPOINT", "http://processor:8080/v1/frames")
-	t.Setenv("AGENT_FRAME_PROCESSING_TOKEN", "secret")
-	t.Setenv("AGENT_FRAME_PROCESSING_PROFILE", "always-trigger")
-	t.Setenv("AGENT_FRAME_PROCESSING_ALLOW_REQUESTED_FRAMES", "true")
-	t.Setenv("AGENT_FRAME_PROCESSING_STREAM", "sub")
-	t.Setenv("AGENT_FRAME_PROCESSING_INTERVAL_SECONDS", "15")
-	t.Setenv("AGENT_FRAME_PROCESSING_WIDTH", "320")
-	t.Setenv("AGENT_FRAME_PROCESSING_HEIGHT", "180")
-	t.Setenv("AGENT_FRAME_PROCESSING_JPEG_QUALITY", "80")
-	t.Setenv("AGENT_FRAME_PROCESSING_REQUEST_TIMEOUT_SECONDS", "7")
-	t.Setenv("AGENT_FRAME_PROCESSING_FRAME_TTL_SECONDS", "45")
-	t.Setenv("AGENT_FRAME_PROCESSING_MAX_FRAME_BYTES", "2097152")
-	t.Setenv("AGENT_FRAME_PROCESSING_PERIODIC_QUEUE_CAPACITY", "2")
-
-	configuration := &models.Configuration{}
-	initConfigPointers(&configuration.Config)
-	applyAgentEnvVars(configuration, "", true)
-
-	got := configuration.Config.FrameProcessing
-	if got == nil {
-		t.Fatal("FrameProcessing is nil")
-	}
-	if got.Enabled != "true" || got.Endpoint != "http://processor:8080/v1/frames" || got.Token != "secret" {
-		t.Fatalf("FrameProcessing identity = %+v", got)
-	}
-	if got.Profile != "always-trigger" || got.AllowRequestedFrames != "true" || got.Stream != "sub" || got.IntervalSeconds != 15 {
-		t.Fatalf("FrameProcessing schedule = %+v", got)
-	}
-	if got.Width != 320 || got.Height != 180 || got.JPEGQuality != 80 {
-		t.Fatalf("FrameProcessing image = %+v", got)
-	}
-	if got.RequestTimeoutSeconds != 7 || got.FrameTTLSeconds != 45 || got.MaxFrameBytes != 2097152 || got.PeriodicQueueCapacity != 2 {
-		t.Fatalf("FrameProcessing delivery = %+v", got)
-	}
-}
-
-func TestApplyAgentEnvVarsFrameProcessingDefaults(t *testing.T) {
-	configuration := &models.Configuration{}
-	initConfigPointers(&configuration.Config)
-	applyAgentEnvVars(configuration, "", true)
-
-	got := configuration.Config.FrameProcessing
-	if got.Profile != "never-trigger" || got.Stream != "auto" || got.IntervalSeconds != 10 {
-		t.Fatalf("FrameProcessing defaults = %+v", got)
-	}
-	if got.Width != 640 || got.Height != 0 || got.JPEGQuality != 70 {
-		t.Fatalf("FrameProcessing image defaults = %+v", got)
-	}
-	if got.RequestTimeoutSeconds != 5 || got.FrameTTLSeconds != 30 || got.MaxFrameBytes != 4<<20 || got.PeriodicQueueCapacity != 1 {
-		t.Fatalf("FrameProcessing delivery defaults = %+v", got)
-	}
-}
-
-func TestOverrideWithEnvironmentVariablesInheritsGlobalFrameProcessing(t *testing.T) {
-	t.Setenv("GLOBAL_AGENT_FRAME_PROCESSING_ENABLED", "true")
-	t.Setenv("GLOBAL_AGENT_FRAME_PROCESSING_ENDPOINT", "https://processor.example/v1/frames")
-	t.Setenv("GLOBAL_AGENT_FRAME_PROCESSING_PROFILE", "never-trigger")
-
-	configuration := &models.Configuration{}
-	OverrideWithEnvironmentVariables(configuration)
-
-	got := configuration.Config.FrameProcessing
-	if got == nil || got.Enabled != "true" || got.Endpoint != "https://processor.example/v1/frames" {
-		t.Fatalf("effective FrameProcessing = %+v", got)
-	}
-	if configuration.CustomConfig.FrameProcessing == nil || configuration.CustomConfig.FrameProcessing.Enabled != "" {
-		t.Fatalf("custom FrameProcessing unexpectedly overrides global config: %+v", configuration.CustomConfig.FrameProcessing)
-	}
-}
-
 func TestNewFactoryConfigReadContextUsesDatabaseTimeout(t *testing.T) {
 	ctx, cancel := newFactoryConfigReadContext()
 	defer cancel()

machinery/src/models/agent_run.go

@@ -20,8 +20,6 @@ var (
 	nextAgentRunID     atomic.Uint64
 )
 
-const defaultFrameProcessingRequestCapacity = 8
-
 type AgentRunClient interface {
 	Close(context.Context) error
 }
@@ -58,21 +56,19 @@ type AgentRun struct {
 	activated bool
 	stopping  bool
 
-	resourcesMu          sync.RWMutex
-	mainClient           AgentRunClient
-	subClient            AgentRunClient
-	backchannelClient    AgentRunClient
-	mainQueue            *packets.Queue
-	subQueue             *packets.Queue
-	frameProcessingQueue *packets.Queue
-	releaseClients       func()
+	resourcesMu       sync.RWMutex
+	mainClient        AgentRunClient
+	subClient         AgentRunClient
+	backchannelClient AgentRunClient
+	mainQueue         *packets.Queue
+	subQueue          *packets.Queue
+	releaseClients    func()
 
-	channelsMu              sync.RWMutex
-	channelsClosed          bool
-	liveHDHandshakes        chan LiveHDHandshake
-	motionEvents            chan MotionDataPartial
-	onvifActions            chan OnvifAction
-	frameProcessingRequests chan FrameProcessingWork
+	channelsMu       sync.RWMutex
+	channelsClosed   bool
+	liveHDHandshakes chan LiveHDHandshake
+	motionEvents     chan MotionDataPartial
+	onvifActions     chan OnvifAction
 
 	shutdownOnce   sync.Once
 	shutdownReport AgentRunShutdownReport
@@ -84,16 +80,15 @@ func NewAgentRun(parent context.Context, communication *Communication, stopUploa
 	}
 	ctx, cancel := context.WithCancelCause(parent)
 	run := &AgentRun{
-		id:                      nextAgentRunID.Add(1),
-		ctx:                     ctx,
-		cancel:                  cancel,
-		supervisor:              lifecycle.NewSupervisor(ctx),
-		communication:           communication,
-		stopUpload:              stopUpload,
-		liveHDHandshakes:        make(chan LiveHDHandshake, 100),
-		motionEvents:            make(chan MotionDataPartial, 10),
-		onvifActions:            make(chan OnvifAction, 10),
-		frameProcessingRequests: make(chan FrameProcessingWork, defaultFrameProcessingRequestCapacity),
+		id:               nextAgentRunID.Add(1),
+		ctx:              ctx,
+		cancel:           cancel,
+		supervisor:       lifecycle.NewSupervisor(ctx),
+		communication:    communication,
+		stopUpload:       stopUpload,
+		liveHDHandshakes: make(chan LiveHDHandshake, 100),
+		motionEvents:     make(chan MotionDataPartial, 10),
+		onvifActions:     make(chan OnvifAction, 10),
 	}
 	log.WithFields(log.Fields{
 		"component":   "agent_run",
@@ -238,12 +233,6 @@ func (r *AgentRun) SetSubQueue(queue *packets.Queue) {
 	r.resourcesMu.Unlock()
 }
 
-func (r *AgentRun) SetFrameProcessingQueue(queue *packets.Queue) {
-	r.resourcesMu.Lock()
-	r.frameProcessingQueue = queue
-	r.resourcesMu.Unlock()
-}
-
 func (r *AgentRun) SetClientRelease(release func()) {
 	r.resourcesMu.Lock()
 	r.releaseClients = release
@@ -280,10 +269,6 @@ func (r *AgentRun) ONVIFActions() <-chan OnvifAction {
 	return r.onvifActions
 }
 
-func (r *AgentRun) FrameProcessingRequests() <-chan FrameProcessingWork {
-	return r.frameProcessingRequests
-}
-
 func (r *AgentRun) TrySendLiveHDHandshake(handshake LiveHDHandshake) bool {
 	if r.isStopping() {
 		return false
@@ -347,30 +332,6 @@ func (r *AgentRun) TrySendONVIF(action OnvifAction) bool {
 	}
 }
 
-func (r *AgentRun) TrySendFrameProcessingRequest(request FrameProcessingRequest) bool {
-	if r.isStopping() {
-		return false
-	}
-	r.channelsMu.RLock()
-	defer r.channelsMu.RUnlock()
-	if r.channelsClosed {
-		return false
-	}
-	r.resourcesMu.RLock()
-	queue := r.frameProcessingQueue
-	r.resourcesMu.RUnlock()
-	if queue == nil {
-		return false
-	}
-	work := FrameProcessingWork{Request: request, Cursor: queue.LatestAtCurrentTail()}
-	select {
-	case r.frameProcessingRequests <- work:
-		return true
-	default:
-		return false
-	}
-}
-
 func (r *AgentRun) Shutdown(ctx context.Context, cause error) AgentRunShutdownReport {
 	if ctx == nil {
 		ctx = context.Background()
@@ -497,7 +458,6 @@ func (r *AgentRun) closeChannels() {
 	close(r.liveHDHandshakes)
 	close(r.motionEvents)
 	close(r.onvifActions)
-	close(r.frameProcessingRequests)
 }
 
 func sendRunStop(ctx context.Context, channel chan<- string) bool {

machinery/src/models/agent_run_test.go

@@ -112,31 +112,6 @@ func TestAgentRunOwnsAndShutsDownResources(t *testing.T) {
 	if _, ok := <-run.ONVIFActions(); ok {
 		t.Fatal("ONVIF channel remained open")
 	}
-	if _, ok := <-run.FrameProcessingRequests(); ok {
-		t.Fatal("frame-processing request channel remained open")
-	}
-}
-
-func TestAgentRunBoundsFrameProcessingRequests(t *testing.T) {
-	communication := &Communication{}
-	run := NewAgentRun(context.Background(), communication, false)
-	queue := packets.NewQueue()
-	run.SetMainQueue(queue)
-	run.SetFrameProcessingQueue(queue)
-	if err := run.Activate(); err != nil {
-		t.Fatal(err)
-	}
-	run.Seal()
-	t.Cleanup(func() { run.Shutdown(context.Background(), errors.New("test complete")) })
-
-	for index := 0; index < defaultFrameProcessingRequestCapacity; index++ {
-		if !communication.TrySendFrameProcessingRequest(FrameProcessingRequest{RequestID: "request"}) {
-			t.Fatalf("request %d was rejected before the queue was full", index)
-		}
-	}
-	if communication.TrySendFrameProcessingRequest(FrameProcessingRequest{RequestID: "overflow"}) {
-		t.Fatal("overflow request was accepted")
-	}
 }
 
 func TestAgentRunShutdownIsConcurrentAndIdempotent(t *testing.T) {

machinery/src/models/communication.go

@@ -71,17 +71,6 @@ type HubRuntimeTelemetry struct {
 	LastSuccessfulHeartbeatAt int64
 }
 
-type FrameProcessingRuntimeTelemetry struct {
-	Configured    bool   `json:"configured"`
-	Sampled       uint64 `json:"sampled"`
-	Queued        uint64 `json:"queued"`
-	Dropped       uint64 `json:"dropped"`
-	Submitted     uint64 `json:"submitted"`
-	Failed        uint64 `json:"failed"`
-	QueueDepth    int64  `json:"queueDepth"`
-	LastSuccessAt int64  `json:"lastSuccessAt"`
-}
-
 type hubRuntimeTelemetry struct {
 	configured                atomic.Bool
 	connected                 atomic.Bool
@@ -89,17 +78,6 @@ type hubRuntimeTelemetry struct {
 	lastSuccessfulHeartbeatAt atomic.Int64
 }
 
-type frameProcessingRuntimeTelemetry struct {
-	configured    atomic.Bool
-	sampled       atomic.Uint64
-	queued        atomic.Uint64
-	dropped       atomic.Uint64
-	submitted     atomic.Uint64
-	failed        atomic.Uint64
-	queueDepth    atomic.Int64
-	lastSuccessAt atomic.Int64
-}
-
 type recoveryTelemetry struct {
 	moqHighReconnects          atomic.Uint64
 	moqHighLastFrameUnixMillis atomic.Int64
@@ -173,7 +151,6 @@ type Communication struct {
 	mainStreamTelemetry          streamRuntimeTelemetry
 	subStreamTelemetry           streamRuntimeTelemetry
 	hubTelemetry                 hubRuntimeTelemetry
-	frameProcessingTelemetry     frameProcessingRuntimeTelemetry
 	recovery                     recoveryTelemetry
 }
 
@@ -254,54 +231,6 @@ func (c *Communication) HubRuntimeTelemetry() HubRuntimeTelemetry {
 	}
 }
 
-func (c *Communication) SetFrameProcessingConfigured(configured bool) {
-	c.frameProcessingTelemetry.configured.Store(configured)
-	if !configured {
-		c.frameProcessingTelemetry.queueDepth.Store(0)
-	}
-}
-
-func (c *Communication) RecordFrameProcessingSample() {
-	c.frameProcessingTelemetry.sampled.Add(1)
-}
-
-func (c *Communication) RecordFrameProcessingQueued(depth int, dropped bool) {
-	c.frameProcessingTelemetry.queueDepth.Store(int64(depth))
-	c.frameProcessingTelemetry.queued.Add(1)
-	if dropped {
-		c.frameProcessingTelemetry.dropped.Add(1)
-	}
-}
-
-func (c *Communication) SetFrameProcessingQueueDepth(depth int) {
-	c.frameProcessingTelemetry.queueDepth.Store(int64(depth))
-}
-
-func (c *Communication) RecordFrameProcessingSuccess(at time.Time) {
-	c.frameProcessingTelemetry.submitted.Add(1)
-	if !at.IsZero() {
-		c.frameProcessingTelemetry.lastSuccessAt.Store(at.Unix())
-	}
-}
-
-func (c *Communication) RecordFrameProcessingFailure() {
-	c.frameProcessingTelemetry.failed.Add(1)
-}
-
-func (c *Communication) FrameProcessingRuntimeTelemetry() FrameProcessingRuntimeTelemetry {
-	telemetry := &c.frameProcessingTelemetry
-	return FrameProcessingRuntimeTelemetry{
-		Configured:    telemetry.configured.Load(),
-		Sampled:       telemetry.sampled.Load(),
-		Queued:        telemetry.queued.Load(),
-		Dropped:       telemetry.dropped.Load(),
-		Submitted:     telemetry.submitted.Load(),
-		Failed:        telemetry.failed.Load(),
-		QueueDepth:    telemetry.queueDepth.Load(),
-		LastSuccessAt: telemetry.lastSuccessAt.Load(),
-	}
-}
-
 func (c *Communication) RecordMoQReconnect(quality string) {
 	if quality == StreamQualityLow {
 		c.recovery.moqLowReconnects.Add(1)
@@ -435,8 +364,3 @@ func (c *Communication) TrySendONVIF(action OnvifAction) bool {
 	}
 	return true
 }
-
-func (c *Communication) TrySendFrameProcessingRequest(request FrameProcessingRequest) bool {
-	run := c.CurrentRun()
-	return run != nil && run.TrySendFrameProcessingRequest(request)
-}

machinery/src/models/config.go

@@ -12,64 +12,44 @@ type Configuration struct {
 // Config is the highlevel struct which contains all the configuration of
 // your Kerberos Open Source instance.
 type Config struct {
-	Type                    string           `json:"type"`
-	Key                     string           `json:"key"`
-	Name                    string           `json:"name"`
-	FriendlyName            string           `json:"friendly_name"`
-	Time                    string           `json:"time" bson:"time"`
-	Offline                 string           `json:"offline"`
-	AutoClean               string           `json:"auto_clean"`
-	RemoveAfterUpload       string           `json:"remove_after_upload"`
-	MaxDirectorySize        int64            `json:"max_directory_size"`
-	MinFreeSpace            int64            `json:"min_free_space,omitempty"`
-	Timezone                string           `json:"timezone"`
-	Capture                 Capture          `json:"capture"`
-	Timetable               []*Timetable     `json:"timetable"`
-	Region                  *Region          `json:"region"`
-	Cloud                   string           `json:"cloud" bson:"cloud"`
-	S3                      *S3              `json:"s3,omitempty" bson:"s3,omitempty"`
-	KStorage                *KStorage        `json:"kstorage,omitempty" bson:"kstorage,omitempty"`
-	KStorageSecondary       *KStorage        `json:"kstorage_secondary,omitempty" bson:"kstorage_secondary,omitempty"`
-	Dropbox                 *Dropbox         `json:"dropbox,omitempty" bson:"dropbox,omitempty"`
-	MQTTURI                 string           `json:"mqtturi" bson:"mqtturi,omitempty"`
-	MQTTUsername            string           `json:"mqtt_username" bson:"mqtt_username"`
-	MQTTPassword            string           `json:"mqtt_password" bson:"mqtt_password"`
-	STUNURI                 string           `json:"stunuri" bson:"stunuri"`
-	ForceTurn               string           `json:"turn_force" bson:"turn_force"`
-	TURNURI                 string           `json:"turnuri" bson:"turnuri"`
-	TURNUsername            string           `json:"turn_username" bson:"turn_username"`
-	TURNPassword            string           `json:"turn_password" bson:"turn_password"`
-	HeartbeatURI            string           `json:"heartbeaturi" bson:"heartbeaturi"` /*obsolete*/
-	HubEncryption           string           `json:"hub_encryption" bson:"hub_encryption"`
-	HubURI                  string           `json:"hub_uri" bson:"hub_uri"`
-	HubKey                  string           `json:"hub_key" bson:"hub_key"`
-	HubPrivateKey           string           `json:"hub_private_key" bson:"hub_private_key"`
-	HubSite                 string           `json:"hub_site" bson:"hub_site"`
-	ConditionURI            string           `json:"condition_uri" bson:"condition_uri"`
-	Encryption              *Encryption      `json:"encryption,omitempty" bson:"encryption,omitempty"`
-	Signing                 *Signing         `json:"signing,omitempty" bson:"signing,omitempty"`
-	FrameProcessing         *FrameProcessing `json:"frameProcessing,omitempty" bson:"frameProcessing,omitempty"`
-	RealtimeProcessing      string           `json:"realtimeprocessing,omitempty" bson:"realtimeprocessing,omitempty"`
-	RealtimeProcessingTopic string           `json:"realtimeprocessing_topic" bson:"realtimeprocessing_topic"`
-}
-
-// FrameProcessing configures keyframe-aligned JPEG delivery to an external
-// processor. It is independent from the legacy MQTT realtimeprocessing output.
-type FrameProcessing struct {
-	Enabled               string `json:"enabled,omitempty" bson:"enabled,omitempty"`
-	Endpoint              string `json:"endpoint,omitempty" bson:"endpoint,omitempty"`
-	Token                 string `json:"-" bson:"-"`
-	Profile               string `json:"profile,omitempty" bson:"profile,omitempty"`
-	AllowRequestedFrames  string `json:"allowRequestedFrames,omitempty" bson:"allowRequestedFrames,omitempty"`
-	Stream                string `json:"stream,omitempty" bson:"stream,omitempty"`
-	IntervalSeconds       int64  `json:"intervalSeconds,omitempty" bson:"intervalSeconds,omitempty"`
-	Width                 int    `json:"width,omitempty" bson:"width,omitempty"`
-	Height                int    `json:"height,omitempty" bson:"height,omitempty"`
-	JPEGQuality           int    `json:"jpegQuality,omitempty" bson:"jpegQuality,omitempty"`
-	RequestTimeoutSeconds int64  `json:"requestTimeoutSeconds,omitempty" bson:"requestTimeoutSeconds,omitempty"`
-	FrameTTLSeconds       int64  `json:"frameTtlSeconds,omitempty" bson:"frameTtlSeconds,omitempty"`
-	MaxFrameBytes         int64  `json:"maxFrameBytes,omitempty" bson:"maxFrameBytes,omitempty"`
-	PeriodicQueueCapacity int    `json:"periodicQueueCapacity,omitempty" bson:"periodicQueueCapacity,omitempty"`
+	Type                    string       `json:"type"`
+	Key                     string       `json:"key"`
+	Name                    string       `json:"name"`
+	FriendlyName            string       `json:"friendly_name"`
+	Time                    string       `json:"time" bson:"time"`
+	Offline                 string       `json:"offline"`
+	AutoClean               string       `json:"auto_clean"`
+	RemoveAfterUpload       string       `json:"remove_after_upload"`
+	MaxDirectorySize        int64        `json:"max_directory_size"`
+	MinFreeSpace            int64        `json:"min_free_space,omitempty"`
+	Timezone                string       `json:"timezone"`
+	Capture                 Capture      `json:"capture"`
+	Timetable               []*Timetable `json:"timetable"`
+	Region                  *Region      `json:"region"`
+	Cloud                   string       `json:"cloud" bson:"cloud"`
+	S3                      *S3          `json:"s3,omitempty" bson:"s3,omitempty"`
+	KStorage                *KStorage    `json:"kstorage,omitempty" bson:"kstorage,omitempty"`
+	KStorageSecondary       *KStorage    `json:"kstorage_secondary,omitempty" bson:"kstorage_secondary,omitempty"`
+	Dropbox                 *Dropbox     `json:"dropbox,omitempty" bson:"dropbox,omitempty"`
+	MQTTURI                 string       `json:"mqtturi" bson:"mqtturi,omitempty"`
+	MQTTUsername            string       `json:"mqtt_username" bson:"mqtt_username"`
+	MQTTPassword            string       `json:"mqtt_password" bson:"mqtt_password"`
+	STUNURI                 string       `json:"stunuri" bson:"stunuri"`
+	ForceTurn               string       `json:"turn_force" bson:"turn_force"`
+	TURNURI                 string       `json:"turnuri" bson:"turnuri"`
+	TURNUsername            string       `json:"turn_username" bson:"turn_username"`
+	TURNPassword            string       `json:"turn_password" bson:"turn_password"`
+	HeartbeatURI            string       `json:"heartbeaturi" bson:"heartbeaturi"` /*obsolete*/
+	HubEncryption           string       `json:"hub_encryption" bson:"hub_encryption"`
+	HubURI                  string       `json:"hub_uri" bson:"hub_uri"`
+	HubKey                  string       `json:"hub_key" bson:"hub_key"`
+	HubPrivateKey           string       `json:"hub_private_key" bson:"hub_private_key"`
+	HubSite                 string       `json:"hub_site" bson:"hub_site"`
+	ConditionURI            string       `json:"condition_uri" bson:"condition_uri"`
+	Encryption              *Encryption  `json:"encryption,omitempty" bson:"encryption,omitempty"`
+	Signing                 *Signing     `json:"signing,omitempty" bson:"signing,omitempty"`
+	RealtimeProcessing      string       `json:"realtimeprocessing,omitempty" bson:"realtimeprocessing,omitempty"`
+	RealtimeProcessingTopic string       `json:"realtimeprocessing_topic" bson:"realtimeprocessing_topic"`
 }
 
 // Capture defines which camera type (Id) you are using (IP, USB or Raspberry Pi camera),
@@ -207,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/models/frame_processing.go

@@ -1,33 +0,0 @@
-package models
-
-import "github.com/kerberos-io/agent/machinery/src/packets"
-
-const (
-	FrameProcessingSchemaVersion = "1.0"
-	FrameProcessingStatusAction  = "frame-processing-status"
-)
-
-type FrameProcessingRequest struct {
-	SchemaVersion     string `json:"schemaVersion"`
-	RequestID         string `json:"requestId"`
-	ProcessingProfile string `json:"processingProfile"`
-	ExpiresAt         int64  `json:"expiresAt"`
-	TraceID           string `json:"traceId,omitempty"`
-}
-
-type FrameProcessingWork struct {
-	Request FrameProcessingRequest
-	Cursor  *packets.QueueCursor
-}
-
-type FrameProcessingStatus struct {
-	SchemaVersion string `json:"schemaVersion"`
-	RequestID     string `json:"requestId"`
-	FrameID       string `json:"frameId,omitempty"`
-	DeviceID      string `json:"deviceId"`
-	Status        string `json:"status"`
-	OccurredAt    int64  `json:"occurredAt"`
-	Retryable     bool   `json:"retryable,omitempty"`
-	Message       string `json:"message,omitempty"`
-	TraceID       string `json:"traceId,omitempty"`
-}

machinery/src/models/frame_processing_test.go

@@ -1,42 +0,0 @@
-package models
-
-import (
-	"encoding/json"
-	"strings"
-	"testing"
-	"time"
-)
-
-func TestFrameProcessingTokenIsNotSerialized(t *testing.T) {
-	config := Config{FrameProcessing: &FrameProcessing{
-		Enabled:  "true",
-		Endpoint: "https://processor.example/v1/frames",
-		Token:    "do-not-expose",
-	}}
-	value, err := json.Marshal(config)
-	if err != nil {
-		t.Fatal(err)
-	}
-	if strings.Contains(string(value), config.FrameProcessing.Token) {
-		t.Fatalf("serialized config exposed frame-processing token: %s", value)
-	}
-}
-
-func TestFrameProcessingRuntimeTelemetry(t *testing.T) {
-	communication := &Communication{}
-	communication.SetFrameProcessingConfigured(true)
-	communication.RecordFrameProcessingSample()
-	communication.RecordFrameProcessingQueued(1, false)
-	communication.RecordFrameProcessingQueued(1, true)
-	communication.SetFrameProcessingQueueDepth(0)
-	communication.RecordFrameProcessingFailure()
-	communication.RecordFrameProcessingSuccess(time.Unix(123, 0))
-
-	got := communication.FrameProcessingRuntimeTelemetry()
-	if !got.Configured || got.Sampled != 1 || got.Queued != 2 || got.Dropped != 1 || got.Failed != 1 || got.Submitted != 1 {
-		t.Fatalf("FrameProcessingRuntimeTelemetry() = %+v", got)
-	}
-	if got.QueueDepth != 0 || got.LastSuccessAt != 123 {
-		t.Fatalf("FrameProcessingRuntimeTelemetry() timing = %+v", got)
-	}
-}

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()
 
@@ -140,20 +154,6 @@ func (self *Queue) Latest() *QueueCursor {
 	return cursor
 }
 
-// LatestAtCurrentTail returns a cursor fixed at the queue tail at call time.
-// Unlike Latest, its start position is not deferred until the first read. This
-// is used by command-driven consumers that must not skip packets arriving after
-// a request was accepted but before their first blocking read begins.
-func (self *Queue) LatestAtCurrentTail() *QueueCursor {
-	self.cond.L.Lock()
-	defer self.cond.L.Unlock()
-	return &QueueCursor{
-		que:    self,
-		pos:    self.buf.Tail,
-		gotpos: true,
-	}
-}
-
 // Create cursor position at oldest buffered packet.
 func (self *Queue) Oldest() *QueueCursor {
 	cursor := self.newCursor()

machinery/src/packets/queue_latest_test.go

@@ -1,20 +0,0 @@
-package packets
-
-import "testing"
-
-func TestLatestAtCurrentTailDoesNotSkipPacketWrittenAfterCreation(t *testing.T) {
-	queue := NewQueue()
-	defer queue.Close()
-	cursor := queue.LatestAtCurrentTail()
-	want := Packet{CurrentTime: 123, Data: []byte{1}}
-	if err := queue.WritePacket(want); err != nil {
-		t.Fatal(err)
-	}
-	got, err := cursor.ReadPacket()
-	if err != nil {
-		t.Fatal(err)
-	}
-	if got.CurrentTime != want.CurrentTime {
-		t.Fatalf("packet timestamp = %d, want %d", got.CurrentTime, want.CurrentTime)
-	}
-}

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

@@ -42,12 +42,11 @@ type HubHealth struct {
 
 // Health describes the Agent process health exposed to API clients.
 type Health struct {
-	Description     string                                 `json:"description"`
-	CameraConnected bool                                   `json:"cameraConnected"`
-	MainStream      StreamHealth                           `json:"mainStream"`
-	SubStream       StreamHealth                           `json:"subStream"`
-	Hub             HubHealth                              `json:"hub"`
-	FrameProcessing models.FrameProcessingRuntimeTelemetry `json:"frameProcessing"`
+	Description     string       `json:"description"`
+	CameraConnected bool         `json:"cameraConnected"`
+	MainStream      StreamHealth `json:"mainStream"`
+	SubStream       StreamHealth `json:"subStream"`
+	Hub             HubHealth    `json:"hub"`
 }
 
 // HealthResponseData contains the typed payload of a health response.
@@ -109,7 +108,6 @@ func buildHealth(communication *models.Communication, now time.Time) Health {
 			LastHeartbeatAttemptAt:    hubTelemetry.LastHeartbeatAttemptAt,
 			LastSuccessfulHeartbeatAt: hubTelemetry.LastSuccessfulHeartbeatAt,
 		},
-		FrameProcessing: communication.FrameProcessingRuntimeTelemetry(),
 	}
 }
 

machinery/src/routers/http/health_test.go

@@ -35,10 +35,6 @@ func TestHealthCheckReturnsStandardPublicResponse(t *testing.T) {
 	communication.SetHubConfigured(true)
 	communication.RecordHubHeartbeatAttempt(now.Add(-2 * time.Second))
 	communication.RecordHubHeartbeatSuccess(now.Add(-time.Second))
-	communication.SetFrameProcessingConfigured(true)
-	communication.RecordFrameProcessingSample()
-	communication.RecordFrameProcessingQueued(1, false)
-	communication.RecordFrameProcessingSuccess(now.Add(-time.Second))
 
 	router := gin.New()
 	AddRoutes(router, authMiddleware, "", nil, communication, nil)
@@ -105,9 +101,6 @@ func TestHealthCheckReturnsStandardPublicResponse(t *testing.T) {
 	if !health.Hub.Configured || !health.Hub.Connected {
 		t.Errorf("data.health.hub = %+v, want configured and connected", health.Hub)
 	}
-	if !health.FrameProcessing.Configured || health.FrameProcessing.Sampled != 1 || health.FrameProcessing.Submitted != 1 {
-		t.Errorf("data.health.frameProcessing = %+v", health.FrameProcessing)
-	}
 }
 
 func TestBuildHealthMarksStaleHubHeartbeatDisconnected(t *testing.T) {

machinery/src/routers/mqtt/main.go

@@ -1,7 +1,6 @@
 package mqtt
 
 import (
-	"bytes"
 	"context"
 	"crypto/rsa"
 	"crypto/tls"
@@ -9,7 +8,6 @@ import (
 	"encoding/base64"
 	"encoding/json"
 	"encoding/pem"
-	"errors"
 	"fmt"
 	"io/ioutil"
 	"math/rand"
@@ -580,8 +578,6 @@ func MQTTListenerHandler(mqttClient mqtt.Client, hubKey string, configDirectory
 					go HandleReceiveHDCandidates(mqttClient, hubKey, payload, configuration, communication)
 				case "trigger-relay":
 					go HandleTriggerRelay(mqttClient, hubKey, payload, configuration, communication)
-				case "capture-frame":
-					HandleCaptureFrame(mqttClient, hubKey, payload, remoteAuthenticated, configuration, communication)
 				case "remote-session-open":
 					go HandleRemoteSessionOpen(mqttClient, hubKey, payload, remoteAuthenticated, configuration)
 				case "remote-session-input":
@@ -610,105 +606,6 @@ func MQTTListenerHandler(mqttClient mqtt.Client, hubKey string, configDirectory
 	}
 }
 
-func HandleCaptureFrame(mqttClient mqtt.Client, hubKey string, payload models.Payload, remoteAuthenticated bool, configuration *models.Configuration, communication *models.Communication) {
-	request, err := decodeFrameProcessingRequest(payload)
-	if err != nil {
-		log.WithError(err).WithFields(log.Fields{
-			"component": "routers/mqtt",
-			"event":     "capture_frame_rejected",
-		}).Warn("Rejected invalid capture-frame command")
-		return
-	}
-	config := configuration.Config
-	if !frameProcessingCommandAuthenticated(config, remoteAuthenticated) {
-		log.WithFields(log.Fields{
-			"component":  "routers/mqtt",
-			"event":      "capture_frame_rejected",
-			"request_id": request.RequestID,
-		}).Warn("Rejected unauthenticated capture-frame command")
-		return
-	}
-	status := "accepted"
-	message := ""
-	frameProcessing := config.FrameProcessing
-	if frameProcessing == nil || frameProcessing.Enabled != "true" || frameProcessing.AllowRequestedFrames != "true" || config.Offline == "true" {
-		status = "rejected"
-		message = "frame processing is not available"
-	} else if request.ExpiresAt <= time.Now().UnixMilli() {
-		status = "expired"
-		message = "capture request expired"
-	} else if request.ExpiresAt-time.Now().UnixMilli() > frameProcessing.FrameTTLSeconds*1000 {
-		status = "rejected"
-		message = "capture request expiry exceeds configured frame TTL"
-	} else if !communication.TrySendFrameProcessingRequest(request) {
-		status = "rejected"
-		message = "requested-frame queue is unavailable or full"
-	}
-	publishFrameProcessingStatus(mqttClient, hubKey, configuration, models.FrameProcessingStatus{
-		SchemaVersion: models.FrameProcessingSchemaVersion,
-		RequestID:     request.RequestID,
-		DeviceID:      config.Key,
-		Status:        status,
-		OccurredAt:    time.Now().UnixMilli(),
-		Retryable:     status == "rejected" && message == "requested-frame queue is unavailable or full",
-		Message:       message,
-		TraceID:       request.TraceID,
-	})
-}
-
-func decodeFrameProcessingRequest(payload models.Payload) (models.FrameProcessingRequest, error) {
-	encoded, err := json.Marshal(payload.Value)
-	if err != nil {
-		return models.FrameProcessingRequest{}, fmt.Errorf("marshal capture-frame value: %w", err)
-	}
-	var request models.FrameProcessingRequest
-	decoder := json.NewDecoder(bytes.NewReader(encoded))
-	decoder.DisallowUnknownFields()
-	if err := decoder.Decode(&request); err != nil {
-		return models.FrameProcessingRequest{}, fmt.Errorf("decode capture-frame value: %w", err)
-	}
-	if request.SchemaVersion != models.FrameProcessingSchemaVersion {
-		return models.FrameProcessingRequest{}, fmt.Errorf("unsupported schemaVersion %q", request.SchemaVersion)
-	}
-	if request.RequestID == "" || request.ProcessingProfile == "" || request.ExpiresAt <= 0 {
-		return models.FrameProcessingRequest{}, errors.New("requestId, processingProfile, and expiresAt are required")
-	}
-	return request, nil
-}
-
-func frameProcessingCommandAuthenticated(config models.Config, remoteAuthenticated bool) bool {
-	hubAuthenticationRequired := config.HubEncryption == "true" && config.HubPrivateKey != ""
-	endToEndAuthenticationRequired := config.Encryption != nil && config.Encryption.Enabled == "true"
-	return remoteAuthenticated || (!hubAuthenticationRequired && !endToEndAuthenticationRequired)
-}
-
-func publishFrameProcessingStatus(mqttClient mqtt.Client, hubKey string, configuration *models.Configuration, status models.FrameProcessingStatus) {
-	if mqttClient == nil || hubKey == "" {
-		return
-	}
-	encoded, err := json.Marshal(status)
-	if err != nil {
-		return
-	}
-	value := make(map[string]interface{})
-	if err := json.Unmarshal(encoded, &value); err != nil {
-		return
-	}
-	payload, err := models.PackageMQTTMessage(configuration, models.Message{
-		Payload: models.Payload{
-			Version:  models.FrameProcessingSchemaVersion,
-			Action:   models.FrameProcessingStatusAction,
-			DeviceId: status.DeviceID,
-			Value:    value,
-		},
-	})
-	if err != nil {
-		log.WithError(err).Warn("Failed to package frame-processing status")
-		return
-	}
-	mqttClient.Publish("kerberos/hub/"+hubKey, 1, false, payload)
-}
-
 func HandleRecording(mqttClient mqtt.Client, hubKey string, payload models.Payload, configuration *models.Configuration, communication *models.Communication) {
 	value := payload.Value
 

machinery/src/routers/mqtt/main_test.go

@@ -157,55 +157,6 @@ func TestRemoteAccessRequiresExplicitOptIn(t *testing.T) {
 	}
 }
 
-func TestDecodeFrameProcessingRequest(t *testing.T) {
-	request, err := decodeFrameProcessingRequest(models.Payload{Value: map[string]interface{}{
-		"schemaVersion":     "1.0",
-		"requestId":         "request-1",
-		"processingProfile": "always-trigger",
-		"expiresAt":         float64(2_000),
-		"traceId":           "trace-1",
-	}})
-	if err != nil {
-		t.Fatal(err)
-	}
-	if request.RequestID != "request-1" || request.ExpiresAt != 2_000 || request.TraceID != "trace-1" {
-		t.Fatalf("decoded request = %+v", request)
-	}
-}
-
-func TestDecodeFrameProcessingRequestRejectsUnknownField(t *testing.T) {
-	_, err := decodeFrameProcessingRequest(models.Payload{Value: map[string]interface{}{
-		"schemaVersion":     "1.0",
-		"requestId":         "request-1",
-		"processingProfile": "always-trigger",
-		"expiresAt":         float64(2_000),
-		"unexpected":        true,
-	}})
-	if err == nil {
-		t.Fatal("decodeFrameProcessingRequest() accepted an unknown field")
-	}
-}
-
-func TestFrameProcessingCommandAuthentication(t *testing.T) {
-	plainConfig := models.Config{}
-	if !frameProcessingCommandAuthenticated(plainConfig, false) {
-		t.Fatal("trusted plaintext broker configuration rejected a command")
-	}
-
-	hiddenConfig := models.Config{HubEncryption: "true", HubPrivateKey: "private"}
-	if frameProcessingCommandAuthenticated(hiddenConfig, false) {
-		t.Fatal("Hub-encrypted configuration accepted plaintext command")
-	}
-	if !frameProcessingCommandAuthenticated(hiddenConfig, true) {
-		t.Fatal("Hub-encrypted configuration rejected authenticated command")
-	}
-
-	e2eConfig := models.Config{Encryption: &models.Encryption{Enabled: "true"}}
-	if frameProcessingCommandAuthenticated(e2eConfig, false) {
-		t.Fatal("end-to-end encrypted configuration accepted plaintext command")
-	}
-}
-
 func TestNormalizeTerminalSize(t *testing.T) {
 	rows, columns := normalizeTerminalSize(0, 0)
 	if rows != 24 || columns != 80 {

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/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