feat: add frame processor reference service

2026-10-19 17:38:25 +00:00 · 2026-09-15 09:26:23 +00:00
42 changed files with 2159 additions and 1389 deletions
@@ -20,7 +20,9 @@ jobs:
          go-version: "1.25.x"
          check-latest: true
          cache: true
-          cache-dependency-path: machinery/go.sum
+          cache-dependency-path: |
+            machinery/go.sum
+            examples/frame-processor/go.sum
      - name: Install native dependencies
        run: |
          sudo apt-get update
@@ -39,3 +41,5 @@ jobs:
        run: cd machinery && go vet -v ./...
      - name: Test
        run: cd machinery && go test -v ./...
+      - name: Test frame processor example
+        run: cd examples/frame-processor && GOWORK=off go test -v ./...
@@ -365,13 +365,11 @@ 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"                        |
@@ -410,94 +408,6 @@ 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

@@ -614,6 +524,11 @@ Once signed in you'll see the dashboard page. After successfull configuration of

 The `machinery` is a **Golang** project which delivers two functions: it acts as the Kerberos Agent which is doing all the heavy lifting with camera processing and other kinds of logic and on the other hand it acts as a webserver (Rest API) that allows communication from the web (React) or any other custom application. The API is documented using `swagger`.

+An executable reference for the Agent frame-processing HTTP and MQTT contracts is
+available in [`examples/frame-processor`](examples/frame-processor). It provides
+deterministic processing profiles for integration testing without requiring a
+machine-learning runtime.
+
 You can simply run the `machinery` using following commands.

    git clone https://github.com/kerberos-io/agent
