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

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