feat: add frame processor reference service

.github/workflows/go.yml

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

README.md

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

examples/frame-processor/Dockerfile

@@ -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"]

examples/frame-processor/MQTT.md

@@ -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`.

examples/frame-processor/README.md

@@ -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 ./...
+```

examples/frame-processor/contract/contract.go

@@ -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"`
+}

examples/frame-processor/contract/contract_test.go

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

examples/frame-processor/go.mod

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

examples/frame-processor/go.sum

@@ -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=

examples/frame-processor/main.go

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

examples/frame-processor/main_test.go

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

examples/frame-processor/mqttpublisher/publisher.go

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

examples/frame-processor/mqttpublisher/publisher_test.go

@@ -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()              {}

examples/frame-processor/openapi.yaml

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

examples/frame-processor/processor/processor.go

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

examples/frame-processor/processor/processor_test.go

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

examples/frame-processor/service/service.go

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

examples/frame-processor/service/service_test.go

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

examples/frame-processor/testdata/capture-frame.json

@@ -0,0 +1,7 @@
+{
+  "schemaVersion": "1.0",
+  "requestId": "request-example-1",
+  "processingProfile": "always-trigger",
+  "expiresAt": 4102444800000,
+  "traceId": "trace-example-1"
+}

examples/frame-processor/testdata/frame-request.json

@@ -0,0 +1,8 @@
+{
+  "schemaVersion": "1.0",
+  "requestId": "request-example-1",
+  "deviceIds": ["camera-1"],
+  "processingProfile": "always-trigger",
+  "expiresAt": 4102444800000,
+  "traceId": "trace-example-1"
+}

examples/frame-processor/testdata/request-recording-window.json

@@ -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"
+}

machinery/data/config/config.json

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

machinery/src/cloud/frameprocessing/frameprocessing.go

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

machinery/src/cloud/frameprocessing/frameprocessing_test.go

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

machinery/src/components/kerberos.go

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

machinery/src/config/main.go

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

machinery/src/config/main_test.go

@@ -44,78 +44,6 @@ func intPointer(value int) *int {
 	return &value
 }
 
-func TestApplyAgentEnvVarsFrameProcessing(t *testing.T) {
-	t.Setenv("AGENT_FRAME_PROCESSING_ENABLED", "true")
-	t.Setenv("AGENT_FRAME_PROCESSING_ENDPOINT", "http://processor:8080/v1/frames")
-	t.Setenv("AGENT_FRAME_PROCESSING_TOKEN", "secret")
-	t.Setenv("AGENT_FRAME_PROCESSING_PROFILE", "always-trigger")
-	t.Setenv("AGENT_FRAME_PROCESSING_ALLOW_REQUESTED_FRAMES", "true")
-	t.Setenv("AGENT_FRAME_PROCESSING_STREAM", "sub")
-	t.Setenv("AGENT_FRAME_PROCESSING_INTERVAL_SECONDS", "15")
-	t.Setenv("AGENT_FRAME_PROCESSING_WIDTH", "320")
-	t.Setenv("AGENT_FRAME_PROCESSING_HEIGHT", "180")
-	t.Setenv("AGENT_FRAME_PROCESSING_JPEG_QUALITY", "80")
-	t.Setenv("AGENT_FRAME_PROCESSING_REQUEST_TIMEOUT_SECONDS", "7")
-	t.Setenv("AGENT_FRAME_PROCESSING_FRAME_TTL_SECONDS", "45")
-	t.Setenv("AGENT_FRAME_PROCESSING_MAX_FRAME_BYTES", "2097152")
-	t.Setenv("AGENT_FRAME_PROCESSING_PERIODIC_QUEUE_CAPACITY", "2")
-
-	configuration := &models.Configuration{}
-	initConfigPointers(&configuration.Config)
-	applyAgentEnvVars(configuration, "", true)
-
-	got := configuration.Config.FrameProcessing
-	if got == nil {
-		t.Fatal("FrameProcessing is nil")
-	}
-	if got.Enabled != "true" || got.Endpoint != "http://processor:8080/v1/frames" || got.Token != "secret" {
-		t.Fatalf("FrameProcessing identity = %+v", got)
-	}
-	if got.Profile != "always-trigger" || got.AllowRequestedFrames != "true" || got.Stream != "sub" || got.IntervalSeconds != 15 {
-		t.Fatalf("FrameProcessing schedule = %+v", got)
-	}
-	if got.Width != 320 || got.Height != 180 || got.JPEGQuality != 80 {
-		t.Fatalf("FrameProcessing image = %+v", got)
-	}
-	if got.RequestTimeoutSeconds != 7 || got.FrameTTLSeconds != 45 || got.MaxFrameBytes != 2097152 || got.PeriodicQueueCapacity != 2 {
-		t.Fatalf("FrameProcessing delivery = %+v", got)
-	}
-}
-
-func TestApplyAgentEnvVarsFrameProcessingDefaults(t *testing.T) {
-	configuration := &models.Configuration{}
-	initConfigPointers(&configuration.Config)
-	applyAgentEnvVars(configuration, "", true)
-
-	got := configuration.Config.FrameProcessing
-	if got.Profile != "never-trigger" || got.Stream != "auto" || got.IntervalSeconds != 10 {
-		t.Fatalf("FrameProcessing defaults = %+v", got)
-	}
-	if got.Width != 640 || got.Height != 0 || got.JPEGQuality != 70 {
-		t.Fatalf("FrameProcessing image defaults = %+v", got)
-	}
-	if got.RequestTimeoutSeconds != 5 || got.FrameTTLSeconds != 30 || got.MaxFrameBytes != 4<<20 || got.PeriodicQueueCapacity != 1 {
-		t.Fatalf("FrameProcessing delivery defaults = %+v", got)
-	}
-}
-
-func TestOverrideWithEnvironmentVariablesInheritsGlobalFrameProcessing(t *testing.T) {
-	t.Setenv("GLOBAL_AGENT_FRAME_PROCESSING_ENABLED", "true")
-	t.Setenv("GLOBAL_AGENT_FRAME_PROCESSING_ENDPOINT", "https://processor.example/v1/frames")
-	t.Setenv("GLOBAL_AGENT_FRAME_PROCESSING_PROFILE", "never-trigger")
-
-	configuration := &models.Configuration{}
-	OverrideWithEnvironmentVariables(configuration)
-
-	got := configuration.Config.FrameProcessing
-	if got == nil || got.Enabled != "true" || got.Endpoint != "https://processor.example/v1/frames" {
-		t.Fatalf("effective FrameProcessing = %+v", got)
-	}
-	if configuration.CustomConfig.FrameProcessing == nil || configuration.CustomConfig.FrameProcessing.Enabled != "" {
-		t.Fatalf("custom FrameProcessing unexpectedly overrides global config: %+v", configuration.CustomConfig.FrameProcessing)
-	}
-}
-
 func TestNewFactoryConfigReadContextUsesDatabaseTimeout(t *testing.T) {
 	ctx, cancel := newFactoryConfigReadContext()
 	defer cancel()

machinery/src/models/agent_run.go

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

machinery/src/models/agent_run_test.go

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

machinery/src/models/communication.go

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

machinery/src/models/config.go

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

machinery/src/models/frame_processing.go

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

machinery/src/models/frame_processing_test.go

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

machinery/src/packets/queue.go

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

machinery/src/packets/queue_latest_test.go

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

machinery/src/routers/http/health.go

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

machinery/src/routers/http/health_test.go

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

machinery/src/routers/mqtt/main.go

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

machinery/src/routers/mqtt/main_test.go

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