@@ -0,0 +1,12 @@
+FROM golang:1.24-bookworm AS build
+
+WORKDIR /src
+COPY go.mod go.sum ./
+RUN go mod download
+COPY . .
+RUN CGO_ENABLED=0 go build -trimpath -o /frame-processor .
+
+FROM gcr.io/distroless/static-debian12:nonroot
+COPY --from=build /frame-processor /frame-processor
+EXPOSE 8080
+ENTRYPOINT ["/frame-processor"]
@@ -0,0 +1,88 @@
+# MQTT control contract
+
+The Frame Processor publishes control messages to
+`kerberos/agent/<hubKey>`. Each message targets one Agent through the envelope's
+`device_id`. It subscribes to `kerberos/hub/<hubKey>` for correlated status
+events.
+
+MQTT transports control data only. JPEG frames use `POST /v1/frames` and full
+recordings use the Agent's existing Vault upload path.
+
+## Envelope
+
+```json
+{
+  "mid": "0c556fc7-637b-4b2a-8a90-2d1cf8450956",
+  "device_id": "camera-1",
+  "timestamp": 1789380000,
+  "encrypted": false,
+  "hidden": false,
+  "public_key": "",
+  "fingerprint": "",
+  "payload": {
+    "version": "1.0",
+    "action": "capture-frame",
+    "device_id": "camera-1",
+    "signature": "",
+    "encrypted_value": "",
+    "hidden_value": "",
+    "value": {}
+  }
+}
+```
+
+The reference service publishes plaintext envelopes for local contract testing.
+Production deployment must use a trusted broker and should adopt the Agent's
+encrypted-message packaging before commands cross an untrusted broker.
+
+## `capture-frame`
+
+```json
+{
+  "schemaVersion": "1.0",
+  "requestId": "request-1",
+  "processingProfile": "always-trigger",
+  "expiresAt": 1789380030000,
+  "traceId": "optional-trace-id"
+}
+```
+
+## `request-recording-window`
+
+```json
+{
+  "schemaVersion": "1.0",
+  "requestId": "request-1",
+  "frameId": "frame-1",
+  "capturedAt": 1789380000123,
+  "preRollSeconds": 10,
+  "eventClipSeconds": 30,
+  "expiresAt": 1789380030000,
+  "processingProfile": "always-trigger",
+  "traceId": "optional-trace-id"
+}
+```
+
+`capturedAt` is generated by the Agent and must be echoed unchanged. The Agent
+uses `requestId` for command idempotency and selects the local recording that
+contains `capturedAt`.
+
+## `frame-processing-status`
+
+```json
+{
+  "schemaVersion": "1.0",
+  "requestId": "request-1",
+  "frameId": "frame-1",
+  "deviceId": "camera-1",
+  "status": "queued",
+  "occurredAt": 1789380001000,
+  "retryable": false,
+  "message": "",
+  "traceId": "optional-trace-id"
+}
+```
+
+Expected statuses are `accepted`, `captured`, `submitted`, `no-event`, `event`,
+`pending-finalisation`, `queued`, `uploaded`, `expired`, `not-found`, `rejected`,
+and `failed`.
@@ -0,0 +1,74 @@
+# Example Frame Processor
+
+This reference service defines and exercises the Kerberos Agent frame-processing
+contract. It accepts Agent JPEGs over HTTP, makes a deterministic decision, and
+publishes Agent control commands over MQTT. It has no RabbitMQ or machine-learning
+runtime dependency.
+
+## Endpoints
+
+- `GET /health`
+- `POST /v1/frames` with multipart `metadata` JSON and `frame` JPEG parts
+- `POST /v1/frame-requests` with JSON to request capture from one or more Agents
+
+See [openapi.yaml](openapi.yaml) and [MQTT.md](MQTT.md) for the versioned wire
+contracts.
+
+## Run
+
+```bash
+export FRAME_PROCESSOR_API_TOKEN=development-token
+export FRAME_PROCESSOR_MQTT_URI=tcp://localhost:1883
+export FRAME_PROCESSOR_HUB_KEY=development-hub
+export FRAME_PROCESSOR_PROFILE=never-trigger
+go run .
+```
+
+The broker credentials are optional when the local broker permits anonymous
+connections:
+
+```bash
+export FRAME_PROCESSOR_MQTT_USERNAME=...
+export FRAME_PROCESSOR_MQTT_PASSWORD=...
+```
+
+Request a frame from an Agent:
+
+```bash
+curl --fail-with-body \
+  -H 'Authorization: Bearer development-token' \
+  -H 'Content-Type: application/json' \
+  --data @testdata/frame-request.json \
+  http://localhost:8080/v1/frame-requests
+```
+
+## Profiles
+
+- `never-trigger`
+- `always-trigger`
+- `every-nth-frame`
+- `brightness-threshold`
+
+Use `FRAME_PROCESSOR_EVERY_N` and `FRAME_PROCESSOR_BRIGHTNESS_THRESHOLD` to tune
+the last two profiles. `FRAME_PROCESSOR_DELAY_MILLISECONDS` and
+`FRAME_PROCESSOR_FORCE_ERROR` provide deterministic latency and failure
+simulation.
+
+Recording commands default to a 30-second event clip with 10 seconds of pre-roll.
+Configure them with `FRAME_PROCESSOR_EVENT_CLIP_SECONDS` and
+`FRAME_PROCESSOR_PRE_ROLL_SECONDS`.
+
+The reference MQTT publisher emits plaintext Agent envelopes for local contract
+testing. Use a trusted broker. Production support for untrusted brokers requires
+the same encrypted-message packaging used by Hub and Agent.
+
+Frame idempotency is guaranteed until the submitted frame's `expiresAt`. The
+service rejects frame TTLs longer than `FRAME_PROCESSOR_MAX_FRAME_TTL_SECONDS`
+(five minutes by default), then evicts the cached result at expiry.
+
+## Verify
+
+```bash
+GOWORK=off go test ./...
+GOWORK=off go vet ./...
+```
@@ -0,0 +1,162 @@
+package contract
+
+import (
+	"errors"
+	"fmt"
+)
+
+const (
+	SchemaVersion     = "1.0"
+	MaxImageDimension = 8192
+
+	ActionCaptureFrame           = "capture-frame"
+	ActionRequestRecordingWindow = "request-recording-window"
+	ActionFrameStatus            = "frame-processing-status"
+)
+
+type FrameMetadata 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"`
+}
+
+func (m FrameMetadata) Validate(nowMillis int64) error {
+	if m.SchemaVersion != SchemaVersion {
+		return fmt.Errorf("unsupported schemaVersion %q", m.SchemaVersion)
+	}
+	if m.RequestID == "" || m.FrameID == "" || m.DeviceID == "" {
+		return errors.New("requestId, frameId, and deviceId are required")
+	}
+	if m.CapturedAt <= 0 {
+		return errors.New("capturedAt must be a positive Unix millisecond timestamp")
+	}
+	if m.ExpiresAt <= m.CapturedAt {
+		return errors.New("expiresAt must be later than capturedAt")
+	}
+	if nowMillis > 0 && m.ExpiresAt <= nowMillis {
+		return errors.New("frame has expired")
+	}
+	if m.ProcessingProfile == "" {
+		return errors.New("processingProfile is required")
+	}
+	if m.SourceStream != "main" && m.SourceStream != "sub" {
+		return errors.New("sourceStream must be main or sub")
+	}
+	if m.Width <= 0 || m.Height <= 0 || m.Width > MaxImageDimension || m.Height > MaxImageDimension {
+		return fmt.Errorf("width and height must be between 1 and %d", MaxImageDimension)
+	}
+	return nil
+}
+
+type FrameRequest struct {
+	SchemaVersion     string   `json:"schemaVersion"`
+	RequestID         string   `json:"requestId,omitempty"`
+	DeviceIDs         []string `json:"deviceIds"`
+	ProcessingProfile string   `json:"processingProfile"`
+	ExpiresAt         int64    `json:"expiresAt"`
+	TraceID           string   `json:"traceId,omitempty"`
+}
+
+func (r FrameRequest) Validate(nowMillis int64) error {
+	if r.SchemaVersion != SchemaVersion {
+		return fmt.Errorf("unsupported schemaVersion %q", r.SchemaVersion)
+	}
+	if len(r.DeviceIDs) == 0 {
+		return errors.New("at least one deviceId is required")
+	}
+	for _, deviceID := range r.DeviceIDs {
+		if deviceID == "" {
+			return errors.New("deviceIds cannot contain empty values")
+		}
+	}
+	if r.ProcessingProfile == "" {
+		return errors.New("processingProfile is required")
+	}
+	if r.ExpiresAt <= nowMillis {
+		return errors.New("expiresAt must be in the future")
+	}
+	return nil
+}
+
+type CaptureFrameCommand struct {
+	SchemaVersion     string `json:"schemaVersion"`
+	RequestID         string `json:"requestId"`
+	ProcessingProfile string `json:"processingProfile"`
+	ExpiresAt         int64  `json:"expiresAt"`
+	TraceID           string `json:"traceId,omitempty"`
+}
+
+type RecordingWindowCommand struct {
+	SchemaVersion     string `json:"schemaVersion"`
+	RequestID         string `json:"requestId"`
+	FrameID           string `json:"frameId"`
+	CapturedAt        int64  `json:"capturedAt"`
+	PreRollSeconds    int64  `json:"preRollSeconds"`
+	EventClipSeconds  int64  `json:"eventClipSeconds"`
+	ExpiresAt         int64  `json:"expiresAt"`
+	ProcessingProfile string `json:"processingProfile"`
+	TraceID           string `json:"traceId,omitempty"`
+}
+
+func (c RecordingWindowCommand) Validate(nowMillis int64) error {
+	if c.SchemaVersion != SchemaVersion {
+		return fmt.Errorf("unsupported schemaVersion %q", c.SchemaVersion)
+	}
+	if c.RequestID == "" || c.FrameID == "" {
+		return errors.New("requestId and frameId are required")
+	}
+	if c.CapturedAt <= 0 {
+		return errors.New("capturedAt must be a positive Unix millisecond timestamp")
+	}
+	if c.EventClipSeconds <= 0 {
+		return errors.New("eventClipSeconds must be positive")
+	}
+	if c.PreRollSeconds < 0 || c.PreRollSeconds > c.EventClipSeconds {
+		return errors.New("preRollSeconds must be between zero and eventClipSeconds")
+	}
+	if c.ExpiresAt <= nowMillis {
+		return errors.New("expiresAt must be in the future")
+	}
+	return nil
+}
+
+type StatusEvent 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"`
+}
+
+type MQTTMessage struct {
+	MID         string      `json:"mid"`
+	DeviceID    string      `json:"device_id"`
+	Timestamp   int64       `json:"timestamp"`
+	Encrypted   bool        `json:"encrypted"`
+	Hidden      bool        `json:"hidden"`
+	PublicKey   string      `json:"public_key"`
+	Fingerprint string      `json:"fingerprint"`
+	Payload     MQTTPayload `json:"payload"`
+}
+
+type MQTTPayload struct {
+	Version        string `json:"version"`
+	Action         string `json:"action"`
+	DeviceID       string `json:"device_id"`
+	Signature      string `json:"signature"`
+	EncryptedValue string `json:"encrypted_value"`
+	HiddenValue    string `json:"hidden_value"`
+	Value          any    `json:"value"`
+}
@@ -0,0 +1,73 @@
+package contract
+
+import (
+	"encoding/json"
+	"os"
+	"testing"
+)
+
+func TestFrameMetadataValidate(t *testing.T) {
+	now := int64(1_000)
+	metadata := FrameMetadata{
+		SchemaVersion:     SchemaVersion,
+		RequestID:         "request-1",
+		FrameID:           "frame-1",
+		DeviceID:          "device-1",
+		CapturedAt:        900,
+		ExpiresAt:         1_100,
+		ProcessingProfile: "always-trigger",
+		SourceStream:      "sub",
+		Width:             640,
+		Height:            480,
+	}
+
+	if err := metadata.Validate(now); err != nil {
+		t.Fatalf("Validate() error = %v", err)
+	}
+
+	metadata.ExpiresAt = now
+	if err := metadata.Validate(now); err == nil {
+		t.Fatal("Validate() accepted an expired frame")
+	}
+}
+
+func TestRecordingWindowCommandValidate(t *testing.T) {
+	command := RecordingWindowCommand{
+		SchemaVersion:    SchemaVersion,
+		RequestID:        "request-1",
+		FrameID:          "frame-1",
+		CapturedAt:       900,
+		PreRollSeconds:   10,
+		EventClipSeconds: 30,
+		ExpiresAt:        2_000,
+	}
+
+	if err := command.Validate(1_000); err != nil {
+		t.Fatalf("Validate() error = %v", err)
+	}
+
+	command.PreRollSeconds = 31
+	if err := command.Validate(1_000); err == nil {
+		t.Fatal("Validate() accepted pre-roll longer than the event clip")
+	}
+}
+
+func TestContractFixturesDecode(t *testing.T) {
+	tests := []struct {
+		path   string
+		target any
+	}{
+		{"../testdata/frame-request.json", &FrameRequest{}},
+		{"../testdata/capture-frame.json", &CaptureFrameCommand{}},
+		{"../testdata/request-recording-window.json", &RecordingWindowCommand{}},
+	}
+	for _, test := range tests {
+		value, err := os.ReadFile(test.path)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if err := json.Unmarshal(value, test.target); err != nil {
+			t.Fatalf("decode %s: %v", test.path, err)
+		}
+	}
+}
@@ -0,0 +1,11 @@
+module github.com/kerberos-io/agent/examples/frame-processor
+
+go 1.24.2
+
+require github.com/eclipse/paho.mqtt.golang v1.5.0
+
+require (
+	github.com/gorilla/websocket v1.5.3 // indirect
+	golang.org/x/net v0.27.0 // indirect
+	golang.org/x/sync v0.7.0 // indirect
+)
@@ -0,0 +1,8 @@
+github.com/eclipse/paho.mqtt.golang v1.5.0 h1:EH+bUVJNgttidWFkLLVKaQPGmkTUfQQqjOsyvMGvD6o=
+github.com/eclipse/paho.mqtt.golang v1.5.0/go.mod h1:du/2qNQVqJf/Sqs4MEL77kR8QTqANF7XU7Fk0aOTAgk=
+github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg=
+github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
+golang.org/x/net v0.27.0 h1:5K3Njcw06/l2y9vpGCSdcxWOYHOUk3dVNGDXN+FvAys=
+golang.org/x/net v0.27.0/go.mod h1:dDi0PyhWNoiUOrAS8uXv/vnScO4wnHQO4mj9fn/RytE=
+golang.org/x/sync v0.7.0 h1:YsImfSBoP9QPYL0xyKJPq0gcaJdG3rInoqxTWbfQu9M=
+golang.org/x/sync v0.7.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
@@ -0,0 +1,184 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"log/slog"
+	"net/http"
+	"os"
+	"os/signal"
+	"strconv"
+	"syscall"
+	"time"
+
+	"github.com/kerberos-io/agent/examples/frame-processor/contract"
+	"github.com/kerberos-io/agent/examples/frame-processor/mqttpublisher"
+	"github.com/kerberos-io/agent/examples/frame-processor/processor"
+	"github.com/kerberos-io/agent/examples/frame-processor/service"
+)
+
+func main() {
+	config, err := loadConfig()
+	if err != nil {
+		slog.Error("invalid configuration", "error", err)
+		os.Exit(1)
+	}
+
+	publisher, err := mqttpublisher.New(config.mqtt, func(status contract.StatusEvent) {
+		slog.Info("Agent frame-processing status",
+			"deviceId", status.DeviceID,
+			"requestId", status.RequestID,
+			"frameId", status.FrameID,
+			"status", status.Status,
+		)
+	})
+	if err != nil {
+		slog.Error("failed to initialize MQTT", "error", err)
+		os.Exit(1)
+	}
+	defer publisher.Close()
+
+	engine := processor.New(config.processor)
+	application := service.New(config.service, engine, publisher, nil)
+	server := &http.Server{
+		Addr:              config.address,
+		Handler:           application.Handler(),
+		ReadHeaderTimeout: 5 * time.Second,
+		IdleTimeout:       60 * time.Second,
+	}
+
+	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
+	defer stop()
+	go func() {
+		<-ctx.Done()
+		shutdownContext, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+		defer cancel()
+		_ = server.Shutdown(shutdownContext)
+	}()
+
+	slog.Info("Frame Processor listening", "address", config.address, "profile", config.processor.DefaultProfile)
+	if err := server.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
+		slog.Error("Frame Processor stopped", "error", err)
+		os.Exit(1)
+	}
+}
+
+type applicationConfig struct {
+	address   string
+	mqtt      mqttpublisher.Config
+	processor processor.Config
+	service   service.Config
+}
+
+func loadConfig() (applicationConfig, error) {
+	brightnessThreshold := envInt("FRAME_PROCESSOR_BRIGHTNESS_THRESHOLD", 200)
+	config := applicationConfig{
+		address: envString("FRAME_PROCESSOR_ADDRESS", ":8080"),
+		mqtt: mqttpublisher.Config{
+			BrokerURI: os.Getenv("FRAME_PROCESSOR_MQTT_URI"),
+			Username:  os.Getenv("FRAME_PROCESSOR_MQTT_USERNAME"),
+			Password:  os.Getenv("FRAME_PROCESSOR_MQTT_PASSWORD"),
+			HubKey:    os.Getenv("FRAME_PROCESSOR_HUB_KEY"),
+			ClientID:  os.Getenv("FRAME_PROCESSOR_MQTT_CLIENT_ID"),
+			Timeout:   envDurationSeconds("FRAME_PROCESSOR_MQTT_TIMEOUT_SECONDS", 10),
+		},
+		processor: processor.Config{
+			DefaultProfile:      envString("FRAME_PROCESSOR_PROFILE", processor.ProfileNeverTrigger),
+			EveryN:              envInt("FRAME_PROCESSOR_EVERY_N", 2),
+			BrightnessThreshold: uint8(brightnessThreshold),
+			Delay:               envDurationMillis("FRAME_PROCESSOR_DELAY_MILLISECONDS", 0),
+			ForceError:          envBool("FRAME_PROCESSOR_FORCE_ERROR", false),
+		},
+		service: service.Config{
+			APIToken:         os.Getenv("FRAME_PROCESSOR_API_TOKEN"),
+			MaxFrameBytes:    int64(envInt("FRAME_PROCESSOR_MAX_FRAME_BYTES", 4<<20)),
+			MaxFrameTTL:      envDurationSeconds("FRAME_PROCESSOR_MAX_FRAME_TTL_SECONDS", 300),
+			CommandTTL:       envDurationSeconds("FRAME_PROCESSOR_COMMAND_TTL_SECONDS", 30),
+			PreRollSeconds:   int64(envInt("FRAME_PROCESSOR_PRE_ROLL_SECONDS", 10)),
+			EventClipSeconds: int64(envInt("FRAME_PROCESSOR_EVENT_CLIP_SECONDS", 30)),
+		},
+	}
+	if config.service.APIToken == "" {
+		return applicationConfig{}, errors.New("FRAME_PROCESSOR_API_TOKEN is required")
+	}
+	if config.mqtt.BrokerURI == "" || config.mqtt.HubKey == "" {
+		return applicationConfig{}, errors.New("FRAME_PROCESSOR_MQTT_URI and FRAME_PROCESSOR_HUB_KEY are required")
+	}
+	if config.processor.EveryN <= 0 {
+		return applicationConfig{}, errors.New("FRAME_PROCESSOR_EVERY_N must be positive")
+	}
+	if !processor.IsProfileSupported(config.processor.DefaultProfile) {
+		return applicationConfig{}, fmt.Errorf("unsupported FRAME_PROCESSOR_PROFILE %q", config.processor.DefaultProfile)
+	}
+	if brightnessThreshold < 0 || brightnessThreshold > 255 {
+		return applicationConfig{}, errors.New("FRAME_PROCESSOR_BRIGHTNESS_THRESHOLD must be between 0 and 255")
+	}
+	if config.processor.Delay < 0 || config.processor.Delay > time.Minute {
+		return applicationConfig{}, errors.New("FRAME_PROCESSOR_DELAY_MILLISECONDS must be between 0 and 60000")
+	}
+	if config.service.MaxFrameBytes <= 0 || config.service.MaxFrameBytes > 100<<20 {
+		return applicationConfig{}, errors.New("FRAME_PROCESSOR_MAX_FRAME_BYTES must be between 1 and 104857600")
+	}
+	if config.service.MaxFrameTTL <= 0 || config.service.MaxFrameTTL > time.Hour {
+		return applicationConfig{}, errors.New("FRAME_PROCESSOR_MAX_FRAME_TTL_SECONDS must be between 1 and 3600")
+	}
+	if config.service.PreRollSeconds < 0 || config.service.PreRollSeconds > config.service.EventClipSeconds {
+		return applicationConfig{}, errors.New("FRAME_PROCESSOR_PRE_ROLL_SECONDS must be between zero and FRAME_PROCESSOR_EVENT_CLIP_SECONDS")
+	}
+	if config.service.CommandTTL <= 0 || config.service.CommandTTL > time.Hour {
+		return applicationConfig{}, errors.New("FRAME_PROCESSOR_COMMAND_TTL_SECONDS must be between 1 and 3600")
+	}
+	if config.service.EventClipSeconds <= 0 || config.service.EventClipSeconds > 24*60*60 {
+		return applicationConfig{}, errors.New("FRAME_PROCESSOR_EVENT_CLIP_SECONDS must be between 1 and 86400")
+	}
+	if config.mqtt.Timeout <= 0 || config.mqtt.Timeout > time.Minute {
+		return applicationConfig{}, errors.New("FRAME_PROCESSOR_MQTT_TIMEOUT_SECONDS must be between 1 and 60")
+	}
+	return config, nil
+}
+
+func envString(name, fallback string) string {
+	if value := os.Getenv(name); value != "" {
+		return value
+	}
+	return fallback
+}
+
+func envInt(name string, fallback int) int {
+	value := os.Getenv(name)
+	if value == "" {
+		return fallback
+	}
+	parsed, err := strconv.Atoi(value)
+	if err != nil {
+		slog.Warn("invalid integer environment value, using default", "name", name)
+		return fallback
+	}
+	return parsed
+}
+
+func envBool(name string, fallback bool) bool {
+	value := os.Getenv(name)
+	if value == "" {
+		return fallback
+	}
+	parsed, err := strconv.ParseBool(value)
+	if err != nil {
+		slog.Warn("invalid boolean environment value, using default", "name", name)
+		return fallback
+	}
+	return parsed
+}
+
+func envDurationSeconds(name string, fallback int) time.Duration {
+	return time.Duration(envInt(name, fallback)) * time.Second
+}
+
+func envDurationMillis(name string, fallback int) time.Duration {
+	return time.Duration(envInt(name, fallback)) * time.Millisecond
+}
+
+func (c applicationConfig) String() string {
+	return fmt.Sprintf("address=%s profile=%s", c.address, c.processor.DefaultProfile)
+}
@@ -0,0 +1,25 @@
+package main
+
+import "testing"
+
+func TestLoadConfigRejectsInvalidBounds(t *testing.T) {
+	t.Setenv("FRAME_PROCESSOR_API_TOKEN", "secret")
+	t.Setenv("FRAME_PROCESSOR_MQTT_URI", "tcp://localhost:1883")
+	t.Setenv("FRAME_PROCESSOR_HUB_KEY", "hub")
+	t.Setenv("FRAME_PROCESSOR_BRIGHTNESS_THRESHOLD", "256")
+
+	if _, err := loadConfig(); err == nil {
+		t.Fatal("loadConfig() accepted an invalid brightness threshold")
+	}
+}
+
+func TestLoadConfigRejectsUnknownProfile(t *testing.T) {
+	t.Setenv("FRAME_PROCESSOR_API_TOKEN", "secret")
+	t.Setenv("FRAME_PROCESSOR_MQTT_URI", "tcp://localhost:1883")
+	t.Setenv("FRAME_PROCESSOR_HUB_KEY", "hub")
+	t.Setenv("FRAME_PROCESSOR_PROFILE", "unknown")
+
+	if _, err := loadConfig(); err == nil {
+		t.Fatal("loadConfig() accepted an unknown profile")
+	}
+}
@@ -0,0 +1,171 @@
+package mqttpublisher
+
+import (
+	"context"
+	"crypto/rand"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"log/slog"
+	"strings"
+	"time"
+
+	mqtt "github.com/eclipse/paho.mqtt.golang"
+	"github.com/kerberos-io/agent/examples/frame-processor/contract"
+)
+
+const (
+	commandQoS = byte(1)
+	statusQoS  = byte(1)
+)
+
+type Config struct {
+	BrokerURI string
+	Username  string
+	Password  string
+	HubKey    string
+	ClientID  string
+	Timeout   time.Duration
+}
+
+type StatusHandler func(contract.StatusEvent)
+
+type Publisher struct {
+	client       mqtt.Client
+	commandTopic string
+	timeout      time.Duration
+}
+
+func New(config Config, handler StatusHandler) (*Publisher, error) {
+	if config.BrokerURI == "" || config.HubKey == "" {
+		return nil, errors.New("MQTT broker URI and Hub key are required")
+	}
+	if config.Timeout <= 0 {
+		config.Timeout = 10 * time.Second
+	}
+	if config.ClientID == "" {
+		config.ClientID = "frame-processor-" + randomID()
+	}
+
+	statusTopic := "kerberos/hub/" + config.HubKey
+	options := mqtt.NewClientOptions().
+		AddBroker(config.BrokerURI).
+		SetClientID(config.ClientID).
+		SetUsername(config.Username).
+		SetPassword(config.Password).
+		SetCleanSession(false).
+		SetResumeSubs(true).
+		SetAutoReconnect(true).
+		SetConnectRetry(true).
+		SetConnectRetryInterval(5 * time.Second).
+		SetMaxReconnectInterval(time.Minute).
+		SetKeepAlive(30 * time.Second).
+		SetPingTimeout(10 * time.Second)
+	if handler != nil {
+		options.SetOnConnectHandler(func(client mqtt.Client) {
+			token := client.Subscribe(statusTopic, statusQoS, statusMessageHandler(handler))
+			if !token.WaitTimeout(config.Timeout) || token.Error() != nil {
+				slog.Error("failed to subscribe to Agent status events", "topic", statusTopic, "error", token.Error())
+			}
+		})
+	}
+
+	client := mqtt.NewClient(options)
+	token := client.Connect()
+	if !token.WaitTimeout(config.Timeout) {
+		return nil, errors.New("MQTT connection timed out")
+	}
+	if err := token.Error(); err != nil {
+		return nil, fmt.Errorf("connect MQTT: %w", err)
+	}
+	return &Publisher{
+		client: client, commandTopic: "kerberos/agent/" + config.HubKey,
+		timeout: config.Timeout,
+	}, nil
+}
+
+func (p *Publisher) Close() {
+	if p != nil && p.client != nil && p.client.IsConnected() {
+		p.client.Disconnect(250)
+	}
+}
+
+func (p *Publisher) PublishCaptureFrame(ctx context.Context, deviceID string, command contract.CaptureFrameCommand) error {
+	return p.publish(ctx, deviceID, contract.ActionCaptureFrame, command)
+}
+
+func (p *Publisher) PublishRecordingWindow(ctx context.Context, deviceID string, command contract.RecordingWindowCommand) error {
+	return p.publish(ctx, deviceID, contract.ActionRequestRecordingWindow, command)
+}
+
+func (p *Publisher) publish(ctx context.Context, deviceID, action string, value any) error {
+	message := newMessage(deviceID, action, value, time.Now())
+	payload, err := json.Marshal(message)
+	if err != nil {
+		return fmt.Errorf("marshal MQTT command: %w", err)
+	}
+	token := p.client.Publish(p.commandTopic, commandQoS, false, payload)
+	timer := time.NewTimer(p.timeout)
+	defer timer.Stop()
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	case <-timer.C:
+		return errors.New("MQTT publish timed out")
+	case <-token.Done():
+		if err := token.Error(); err != nil {
+			return fmt.Errorf("publish MQTT command: %w", err)
+		}
+		return nil
+	}
+}
+
+func newMessage(deviceID, action string, value any, now time.Time) contract.MQTTMessage {
+	return contract.MQTTMessage{
+		MID:       randomID(),
+		DeviceID:  deviceID,
+		Timestamp: now.Unix(),
+		Payload: contract.MQTTPayload{
+			Version:  contract.SchemaVersion,
+			Action:   action,
+			DeviceID: deviceID,
+			Value:    value,
+		},
+	}
+}
+
+func randomID() string {
+	value := make([]byte, 16)
+	if _, err := rand.Read(value); err != nil {
+		return fmt.Sprintf("%d", time.Now().UnixNano())
+	}
+	value[6] = (value[6] & 0x0f) | 0x40
+	value[8] = (value[8] & 0x3f) | 0x80
+	encoded := hex.EncodeToString(value)
+	return strings.Join([]string{encoded[0:8], encoded[8:12], encoded[12:16], encoded[16:20], encoded[20:32]}, "-")
+}
+
+func statusMessageHandler(handler StatusHandler) mqtt.MessageHandler {
+	return func(_ mqtt.Client, message mqtt.Message) {
+		var envelope contract.MQTTMessage
+		if err := json.Unmarshal(message.Payload(), &envelope); err != nil {
+			slog.Warn("discarding malformed Agent status envelope", "error", err)
+			return
+		}
+		if envelope.Payload.Action != contract.ActionFrameStatus {
+			return
+		}
+		value, err := json.Marshal(envelope.Payload.Value)
+		if err != nil {
+			slog.Warn("discarding unencodable Agent status value", "error", err)
+			return
+		}
+		var status contract.StatusEvent
+		if err := json.Unmarshal(value, &status); err != nil {
+			slog.Warn("discarding malformed Agent status value", "error", err)
+			return
+		}
+		handler(status)
+	}
+}
@@ -0,0 +1,54 @@
+package mqttpublisher
+
+import (
+	"encoding/json"
+	"testing"
+	"time"
+
+	"github.com/kerberos-io/agent/examples/frame-processor/contract"
+)
+
+func TestNewMessageMatchesAgentEnvelope(t *testing.T) {
+	command := contract.CaptureFrameCommand{
+		SchemaVersion:     contract.SchemaVersion,
+		RequestID:         "request-1",
+		ProcessingProfile: "always-trigger",
+		ExpiresAt:         2_000,
+	}
+	message := newMessage("device-1", contract.ActionCaptureFrame, command, time.Unix(1_000, 0))
+	payload, err := json.Marshal(message)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	var decoded map[string]any
+	if err := json.Unmarshal(payload, &decoded); err != nil {
+		t.Fatal(err)
+	}
+	if decoded["device_id"] != "device-1" || decoded["timestamp"] != float64(1_000) {
+		t.Fatalf("envelope = %s", payload)
+	}
+	inner := decoded["payload"].(map[string]any)
+	if inner["action"] != contract.ActionCaptureFrame || inner["device_id"] != "device-1" {
+		t.Fatalf("payload = %#v", inner)
+	}
+}
+
+func TestStatusMessageHandlerIgnoresOtherActions(t *testing.T) {
+	called := false
+	handler := statusMessageHandler(func(contract.StatusEvent) { called = true })
+	handler(nil, fakeMessage(`{"payload":{"action":"motion","value":{}}}`))
+	if called {
+		t.Fatal("handler accepted an unrelated action")
+	}
+}
+
+type fakeMessage string
+
+func (m fakeMessage) Duplicate() bool   { return false }
+func (m fakeMessage) Qos() byte         { return 1 }
+func (m fakeMessage) Retained() bool    { return false }
+func (m fakeMessage) Topic() string     { return "test" }
+func (m fakeMessage) MessageID() uint16 { return 1 }
+func (m fakeMessage) Payload() []byte   { return []byte(m) }
+func (m fakeMessage) Ack()              {}
@@ -0,0 +1,216 @@
+openapi: 3.0.3
+info:
+  title: Kerberos Frame Processing API
+  version: 1.0.0
+  description: Reference contract for Agent frame submission and externally requested capture.
+servers:
+  - url: http://localhost:8080
+security:
+  - bearerAuth: []
+paths:
+  /health:
+    get:
+      security: []
+      summary: Check service liveness
+      responses:
+        "200":
+          description: Service is healthy
+          content:
+            application/json:
+              schema:
+                type: object
+                required: [status]
+                properties:
+                  status:
+                    type: string
+                    enum: [healthy]
+  /v1/frames:
+    post:
+      summary: Process one Agent frame
+      requestBody:
+        required: true
+        content:
+          multipart/form-data:
+            schema:
+              type: object
+              required: [metadata, frame]
+              properties:
+                metadata:
+                  $ref: "#/components/schemas/FrameMetadata"
+                frame:
+                  type: string
+                  format: binary
+            encoding:
+              metadata:
+                contentType: application/json
+              frame:
+                contentType: image/jpeg
+      responses:
+        "200":
+          description: Frame processed
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/FrameResponse"
+        "400":
+          $ref: "#/components/responses/BadRequest"
+        "401":
+          $ref: "#/components/responses/Unauthorized"
+        "413":
+          description: Frame exceeds the configured request limit
+        "415":
+          description: Frame is not a valid JPEG
+        "422":
+          description: Metadata is invalid or expired
+        "502":
+          description: Processing or MQTT command publication failed
+  /v1/frame-requests:
+    post:
+      summary: Request a new frame from one or more Agents
+      requestBody:
+        required: true
+        content:
+          application/json:
+            schema:
+              $ref: "#/components/schemas/FrameRequest"
+      responses:
+        "202":
+          description: Capture commands accepted for publication
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/FrameRequestResponse"
+        "400":
+          $ref: "#/components/responses/BadRequest"
+        "401":
+          $ref: "#/components/responses/Unauthorized"
+        "422":
+          description: Request is invalid or expired
+        "502":
+          description: MQTT command publication failed
+components:
+  securitySchemes:
+    bearerAuth:
+      type: http
+      scheme: bearer
+  responses:
+    BadRequest:
+      description: Malformed request
+      content:
+        application/json:
+          schema:
+            $ref: "#/components/schemas/Error"
+    Unauthorized:
+      description: Missing or invalid bearer token
+      content:
+        application/json:
+          schema:
+            $ref: "#/components/schemas/Error"
+  schemas:
+    FrameMetadata:
+      type: object
+      required:
+        - schemaVersion
+        - requestId
+        - frameId
+        - deviceId
+        - capturedAt
+        - expiresAt
+        - processingProfile
+        - sourceStream
+        - width
+        - height
+      properties:
+        schemaVersion:
+          type: string
+          enum: ["1.0"]
+        requestId:
+          type: string
+        frameId:
+          type: string
+        deviceId:
+          type: string
+        capturedAt:
+          type: integer
+          format: int64
+          description: Agent wall-clock capture time in Unix milliseconds.
+        expiresAt:
+          type: integer
+          format: int64
+          description: Unix milliseconds after which the frame must not be processed.
+        processingProfile:
+          type: string
+        sourceStream:
+          type: string
+          enum: [main, sub]
+        width:
+          type: integer
+          minimum: 1
+        height:
+          type: integer
+          minimum: 1
+        traceId:
+          type: string
+    FrameRequest:
+      type: object
+      required: [schemaVersion, deviceIds, processingProfile, expiresAt]
+      properties:
+        schemaVersion:
+          type: string
+          enum: ["1.0"]
+        requestId:
+          type: string
+          description: Generated by the service when omitted.
+        deviceIds:
+          type: array
+          minItems: 1
+          items:
+            type: string
+        processingProfile:
+          type: string
+        expiresAt:
+          type: integer
+          format: int64
+        traceId:
+          type: string
+    FrameResponse:
+      type: object
+      required: [schemaVersion, requestId, frameId, decision, reason]
+      properties:
+        schemaVersion:
+          type: string
+          enum: ["1.0"]
+        requestId:
+          type: string
+        frameId:
+          type: string
+        decision:
+          type: string
+          enum: [no-event, event]
+        reason:
+          type: string
+    FrameRequestResponse:
+      type: object
+      required: [schemaVersion, requestId, deviceIds, status]
+      properties:
+        schemaVersion:
+          type: string
+          enum: ["1.0"]
+        requestId:
+          type: string
+        deviceIds:
+          type: array
+          items:
+            type: string
+        status:
+          type: string
+          enum: [accepted]
+    Error:
+      type: object
+      required: [schemaVersion, error]
+      properties:
+        schemaVersion:
+          type: string
+          enum: ["1.0"]
+        error:
+          type: string
@@ -0,0 +1,117 @@
+package processor
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"fmt"
+	"image/jpeg"
+	"sync"
+	"time"
+
+	"github.com/kerberos-io/agent/examples/frame-processor/contract"
+)
+
+const (
+	ProfileNeverTrigger        = "never-trigger"
+	ProfileAlwaysTrigger       = "always-trigger"
+	ProfileEveryNthFrame       = "every-nth-frame"
+	ProfileBrightnessThreshold = "brightness-threshold"
+)
+
+type Decision struct {
+	Triggered bool   `json:"triggered"`
+	Reason    string `json:"reason"`
+}
+
+type Config struct {
+	DefaultProfile      string
+	EveryN              int
+	BrightnessThreshold uint8
+	Delay               time.Duration
+	ForceError          bool
+}
+
+type Engine struct {
+	config Config
+	mu     sync.Mutex
+	counts map[string]int
+}
+
+func IsProfileSupported(profile string) bool {
+	switch profile {
+	case ProfileNeverTrigger, ProfileAlwaysTrigger, ProfileEveryNthFrame, ProfileBrightnessThreshold:
+		return true
+	default:
+		return false
+	}
+}
+
+func New(config Config) *Engine {
+	if config.DefaultProfile == "" {
+		config.DefaultProfile = ProfileNeverTrigger
+	}
+	if config.EveryN <= 0 {
+		config.EveryN = 2
+	}
+	return &Engine{config: config, counts: make(map[string]int)}
+}
+
+func (e *Engine) Process(ctx context.Context, metadata contract.FrameMetadata, frame []byte) (Decision, error) {
+	if e.config.Delay > 0 {
+		timer := time.NewTimer(e.config.Delay)
+		defer timer.Stop()
+		select {
+		case <-ctx.Done():
+			return Decision{}, ctx.Err()
+		case <-timer.C:
+		}
+	}
+	if e.config.ForceError {
+		return Decision{}, errors.New("configured processing failure")
+	}
+
+	profile := metadata.ProcessingProfile
+	if profile == "" {
+		profile = e.config.DefaultProfile
+	}
+	switch profile {
+	case ProfileNeverTrigger:
+		return Decision{Reason: ProfileNeverTrigger}, nil
+	case ProfileAlwaysTrigger:
+		return Decision{Triggered: true, Reason: ProfileAlwaysTrigger}, nil
+	case ProfileEveryNthFrame:
+		e.mu.Lock()
+		e.counts[metadata.DeviceID]++
+		count := e.counts[metadata.DeviceID]
+		e.mu.Unlock()
+		return Decision{
+			Triggered: count%e.config.EveryN == 0,
+			Reason:    fmt.Sprintf("frame %d of every %d", count, e.config.EveryN),
+		}, nil
+	case ProfileBrightnessThreshold:
+		image, err := jpeg.Decode(bytes.NewReader(frame))
+		if err != nil {
+			return Decision{}, fmt.Errorf("decode JPEG: %w", err)
+		}
+		bounds := image.Bounds()
+		var total uint64
+		for y := bounds.Min.Y; y < bounds.Max.Y; y++ {
+			for x := bounds.Min.X; x < bounds.Max.X; x++ {
+				gray, _, _, _ := image.At(x, y).RGBA()
+				total += uint64(gray >> 8)
+			}
+		}
+		pixels := uint64(bounds.Dx() * bounds.Dy())
+		if pixels == 0 {
+			return Decision{}, errors.New("JPEG has no pixels")
+		}
+		average := uint8(total / pixels)
+		return Decision{
+			Triggered: average >= e.config.BrightnessThreshold,
+			Reason:    fmt.Sprintf("average brightness %d, threshold %d", average, e.config.BrightnessThreshold),
+		}, nil
+	default:
+		return Decision{}, fmt.Errorf("unknown processing profile %q", profile)
+	}
+}
@@ -0,0 +1,41 @@
+package processor
+
+import (
+	"context"
+	"testing"
+
+	"github.com/kerberos-io/agent/examples/frame-processor/contract"
+)
+
+func TestEveryNthFrameIsTrackedPerDevice(t *testing.T) {
+	engine := New(Config{EveryN: 2})
+	metadata := contract.FrameMetadata{ProcessingProfile: ProfileEveryNthFrame, DeviceID: "device-1"}
+
+	first, err := engine.Process(context.Background(), metadata, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	second, err := engine.Process(context.Background(), metadata, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	metadata.DeviceID = "device-2"
+	otherDevice, err := engine.Process(context.Background(), metadata, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if first.Triggered || !second.Triggered || otherDevice.Triggered {
+		t.Fatalf("decisions = first:%t second:%t other:%t", first.Triggered, second.Triggered, otherDevice.Triggered)
+	}
+}
+
+func TestUnknownProfileFails(t *testing.T) {
+	engine := New(Config{})
+	_, err := engine.Process(context.Background(), contract.FrameMetadata{
+		ProcessingProfile: "missing-profile",
+	}, nil)
+	if err == nil {
+		t.Fatal("Process() accepted an unknown profile")
+	}
+}
@@ -0,0 +1,358 @@
+package service
+
+import (
+	"bytes"
+	"context"
+	"crypto/subtle"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"image"
+	_ "image/jpeg"
+	"io"
+	"log/slog"
+	"mime/multipart"
+	"net/http"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/kerberos-io/agent/examples/frame-processor/contract"
+	"github.com/kerberos-io/agent/examples/frame-processor/processor"
+)
+
+const maxMetadataBytes = 64 << 10
+
+var errRequestTooLarge = errors.New("request exceeds maximum size")
+
+type Publisher interface {
+	PublishCaptureFrame(context.Context, string, contract.CaptureFrameCommand) error
+	PublishRecordingWindow(context.Context, string, contract.RecordingWindowCommand) error
+}
+
+type Processor interface {
+	Process(context.Context, contract.FrameMetadata, []byte) (processor.Decision, error)
+}
+
+type Config struct {
+	APIToken         string
+	MaxFrameBytes    int64
+	MaxFrameTTL      time.Duration
+	CommandTTL       time.Duration
+	PreRollSeconds   int64
+	EventClipSeconds int64
+}
+
+type Service struct {
+	config    Config
+	processor Processor
+	publisher Publisher
+	now       func() time.Time
+	newID     func() string
+
+	resultsMu sync.Mutex
+	results   map[string]*frameResult
+}
+
+type frameResult struct {
+	done      chan struct{}
+	expiresAt int64
+	response  FrameResponse
+	err       error
+}
+
+type FrameResponse struct {
+	SchemaVersion string `json:"schemaVersion"`
+	RequestID     string `json:"requestId"`
+	FrameID       string `json:"frameId"`
+	Decision      string `json:"decision"`
+	Reason        string `json:"reason"`
+}
+
+type FrameRequestResponse struct {
+	SchemaVersion string   `json:"schemaVersion"`
+	RequestID     string   `json:"requestId"`
+	DeviceIDs     []string `json:"deviceIds"`
+	Status        string   `json:"status"`
+}
+
+func New(config Config, frameProcessor Processor, publisher Publisher, newID func() string) *Service {
+	if config.MaxFrameBytes <= 0 {
+		config.MaxFrameBytes = 4 << 20
+	}
+	if config.CommandTTL <= 0 {
+		config.CommandTTL = 30 * time.Second
+	}
+	if config.MaxFrameTTL <= 0 {
+		config.MaxFrameTTL = 5 * time.Minute
+	}
+	if config.EventClipSeconds <= 0 {
+		config.EventClipSeconds = 30
+	}
+	if newID == nil {
+		newID = func() string { return fmt.Sprintf("request-%d", time.Now().UnixNano()) }
+	}
+	return &Service{
+		config: config, processor: frameProcessor, publisher: publisher,
+		now: time.Now, newID: newID, results: make(map[string]*frameResult),
+	}
+}
+
+func (s *Service) Handler() http.Handler {
+	mux := http.NewServeMux()
+	mux.HandleFunc("GET /health", s.handleHealth)
+	mux.HandleFunc("POST /v1/frames", s.authorize(s.handleFrame))
+	mux.HandleFunc("POST /v1/frame-requests", s.authorize(s.handleFrameRequest))
+	return mux
+}
+
+func (s *Service) authorize(next http.HandlerFunc) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		if s.config.APIToken == "" {
+			writeError(w, http.StatusServiceUnavailable, "service authentication is not configured")
+			return
+		}
+		provided := strings.TrimPrefix(r.Header.Get("Authorization"), "Bearer ")
+		if subtle.ConstantTimeCompare([]byte(provided), []byte(s.config.APIToken)) != 1 {
+			writeError(w, http.StatusUnauthorized, "unauthorized")
+			return
+		}
+		next(w, r)
+	}
+}
+
+func (s *Service) handleHealth(w http.ResponseWriter, _ *http.Request) {
+	writeJSON(w, http.StatusOK, map[string]string{"status": "healthy"})
+}
+
+func (s *Service) handleFrame(w http.ResponseWriter, r *http.Request) {
+	metadata, frame, err := readFrame(w, r, s.config.MaxFrameBytes)
+	if err != nil {
+		if errors.Is(err, errRequestTooLarge) {
+			writeError(w, http.StatusRequestEntityTooLarge, err.Error())
+			return
+		}
+		writeError(w, http.StatusBadRequest, err.Error())
+		return
+	}
+	nowMillis := s.now().UnixMilli()
+	if err := metadata.Validate(nowMillis); err != nil {
+		writeError(w, http.StatusUnprocessableEntity, err.Error())
+		return
+	}
+	if time.Duration(metadata.ExpiresAt-nowMillis)*time.Millisecond > s.config.MaxFrameTTL {
+		writeError(w, http.StatusUnprocessableEntity, "expiresAt exceeds maximum frame TTL")
+		return
+	}
+	imageConfig, _, err := image.DecodeConfig(bytes.NewReader(frame))
+	if err != nil {
+		writeError(w, http.StatusUnsupportedMediaType, "frame must be a valid JPEG")
+		return
+	}
+	if imageConfig.Width != metadata.Width || imageConfig.Height != metadata.Height {
+		writeError(w, http.StatusUnprocessableEntity, "frame dimensions do not match metadata")
+		return
+	}
+
+	result, owner := s.beginFrame(metadata.FrameID, metadata.ExpiresAt, nowMillis)
+	if !owner {
+		select {
+		case <-r.Context().Done():
+			writeError(w, http.StatusRequestTimeout, "request cancelled")
+			return
+		case <-result.done:
+		}
+		if result.err != nil {
+			writeError(w, http.StatusBadGateway, result.err.Error())
+			return
+		}
+		writeJSON(w, http.StatusOK, result.response)
+		return
+	}
+
+	response, processErr := s.processFrame(r.Context(), metadata, frame)
+	s.finishFrame(result, response, processErr)
+	if processErr != nil {
+		writeError(w, http.StatusBadGateway, processErr.Error())
+		return
+	}
+	writeJSON(w, http.StatusOK, response)
+}
+
+func (s *Service) processFrame(ctx context.Context, metadata contract.FrameMetadata, frame []byte) (FrameResponse, error) {
+	decision, err := s.processor.Process(ctx, metadata, frame)
+	if err != nil {
+		return FrameResponse{}, fmt.Errorf("process frame: %w", err)
+	}
+	response := FrameResponse{
+		SchemaVersion: contract.SchemaVersion,
+		RequestID:     metadata.RequestID,
+		FrameID:       metadata.FrameID,
+		Decision:      "no-event",
+		Reason:        decision.Reason,
+	}
+	if !decision.Triggered {
+		return response, nil
+	}
+	command := contract.RecordingWindowCommand{
+		SchemaVersion:     contract.SchemaVersion,
+		RequestID:         metadata.RequestID,
+		FrameID:           metadata.FrameID,
+		CapturedAt:        metadata.CapturedAt,
+		PreRollSeconds:    s.config.PreRollSeconds,
+		EventClipSeconds:  s.config.EventClipSeconds,
+		ExpiresAt:         s.now().Add(s.config.CommandTTL).UnixMilli(),
+		ProcessingProfile: metadata.ProcessingProfile,
+		TraceID:           metadata.TraceID,
+	}
+	if err := command.Validate(s.now().UnixMilli()); err != nil {
+		return FrameResponse{}, fmt.Errorf("build recording command: %w", err)
+	}
+	if err := s.publisher.PublishRecordingWindow(ctx, metadata.DeviceID, command); err != nil {
+		return FrameResponse{}, fmt.Errorf("publish recording command: %w", err)
+	}
+	response.Decision = "event"
+	return response, nil
+}
+
+func (s *Service) handleFrameRequest(w http.ResponseWriter, r *http.Request) {
+	var request contract.FrameRequest
+	decoder := json.NewDecoder(http.MaxBytesReader(w, r.Body, maxMetadataBytes))
+	decoder.DisallowUnknownFields()
+	if err := decoder.Decode(&request); err != nil {
+		writeError(w, http.StatusBadRequest, "invalid JSON request")
+		return
+	}
+	if request.RequestID == "" {
+		request.RequestID = s.newID()
+	}
+	nowMillis := s.now().UnixMilli()
+	if err := request.Validate(nowMillis); err != nil {
+		writeError(w, http.StatusUnprocessableEntity, err.Error())
+		return
+	}
+	if time.Duration(request.ExpiresAt-nowMillis)*time.Millisecond > s.config.MaxFrameTTL {
+		writeError(w, http.StatusUnprocessableEntity, "expiresAt exceeds maximum frame TTL")
+		return
+	}
+	for _, deviceID := range request.DeviceIDs {
+		command := contract.CaptureFrameCommand{
+			SchemaVersion:     contract.SchemaVersion,
+			RequestID:         request.RequestID,
+			ProcessingProfile: request.ProcessingProfile,
+			ExpiresAt:         request.ExpiresAt,
+			TraceID:           request.TraceID,
+		}
+		if err := s.publisher.PublishCaptureFrame(r.Context(), deviceID, command); err != nil {
+			writeError(w, http.StatusBadGateway, "failed to publish capture command")
+			return
+		}
+	}
+	writeJSON(w, http.StatusAccepted, FrameRequestResponse{
+		SchemaVersion: contract.SchemaVersion,
+		RequestID:     request.RequestID,
+		DeviceIDs:     request.DeviceIDs,
+		Status:        "accepted",
+	})
+}
+
+func (s *Service) beginFrame(frameID string, expiresAt, nowMillis int64) (*frameResult, bool) {
+	s.resultsMu.Lock()
+	defer s.resultsMu.Unlock()
+	for id, result := range s.results {
+		if result.expiresAt <= nowMillis {
+			delete(s.results, id)
+		}
+	}
+	if result, ok := s.results[frameID]; ok {
+		return result, false
+	}
+	result := &frameResult{done: make(chan struct{}), expiresAt: expiresAt}
+	s.results[frameID] = result
+	delay := time.Duration(expiresAt-nowMillis) * time.Millisecond
+	time.AfterFunc(delay, func() {
+		s.resultsMu.Lock()
+		if s.results[frameID] == result {
+			delete(s.results, frameID)
+		}
+		s.resultsMu.Unlock()
+	})
+	return result, true
+}
+
+func (s *Service) finishFrame(result *frameResult, response FrameResponse, err error) {
+	s.resultsMu.Lock()
+	result.response = response
+	result.err = err
+	close(result.done)
+	s.resultsMu.Unlock()
+}
+
+func readFrame(w http.ResponseWriter, r *http.Request, maxFrameBytes int64) (contract.FrameMetadata, []byte, error) {
+	r.Body = http.MaxBytesReader(w, r.Body, maxFrameBytes+maxMetadataBytes)
+	reader, err := r.MultipartReader()
+	if err != nil {
+		return contract.FrameMetadata{}, nil, errors.New("content type must be multipart/form-data")
+	}
+	var metadata contract.FrameMetadata
+	var frame []byte
+	for {
+		part, err := reader.NextPart()
+		if errors.Is(err, io.EOF) {
+			break
+		}
+		if err != nil {
+			var tooLarge *http.MaxBytesError
+			if errors.As(err, &tooLarge) {
+				return contract.FrameMetadata{}, nil, errRequestTooLarge
+			}
+			return contract.FrameMetadata{}, nil, errors.New("invalid multipart body")
+		}
+		switch part.FormName() {
+		case "metadata":
+			if err := decodeMetadataPart(part, &metadata); err != nil {
+				return contract.FrameMetadata{}, nil, err
+			}
+		case "frame":
+			if part.Header.Get("Content-Type") != "image/jpeg" {
+				return contract.FrameMetadata{}, nil, errors.New("frame content type must be image/jpeg")
+			}
+			frame, err = io.ReadAll(io.LimitReader(part, maxFrameBytes+1))
+			if err != nil || int64(len(frame)) > maxFrameBytes {
+				return contract.FrameMetadata{}, nil, errRequestTooLarge
+			}
+		}
+	}
+	if metadata.FrameID == "" || len(frame) == 0 {
+		return contract.FrameMetadata{}, nil, errors.New("metadata and frame parts are required")
+	}
+	return metadata, frame, nil
+}
+
+func decodeMetadataPart(part *multipart.Part, target *contract.FrameMetadata) error {
+	value, err := io.ReadAll(io.LimitReader(part, maxMetadataBytes+1))
+	if err != nil || len(value) > maxMetadataBytes {
+		return errRequestTooLarge
+	}
+	decoder := json.NewDecoder(bytes.NewReader(value))
+	decoder.DisallowUnknownFields()
+	if err := decoder.Decode(target); err != nil {
+		return errors.New("invalid metadata JSON")
+	}
+	return nil
+}
+
+func writeError(w http.ResponseWriter, status int, message string) {
+	writeJSON(w, status, map[string]any{
+		"schemaVersion": contract.SchemaVersion,
+		"error":         message,
+	})
+}
+
+func writeJSON(w http.ResponseWriter, status int, value any) {
+	w.Header().Set("Content-Type", "application/json")
+	w.WriteHeader(status)
+	if err := json.NewEncoder(w).Encode(value); err != nil {
+		slog.Error("failed to encode HTTP response", "error", err)
+	}
+}
@@ -0,0 +1,267 @@
+package service
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"image"
+	"image/color"
+	"image/jpeg"
+	"mime/multipart"
+	"net/http"
+	"net/http/httptest"
+	"net/textproto"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/kerberos-io/agent/examples/frame-processor/contract"
+	"github.com/kerberos-io/agent/examples/frame-processor/processor"
+)
+
+type recordingPublish struct {
+	deviceID string
+	command  contract.RecordingWindowCommand
+}
+
+type fakePublisher struct {
+	mu         sync.Mutex
+	captures   []contract.CaptureFrameCommand
+	recordings []recordingPublish
+}
+
+type blockingProcessor struct {
+	started chan struct{}
+	release chan struct{}
+	mu      sync.Mutex
+	calls   int
+}
+
+func (p *blockingProcessor) Process(ctx context.Context, _ contract.FrameMetadata, _ []byte) (processor.Decision, error) {
+	p.mu.Lock()
+	p.calls++
+	if p.calls == 1 {
+		close(p.started)
+	}
+	p.mu.Unlock()
+	select {
+	case <-ctx.Done():
+		return processor.Decision{}, ctx.Err()
+	case <-p.release:
+		return processor.Decision{Triggered: true, Reason: "test"}, nil
+	}
+}
+
+func (p *fakePublisher) PublishCaptureFrame(_ context.Context, _ string, command contract.CaptureFrameCommand) error {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	p.captures = append(p.captures, command)
+	return nil
+}
+
+func (p *fakePublisher) PublishRecordingWindow(_ context.Context, deviceID string, command contract.RecordingWindowCommand) error {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	p.recordings = append(p.recordings, recordingPublish{deviceID: deviceID, command: command})
+	return nil
+}
+
+func TestFrameAlwaysTriggerPublishesOneIdempotentRecordingCommand(t *testing.T) {
+	publisher := &fakePublisher{}
+	service := New(Config{
+		APIToken: "secret", CommandTTL: time.Minute,
+		PreRollSeconds: 10, EventClipSeconds: 30,
+	}, processor.New(processor.Config{}), publisher, nil)
+	service.now = func() time.Time { return time.UnixMilli(1_000) }
+	server := httptest.NewServer(service.Handler())
+	defer server.Close()
+
+	metadata := validMetadata()
+	for range 2 {
+		response := postFrame(t, server.URL, "secret", metadata, jpegFrame(t, 2, 2, 255))
+		if response.StatusCode != http.StatusOK {
+			t.Fatalf("POST /v1/frames status = %d", response.StatusCode)
+		}
+		response.Body.Close()
+	}
+
+	if got := len(publisher.recordings); got != 1 {
+		t.Fatalf("recording commands = %d, want 1", got)
+	}
+	published := publisher.recordings[0]
+	if published.deviceID != metadata.DeviceID || published.command.CapturedAt != metadata.CapturedAt {
+		t.Fatalf("published command = %#v", published)
+	}
+}
+
+func TestFrameRequestPublishesCaptureCommand(t *testing.T) {
+	publisher := &fakePublisher{}
+	service := New(Config{APIToken: "secret"}, processor.New(processor.Config{}), publisher, func() string { return "generated-request" })
+	service.now = func() time.Time { return time.UnixMilli(1_000) }
+	server := httptest.NewServer(service.Handler())
+	defer server.Close()
+
+	body := `{"schemaVersion":"1.0","deviceIds":["device-1"],"processingProfile":"always-trigger","expiresAt":2000}`
+	request, err := http.NewRequest(http.MethodPost, server.URL+"/v1/frame-requests", strings.NewReader(body))
+	if err != nil {
+		t.Fatal(err)
+	}
+	request.Header.Set("Authorization", "Bearer secret")
+	request.Header.Set("Content-Type", "application/json")
+	response, err := http.DefaultClient.Do(request)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer response.Body.Close()
+	if response.StatusCode != http.StatusAccepted {
+		t.Fatalf("POST /v1/frame-requests status = %d", response.StatusCode)
+	}
+	if got := len(publisher.captures); got != 1 || publisher.captures[0].RequestID != "generated-request" {
+		t.Fatalf("capture commands = %#v", publisher.captures)
+	}
+}
+
+func TestFrameRejectsUnauthorizedRequest(t *testing.T) {
+	service := New(Config{APIToken: "secret"}, processor.New(processor.Config{}), &fakePublisher{}, nil)
+	server := httptest.NewServer(service.Handler())
+	defer server.Close()
+
+	response := postFrame(t, server.URL, "wrong", validMetadata(), jpegFrame(t, 2, 2, 255))
+	defer response.Body.Close()
+	if response.StatusCode != http.StatusUnauthorized {
+		t.Fatalf("POST /v1/frames status = %d", response.StatusCode)
+	}
+}
+
+func TestFrameFailsClosedWithoutConfiguredToken(t *testing.T) {
+	application := New(Config{}, processor.New(processor.Config{}), &fakePublisher{}, nil)
+	server := httptest.NewServer(application.Handler())
+	defer server.Close()
+
+	response := postFrame(t, server.URL, "", validMetadata(), jpegFrame(t, 2, 2, 255))
+	defer response.Body.Close()
+	if response.StatusCode != http.StatusServiceUnavailable {
+		t.Fatalf("POST /v1/frames status = %d", response.StatusCode)
+	}
+}
+
+func TestConcurrentDuplicateFramesPublishOneRecordingCommand(t *testing.T) {
+	publisher := &fakePublisher{}
+	frameProcessor := &blockingProcessor{started: make(chan struct{}), release: make(chan struct{})}
+	application := New(Config{
+		APIToken: "secret", CommandTTL: time.Minute,
+		PreRollSeconds: 10, EventClipSeconds: 30,
+	}, frameProcessor, publisher, nil)
+	application.now = func() time.Time { return time.UnixMilli(1_000) }
+	server := httptest.NewServer(application.Handler())
+	defer server.Close()
+
+	metadata := validMetadata()
+	statuses := make(chan int, 2)
+	go func() {
+		response := postFrame(t, server.URL, "secret", metadata, jpegFrame(t, 2, 2, 255))
+		defer response.Body.Close()
+		statuses <- response.StatusCode
+	}()
+	<-frameProcessor.started
+	go func() {
+		response := postFrame(t, server.URL, "secret", metadata, jpegFrame(t, 2, 2, 255))
+		defer response.Body.Close()
+		statuses <- response.StatusCode
+	}()
+	close(frameProcessor.release)
+
+	for range 2 {
+		if status := <-statuses; status != http.StatusOK {
+			t.Fatalf("POST /v1/frames status = %d", status)
+		}
+	}
+	if got := len(publisher.recordings); got != 1 {
+		t.Fatalf("recording commands = %d, want 1", got)
+	}
+	frameProcessor.mu.Lock()
+	defer frameProcessor.mu.Unlock()
+	if frameProcessor.calls != 1 {
+		t.Fatalf("processor calls = %d, want 1", frameProcessor.calls)
+	}
+}
+
+func TestFrameRejectsTTLAboveConfiguredMaximum(t *testing.T) {
+	application := New(Config{APIToken: "secret", MaxFrameTTL: time.Second}, processor.New(processor.Config{}), &fakePublisher{}, nil)
+	application.now = func() time.Time { return time.UnixMilli(1_000) }
+	server := httptest.NewServer(application.Handler())
+	defer server.Close()
+
+	metadata := validMetadata()
+	metadata.ExpiresAt = 2_001
+	response := postFrame(t, server.URL, "secret", metadata, jpegFrame(t, 2, 2, 255))
+	defer response.Body.Close()
+	if response.StatusCode != http.StatusUnprocessableEntity {
+		t.Fatalf("POST /v1/frames status = %d", response.StatusCode)
+	}
+}
+
+func validMetadata() contract.FrameMetadata {
+	return contract.FrameMetadata{
+		SchemaVersion: contract.SchemaVersion,
+		RequestID:     "request-1", FrameID: "frame-1", DeviceID: "device-1",
+		CapturedAt: 900, ExpiresAt: 2_000, ProcessingProfile: processor.ProfileAlwaysTrigger,
+		SourceStream: "sub", Width: 2, Height: 2,
+	}
+}
+
+func jpegFrame(t *testing.T, width, height int, brightness uint8) []byte {
+	t.Helper()
+	frame := image.NewGray(image.Rect(0, 0, width, height))
+	for index := range frame.Pix {
+		frame.Pix[index] = brightness
+	}
+	frame.SetGray(0, 0, color.Gray{Y: brightness})
+	var output bytes.Buffer
+	if err := jpeg.Encode(&output, frame, nil); err != nil {
+		t.Fatal(err)
+	}
+	return output.Bytes()
+}
+
+func postFrame(t *testing.T, baseURL, token string, metadata contract.FrameMetadata, frame []byte) *http.Response {
+	t.Helper()
+	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")
+	part, err := writer.CreatePart(metadataHeader)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := json.NewEncoder(part).Encode(metadata); err != nil {
+		t.Fatal(err)
+	}
+	frameHeader := make(textproto.MIMEHeader)
+	frameHeader.Set("Content-Disposition", `form-data; name="frame"; filename="frame.jpg"`)
+	frameHeader.Set("Content-Type", "image/jpeg")
+	part, err = writer.CreatePart(frameHeader)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := part.Write(frame); err != nil {
+		t.Fatal(err)
+	}
+	if err := writer.Close(); err != nil {
+		t.Fatal(err)
+	}
+
+	request, err := http.NewRequest(http.MethodPost, baseURL+"/v1/frames", &body)
+	if err != nil {
+		t.Fatal(err)
+	}
+	request.Header.Set("Authorization", "Bearer "+token)
+	request.Header.Set("Content-Type", writer.FormDataContentType())
+	response, err := http.DefaultClient.Do(request)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return response
+}
@@ -0,0 +1,7 @@
+{
+  "schemaVersion": "1.0",
+  "requestId": "request-example-1",
+  "processingProfile": "always-trigger",
+  "expiresAt": 4102444800000,
+  "traceId": "trace-example-1"
+}
@@ -0,0 +1,8 @@
+{
+  "schemaVersion": "1.0",
+  "requestId": "request-example-1",
+  "deviceIds": ["camera-1"],
+  "processingProfile": "always-trigger",
+  "expiresAt": 4102444800000,
+  "traceId": "trace-example-1"
+}
@@ -0,0 +1,11 @@
+{
+  "schemaVersion": "1.0",
+  "requestId": "request-example-1",
+  "frameId": "frame-example-1",
+  "capturedAt": 1789380000123,
+  "preRollSeconds": 10,
+  "eventClipSeconds": 30,
+  "expiresAt": 4102444800000,
+  "processingProfile": "always-trigger",
+  "traceId": "trace-example-1"
+}
@@ -1296,9 +1296,6 @@ const docTemplate = `{
                    "description": "old way, remove this",
                    "type": "string"
                },
-                "custom_headers": {
-                    "type": "string"
-                },
                "directory": {
                    "type": "string"
                },
@@ -1289,9 +1289,6 @@
                    "description": "old way, remove this",
                    "type": "string"
                },
-                "custom_headers": {
-                    "type": "string"
-                },
                "directory": {
                    "type": "string"
                },
@@ -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.3
+	github.com/kerberos-io/onvif v1.2.2
 	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
@@ -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.3 h1:RDnDvLBd92VCgFOrlbGavzCVN9lQA/qs9novdNFIAOU=
-github.com/kerberos-io/onvif v1.2.3/go.mod h1:XSgEQXmEDjUQTbdXvsaRJt6Az8YPGj7L+j5iXKEGijU=
+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/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=
@@ -1070,10 +1070,6 @@ 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
@@ -1117,10 +1113,6 @@ 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
@@ -1,68 +0,0 @@
-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")
-					}
-				})
-			}
-		}
-	}
-}
@@ -541,23 +541,13 @@ func (state *heartbeatONVIFState) fetchEvents(device *goonvif.Device) ([]byte, b
 		}
 	}

-	// 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) {
+	if !initialStateSubscriptionCreated && state.loopPullPoint == "" {
 		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()
@@ -15,7 +15,6 @@ 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) {
@@ -591,60 +590,3 @@ 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)
-	}
-}
@@ -1,195 +0,0 @@
-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
-}
@@ -1,100 +0,0 @@
-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)
-	}
-}
@@ -178,10 +178,6 @@ 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)
@@ -207,9 +203,6 @@ 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)
@@ -274,7 +267,7 @@ func newVaultHTTPClient(timeout time.Duration) *http.Client {
 		transport.TLSClientConfig.InsecureSkipVerify = true
 	}

-	client := &http.Client{Transport: transport, CheckRedirect: stripVaultHeadersOnCrossHostRedirect}
+	client := &http.Client{Transport: transport}
 	if timeout > 0 {
 		client.Timeout = timeout
 	}
@@ -141,6 +141,19 @@ 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)
@@ -298,10 +311,6 @@ 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,
@@ -312,22 +321,10 @@ 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)
 }
@@ -3,7 +3,6 @@ package cloud
 import (
 	"bytes"
 	"encoding/base64"
-	"encoding/json"
 	"fmt"
 	"io"
 	"net"
@@ -305,9 +304,7 @@ 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}`)

-	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")
+	uploaded, responded, supported, _, err := uploadVaultResumable(testVault(ts.URL), "pk", "dev", fileName, "test", "primary")
 	if err != nil {
 		t.Fatalf("unexpected error: %v", err)
 	}
@@ -331,23 +328,6 @@ 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) {
@@ -315,15 +315,6 @@ 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{}
@@ -612,9 +603,6 @@ 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":
@@ -646,9 +634,6 @@ 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":
@@ -3,7 +3,6 @@ package config
 import (
 	"context"
 	"errors"
-	"os"
 	"testing"
 	"time"

@@ -12,53 +11,6 @@ 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
@@ -88,22 +40,6 @@ 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
 }
@@ -187,7 +187,6 @@ 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"`
 }
@@ -5,9 +5,7 @@ import (
 	"encoding/json"
 	"encoding/xml"
 	"errors"
-	"fmt"
 	"io"
-	"math"
 	"net/http"
 	"strconv"
 	"strings"
@@ -132,7 +130,7 @@ func HandleONVIFActions(configuration *models.Configuration, communication *mode

 					if onvifAction.Action == "absolute-move" {

-						// Move the camera to the saved absolute position.
+						// We will move the camera to zero position.
 						x := ptzAction.X
 						y := ptzAction.Y
 						z := ptzAction.Z
@@ -143,13 +141,42 @@ func HandleONVIFActions(configuration *models.Configuration, communication *mode
 						// Log functions
 						log.Debug("onvif.HandleONVIFActions(): functions: " + strings.Join(functions, ", "))

-						err = MoveToPosition(device, configurations, token, x, y, z)
+						// 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)
 						if err != nil {
-							log.Error("onvif.HandleONVIFActions() - MoveToPosition: " + err.Error())
+							log.Debug("onvif.HandleONVIFActions() - AbsolutePanTitleMoveFake: " + err.Error())
 						} else {
-							log.Info("onvif.HandleONVIFActions() - MoveToPosition: successfully moved camera.")
+							log.Info("onvif.HandleONVIFActions() - AbsolutePanTitleMoveFake: 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
@@ -444,20 +471,16 @@ 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: ptzConfiguration.DefaultAbsolutePantTiltPositionSpace,
+		Space: configuration.PTZConfiguration[0].DefaultAbsolutePantTiltPositionSpace,
 	}

 	absoluteZoomVector := xsdonvif.Vector1D{
 		X:     zoom,
-		Space: ptzConfiguration.DefaultAbsoluteZoomPositionSpace,
+		Space: configuration.PTZConfiguration[0].DefaultAbsoluteZoomPositionSpace,
 	}

 	resp, err := device.CallMethod(ptz.AbsoluteMove{
@@ -470,325 +493,240 @@ func AbsolutePanTiltMove(device *onvif.Device, configuration ptz.GetConfiguratio
 	return handleONVIFResponse("absolute_move", resp, err)
 }

-type ptzMoveClient interface {
-	GetPosition() (xsdonvif.PTZVector, error)
-	AbsoluteMove(ptzTarget) error
-	ContinuousMove(ptzVelocity) error
-	Stop() error
-}
+// 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 {

-type ptzTarget struct {
-	Pan  float64
-	Tilt float64
-	Zoom float64
-}
+		// 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

-type ptzVelocity struct {
-	Pan  float64
-	Tilt float64
-	Zoom float64
-}
+		// 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)

-type ptzMoveSupport struct {
-	Absolute   bool
-	Continuous bool
-}
+		// Now we'll move a bit slower to make sure we are ok (will be more accurate)
+		speed = 0.1
+		wait = 200 * time.Millisecond

-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
-}
+		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)

-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},
+		return err
 	}
-	return AbsolutePanTiltMove(client.device, configuration, client.token, target.Pan, target.Tilt, target.Zoom)
+	return err
 }

-func (client *devicePTZMoveClient) ContinuousMove(velocity ptzVelocity) error {
-	panTilt := xsdonvif.Vector2D{
-		X:     velocity.Pan,
-		Y:     velocity.Tilt,
-		Space: client.configuration.DefaultContinuousPanTiltVelocitySpace,
+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,
 	}
-	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,
+	_, err := device.CallMethod(ptz.ContinuousMove{
+		ProfileToken: &token,
+		Velocity: xsdonvif.PTZSpeedZoom{
+			Zoom: zoomOut,
 		},
 	})
-	return handleONVIFResponse("continuous_move_to_position", resp, err)
-}
+	if err != nil {
+		log.Error("ZoomOutCompletely: " + err.Error())
+	}

-func (client *devicePTZMoveClient) Stop() error {
-	resp, err := client.device.CallMethod(ptz.Stop{
-		ProfileToken: client.token,
-		PanTilt:      true,
+	for {
+		position, _ := GetPosition(device, token)
+		if position.Zoom.X == 0 {
+			break
+		}
+		time.Sleep(250 * time.Millisecond)
+	}
+
+	_, err = device.CallMethod(ptz.Stop{
+		ProfileToken: token,
 		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 {
-		return fmt.Errorf("read initial PTZ position: %w", err)
+		log.Error("ZoomOutCompletely: " + err.Error())
 	}
-	_, _, 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
+	return err
 }

-func waitForPTZTarget(client ptzMoveClient, target ptzTarget, timeout time.Duration, options ptzMoveOptions) error {
-	deadline := time.Now().Add(timeout)
-	lastProgress := time.Now()
-	bestDistance := math.Inf(1)
+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)

-	for {
-		position, err := client.GetPosition()
-		if err != nil {
-			return fmt.Errorf("read PTZ position: %w", err)
-		}
-		_, distance, reached, err := ptzTargetError(position, target, options)
-		if err != nil {
-			return err
-		}
-		if reached {
-			return nil
+	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()
-		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)
-	}
-}
-
-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))
-		}
-	}()
-
-	deadline := time.Now().Add(options.ContinuousTimeout)
-	lastProgress := time.Now()
-	bestDistance := math.Inf(1)
-	var lastVelocity ptzVelocity
-	haveVelocity := false
-
-	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
-		}
-
-		now := time.Now()
-		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)
-		}
-
-		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)
+		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
 			}
-			lastVelocity = velocity
-			haveVelocity = true
+			if time.Since(now) > 3*time.Second {
+				break
+			}
+			time.Sleep(wait)
 		}
-		time.Sleep(options.PollInterval)
-	}
-}

-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
-}
+		_, err = device.CallMethod(ptz.Stop{
+			ProfileToken: token,
+			PanTilt:      true,
+			Zoom:         true,
+		})

-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)
-}
-
-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
-}
-
-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)
+		if err != nil {
+			log.Error("ContinuousPanTiltMove (Pan): " + err.Error())
 		}
 	}
-	return nil
+	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)
+
+	if position.PanTilt.Y >= tilt-0.005 && position.PanTilt.Y <= tilt+0.005 {
+
+	} 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
+		}
+
+		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)
+		}
+
+		_, err = device.CallMethod(ptz.Stop{
+			ProfileToken: token,
+			PanTilt:      true,
+			Zoom:         true,
+		})
+
+		if err != nil {
+			log.Error("ContinuousPanTiltMove (Tilt): " + err.Error())
+		}
+	}
+	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)
+
+	if position.Zoom.X >= zoom-0.005 && position.Zoom.X <= zoom+0.005 {
+
+	} 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.
+		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())
+		}
+	}
+	return err
 }

 func ContinuousPanTilt(device *onvif.Device, configuration ptz.GetConfigurationsResponse, token xsdonvif.ReferenceToken, pan float64, tilt float64) error {
@@ -957,10 +895,6 @@ 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
@@ -1108,23 +1042,6 @@ 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
@@ -1138,7 +1055,7 @@ func UnsubscribePullPoint(dev *onvif.Device, pullPointAddress string) error {
 		return err
 	}

-	res, err := dev.SendSoapWithHeader(pullPointAddress, string(requestBody), subscriptionAddressingHeader(pullPointAddress, unsubscribeAction))
+	res, err := dev.SendSoap(pullPointAddress, string(requestBody))
 	if err != nil {
 		log.WithError(err).WithFields(log.Fields{
 			"component": "onvif",
@@ -1220,7 +1137,7 @@ func GetEventMessages(dev *onvif.Device, pullPointAddress string) ([]ONVIFEvents
 				log.Error("onvif.main.GetEventMessages(pullMessages): " + err.Error())
 				return eventsArray, err
 			}
-			res, err := dev.SendSoapWithHeader(string(subscriptionURI), string(requestBody), subscriptionAddressingHeader(subscriptionURI, pullMessagesAction))
+			res, err := dev.SendSoap(string(subscriptionURI), string(requestBody))
 			if err != nil {
 				log.Error("onvif.main.GetEventMessages(pullMessages): " + err.Error())
 				return eventsArray, err
@@ -1,264 +0,0 @@
-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)
-	}
-}
@@ -1,88 +0,0 @@
-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)
-}
@@ -223,8 +223,6 @@
      "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",
@@ -2542,28 +2542,6 @@ 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(
@@ -2837,28 +2815,6 @@ 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