feat: add frame processor reference service

feature/add-encrypted-metadata-field

.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

@@ -341,9 +341,10 @@ See [RTSPS and TLS certificates](README-RTSPS-TLS.md) for the complete Bosch UI,
 | `AGENT_CAPTURE_PIXEL_CHANGE`                | If `CONTINUOUS` set to `false`, the number of pixel require to change before motion triggers.   | "150"                          |
 | `AGENT_CAPTURE_FRAGMENTED`                  | Set the format of the recorded MP4 to fragmented (suitable for HLS).                            | "false"                        |
 | `AGENT_CAPTURE_FRAGMENTED_DURATION`         | If `AGENT_CAPTURE_FRAGMENTED` set to `true`, define the duration (seconds) of a fragment.       | "8"                            |
-| `AGENT_MQTT_URI`                            | An MQTT broker endpoint that is used for bi-directional communication (live view, onvif, etc)   | "tcp://mqtt.kerberos.io:1883"  |
+| `AGENT_MQTT_URI`                            | MQTT broker endpoint for bi-directional communication. Accepts ActiveMQ `mqtt+ssl://` URLs.     | "tcp://mqtt.kerberos.io:1883"  |
 | `AGENT_MQTT_USERNAME`                       | Username of the MQTT broker.                                                                    | ""                             |
 | `AGENT_MQTT_PASSWORD`                       | Password of the MQTT broker.                                                                    | ""                             |
+| `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_STUN_URI`                            | When using WebRTC, you'll need to provide a STUN server.                                        | "stun:turn-fra1.kerberos.io:3478"|
@@ -379,6 +380,13 @@ See [RTSPS and TLS certificates](README-RTSPS-TLS.md) for the complete Bosch UI,
 | `AGENT_SIGNING`                             | Enable 'true' or disable 'false' for signing recordings.                                        | "true"                         |
 | `AGENT_SIGNING_PRIVATE_KEY`                 | The private key (RSA) to sign the recordings fingerprint to validate origin.                    | "" - uses default one if empty |
 
+Remote console access is disabled unless `AGENT_REMOTE_ACCESS_ENABLED=true`.
+The Agent also rejects remote session messages unless Hub encryption or
+end-to-end MQTT encryption is configured and used. A remote shell runs inside
+the Agent process environment as the Agent operating-system user; it is not an
+SSH server and does not expose a new network port. Keep the feature disabled on
+deployments where Hub owners should not have operating-system access.
+
 ### Resumable upload chunk size
 
 Hub and Vault resumable uploads use `AGENT_TUS_CHUNK_SIZE_BYTES` as the maximum
@@ -516,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/go.mod

@@ -11,6 +11,7 @@ require (
 	github.com/bluenviron/gortsplib/v5 v5.6.3
 	github.com/bluenviron/mediacommon v1.14.0
 	github.com/cedricve/go-onvif v0.0.0-20200222191200-567e8ce298f6
+	github.com/creack/pty v1.1.24
 	github.com/dromara/carbon/v2 v2.6.8
 	github.com/dropbox/dropbox-sdk-go-unofficial/v6 v6.0.5
 	github.com/eclipse/paho.mqtt.golang v1.5.0

machinery/go.sum

@@ -456,6 +456,8 @@ github.com/cncf/xds/go v0.0.0-20240905190251-b4127c9b8d78/go.mod h1:W+zGtBO5Y1Ig
 github.com/cncf/xds/go v0.0.0-20241223141626-cff3c89139a3/go.mod h1:W+zGtBO5Y1IgJhy4+A9GOqVhqLpfZi+vwmdNXUehLA8=
 github.com/cncf/xds/go v0.0.0-20250121191232-2f005788dc42/go.mod h1:W+zGtBO5Y1IgJhy4+A9GOqVhqLpfZi+vwmdNXUehLA8=
 github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
+github.com/creack/pty v1.1.24 h1:bJrF4RRfyJnbTJqzRLHzcGaZK1NeM5kTC9jGgovnR1s=
+github.com/creack/pty v1.1.24/go.mod h1:08sCNb52WyoAwi2QDyzUCTgcvVFhUzewun7wtTfvcwE=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=

machinery/src/capture/main.go

@@ -53,12 +53,13 @@ func publishRecordingState(mqttClient mqtt.Client, hubKey string, configuration
 	}
 }
 
-func recordingUploadMetadata(name, deviceKey string, timestamp int64, mp4Video *video.MP4) models.RecordingUploadMetadata {
+func recordingUploadMetadata(name, deviceKey string, timestamp int64, mp4Video *video.MP4, encrypted bool) models.RecordingUploadMetadata {
 	metadata := models.RecordingUploadMetadata{
 		FileName:  filepath.Base(name),
 		DeviceKey: deviceKey,
 		Timestamp: timestamp,
 		Duration:  mp4Video.VideoTotalDuration,
+		Encrypted: encrypted,
 	}
 	value := mp4Video.AverageFPS()
 	if value > 0 && value <= 240 && !math.IsInf(value, 0) && !math.IsNaN(value) {
@@ -481,6 +482,7 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
 						log.Info("capture.main.HandleRecordStream(continuous): no video data recorded, not renaming file.")
 					}
 
+					encrypted := false
 					// Check if we need to encrypt the recording.
 					if config.Encryption != nil && config.Encryption.Enabled == "true" && config.Encryption.Recordings == "true" && config.Encryption.SymmetricKey != "" {
 						// reopen file into memory 'fullName'
@@ -493,6 +495,8 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
 								err := os.WriteFile(fullName, []byte(encryptedContents), 0644)
 								if err != nil {
 									log.Error("capture.main.HandleRecordStream(continuous): error writing file: " + err.Error())
+								} else {
+									encrypted = true
 								}
 							} else {
 								log.Error("capture.main.HandleRecordStream(continuous): error encrypting file: " + err.Error())
@@ -502,7 +506,7 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
 						}
 					}
 
-					queueRecordingForUpload(configDirectory, recordingUploadMetadata(name, config.Key, startRecording, mp4Video))
+					queueRecordingForUpload(configDirectory, recordingUploadMetadata(name, config.Key, startRecording, mp4Video, encrypted))
 
 					recordingStatus = "idle"
 
@@ -638,6 +642,7 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
 						log.Info("capture.main.HandleRecordStream(continuous): no video data recorded, not renaming file.")
 					}
 
+					encrypted := false
 					// Check if we need to encrypt the recording.
 					if config.Encryption != nil && config.Encryption.Enabled == "true" && config.Encryption.Recordings == "true" && config.Encryption.SymmetricKey != "" {
 						// reopen file into memory 'fullName'
@@ -650,6 +655,8 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
 								err := os.WriteFile(fullName, []byte(encryptedContents), 0644)
 								if err != nil {
 									log.Error("capture.main.HandleRecordStream(motiondetection): error writing file: " + err.Error())
+								} else {
+									encrypted = true
 								}
 							} else {
 								log.Error("capture.main.HandleRecordStream(motiondetection): error encrypting file: " + err.Error())
@@ -659,7 +666,7 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
 						}
 					}
 
-					queueRecordingForUpload(configDirectory, recordingUploadMetadata(name, config.Key, startRecording, mp4Video))
+					queueRecordingForUpload(configDirectory, recordingUploadMetadata(name, config.Key, startRecording, mp4Video, encrypted))
 
 					recordingStatus = "idle"
 
@@ -905,6 +912,7 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
 					log.Info("capture.main.HandleRecordStream(motiondetection): no video data recorded, not renaming file.")
 				}
 
+				encrypted := false
 				// Check if we need to encrypt the recording.
 				if config.Encryption != nil && config.Encryption.Enabled == "true" && config.Encryption.Recordings == "true" && config.Encryption.SymmetricKey != "" {
 					// reopen file into memory 'fullName'
@@ -917,6 +925,8 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
 							err := os.WriteFile(fullName, []byte(encryptedContents), 0644)
 							if err != nil {
 								log.Error("capture.main.HandleRecordStream(motiondetection): error writing file: " + err.Error())
+							} else {
+								encrypted = true
 							}
 						} else {
 							log.Error("capture.main.HandleRecordStream(motiondetection): error encrypting file: " + err.Error())
@@ -926,7 +936,7 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
 					}
 				}
 
-				queueRecordingForUpload(configDirectory, recordingUploadMetadata(name, config.Key, displayTime, mp4Video))
+				queueRecordingForUpload(configDirectory, recordingUploadMetadata(name, config.Key, displayTime, mp4Video, encrypted))
 
 				// Clean up the recording directory if necessary.
 				CleanupRecordingDirectory(configDirectory, configuration)

machinery/src/capture/main_test.go

@@ -42,7 +42,7 @@ func TestQueueRecordingForUploadStoresFinalizedMetadata(t *testing.T) {
 	}
 
 	mp4Video := &video.MP4{VideoTotalDuration: 20452, SampleCount: 613}
-	metadata := recordingUploadMetadata("recording.mp4", "device-key", 1785934709414, mp4Video)
+	metadata := recordingUploadMetadata("recording.mp4", "device-key", 1785934709414, mp4Video, true)
 	queueRecordingForUpload(configDirectory, metadata)
 
 	got, err := os.ReadFile(filepath.Join(configDirectory, "data", "cloud", "recording.metadata"))
@@ -54,7 +54,7 @@ func TestQueueRecordingForUploadStoresFinalizedMetadata(t *testing.T) {
 		t.Fatalf("decode upload marker: %v", err)
 	}
 	expectedFPS := mp4Video.AverageFPS()
-	if stored.FileName != "recording.mp4" || stored.DeviceKey != "device-key" || stored.Timestamp != 1785934709414 || stored.Duration != 20452 || math.Abs(stored.FPS-expectedFPS) > 1e-9 {
+	if stored.FileName != "recording.mp4" || stored.DeviceKey != "device-key" || stored.Timestamp != 1785934709414 || stored.Duration != 20452 || math.Abs(stored.FPS-expectedFPS) > 1e-9 || !stored.Encrypted {
 		t.Fatalf("upload marker = %+v", stored)
 	}
 	if stored.FPS == math.Floor(stored.FPS) {

machinery/src/cloud/recording_metadata.go

@@ -15,6 +15,7 @@ import (
 const recordingFPSHeader = "X-Kerberos-Storage-Fps"
 const recordingDurationHeader = "X-Kerberos-Storage-Duration"
 const recordingTimestampHeader = "X-Kerberos-Storage-Timestamp"
+const recordingEncryptedHeader = "X-Kerberos-Storage-Encrypted"
 
 // queuedRecordingFPS reads the FPS snapshot written into the upload marker
 // when the recording was finalized. Historical empty markers intentionally
@@ -84,5 +85,8 @@ func setQueuedRecordingMetadataHeaders(header http.Header, fileName string) {
 		if metadata.Timestamp > 0 {
 			header.Set(recordingTimestampHeader, strconv.FormatInt(metadata.Timestamp, 10))
 		}
+		if metadata.Encrypted {
+			header.Set(recordingEncryptedHeader, "true")
+		}
 	}
 }

machinery/src/cloud/tus_client.go

@@ -361,6 +361,9 @@ func addRecordingTusMetadata(values map[string]string, fileName string) {
 	if metadata.Timestamp > 0 {
 		values["timestamp"] = strconv.FormatInt(metadata.Timestamp, 10)
 	}
+	if metadata.Encrypted {
+		values["encrypted"] = "true"
+	}
 }
 
 // tusCreate performs the tus "creation" request (POST). On success it returns

machinery/src/cloud/tus_client_test.go

@@ -386,7 +386,7 @@ func TestQueuedRecordingFPSAllowsMissingHistoricalMarker(t *testing.T) {
 func TestQueuedRecordingMetadataHeaders(t *testing.T) {
 	fileName := "recording.mp4"
 	withRecording(t, fileName, []byte("recording"))
-	withQueuedRecordingFPS(t, fileName, `{"filename":"recording.mp4","device_key":"device-key","timestamp":1785934709414,"duration":20452,"fps":25}`)
+	withQueuedRecordingFPS(t, fileName, `{"filename":"recording.mp4","device_key":"device-key","timestamp":1785934709414,"duration":20452,"fps":25,"encrypted":true}`)
 
 	header := make(http.Header)
 	setQueuedRecordingMetadataHeaders(header, fileName)
@@ -399,6 +399,14 @@ func TestQueuedRecordingMetadataHeaders(t *testing.T) {
 	if got := header.Get(recordingTimestampHeader); got != "1785934709414" {
 		t.Fatalf("timestamp header = %q", got)
 	}
+	if got := header.Get(recordingEncryptedHeader); got != "true" {
+		t.Fatalf("encrypted header = %q", got)
+	}
+	metadata := map[string]string{}
+	addRecordingTusMetadata(metadata, fileName)
+	if got := metadata["encrypted"]; got != "true" {
+		t.Fatalf("encrypted TUS metadata = %q", got)
+	}
 }
 
 func TestQueuedRecordingFPSAllowsLegacyMarkerFileName(t *testing.T) {

machinery/src/models/mqtt.go

@@ -324,3 +324,23 @@ type TriggerRelay struct {
 	DeviceId  string `json:"device_id"` // device id
 	Token     string `json:"token"`     // token
 }
+
+// RemoteSessionPayload controls an interactive shell or log stream over MQTT.
+// Data is base64 encoded so terminal control bytes remain valid JSON.
+type RemoteSessionPayload struct {
+	Timestamp int64  `json:"timestamp"`
+	SessionID string `json:"session_id"`
+	Kind      string `json:"kind,omitempty"`
+	Data      string `json:"data,omitempty"`
+	Rows      uint16 `json:"rows,omitempty"`
+	Columns   uint16 `json:"columns,omitempty"`
+	Tail      int    `json:"tail,omitempty"`
+}
+
+type RemoteSessionStatus struct {
+	Timestamp int64  `json:"timestamp"`
+	SessionID string `json:"session_id"`
+	Kind      string `json:"kind"`
+	State     string `json:"state"`
+	Error     string `json:"error,omitempty"`
+}

machinery/src/models/recording_upload_metadata.go

@@ -16,6 +16,7 @@ type RecordingUploadMetadata struct {
 	Timestamp int64   `json:"timestamp"` // Unix milliseconds.
 	Duration  uint64  `json:"duration"`  // Milliseconds.
 	FPS       float64 `json:"fps,omitempty"`
+	Encrypted bool    `json:"encrypted,omitempty"`
 }
 
 // RecordingUploadMetadataFileName returns the queue marker name associated

machinery/src/routers/mqtt/main.go

@@ -1,7 +1,9 @@
 package mqtt
 
 import (
+	"context"
 	"crypto/rsa"
+	"crypto/tls"
 	"crypto/x509"
 	"encoding/base64"
 	"encoding/json"
@@ -9,13 +11,14 @@ import (
 	"fmt"
 	"io/ioutil"
 	"math/rand"
+	"net"
+	"net/url"
+	"os"
 	"strconv"
 	"strings"
 	"sync"
 	"time"
 
-	"context"
-
 	mqtt "github.com/eclipse/paho.mqtt.golang"
 	"github.com/kerberos-io/agent/machinery/src/capture"
 	configService "github.com/kerberos-io/agent/machinery/src/config"
@@ -24,6 +27,7 @@ import (
 	"github.com/kerberos-io/agent/machinery/src/onvif"
 	"github.com/kerberos-io/agent/machinery/src/webrtc"
 	log "github.com/sirupsen/logrus"
+	"golang.org/x/net/proxy"
 )
 
 // We'll cache the MQTT settings to know if we need to reinitialize the MQTT client connection.
@@ -34,6 +38,177 @@ var PREV_MQTTPassword string
 var PREV_HubKey string
 var PREV_AgentKey string
 
+type pahoErrorLogger struct{}
+
+func (pahoErrorLogger) Println(values ...interface{}) {
+	log.WithFields(log.Fields{
+		"component": "routers/mqtt",
+		"event":     "paho_error",
+	}).Error(strings.TrimSpace(fmt.Sprintln(values...)))
+}
+
+func (pahoErrorLogger) Printf(format string, values ...interface{}) {
+	log.WithFields(log.Fields{
+		"component": "routers/mqtt",
+		"event":     "paho_error",
+	}).Errorf(strings.TrimSpace(format), values...)
+}
+
+func init() {
+	mqtt.ERROR = pahoErrorLogger{}
+}
+
+func enableMQTTConnectionDiagnostics(options *mqtt.ClientOptions, brokerURL string) {
+	if !strings.Contains(brokerURL, "://") {
+		options.SetCustomOpenConnectionFn(openMQTTConnection)
+		return
+	}
+
+	parsedURL, err := url.Parse(brokerURL)
+	if err != nil {
+		return
+	}
+
+	switch strings.ToLower(parsedURL.Scheme) {
+	case "", "mqtt", "tcp", "ssl", "tls", "mqtts", "mqtt+ssl", "tcps":
+		options.SetCustomOpenConnectionFn(openMQTTConnection)
+	}
+}
+
+func openMQTTConnection(uri *url.URL, options mqtt.ClientOptions) (net.Conn, error) {
+	host := uri.Hostname()
+	fields := log.Fields{
+		"component": "routers/mqtt",
+		"host":      host,
+		"port":      uri.Port(),
+		"scheme":    uri.Scheme,
+	}
+	logMQTTDNSResolution(host, options.ConnectTimeout, fields)
+
+	connectionStartedAt := time.Now()
+	dialer := options.Dialer
+	if dialer == nil {
+		dialer = &net.Dialer{Timeout: options.ConnectTimeout}
+	}
+
+	proxyConfigured := os.Getenv("all_proxy") != ""
+	proxyMode := "direct"
+	if proxyConfigured {
+		proxyMode = "socks"
+	}
+	fields["proxy_mode"] = proxyMode
+	log.WithFields(fields).Info("Opening MQTT TCP connection")
+
+	var (
+		connection net.Conn
+		err        error
+	)
+	if proxyConfigured {
+		connection, err = proxy.FromEnvironment().Dial("tcp", uri.Host)
+	} else {
+		connection, err = dialer.Dial("tcp", uri.Host)
+	}
+	fields["duration_ms"] = time.Since(connectionStartedAt).Milliseconds()
+	if err != nil {
+		logMQTTNetworkError("MQTT TCP connection failed", err, fields)
+		return nil, err
+	}
+
+	fields["local_address"] = connection.LocalAddr().String()
+	fields["remote_address"] = connection.RemoteAddr().String()
+	log.WithFields(fields).Info("MQTT TCP connection established")
+
+	if !isSecureMQTTScheme(uri.Scheme) {
+		return connection, nil
+	}
+
+	tlsConfig := options.TLSConfig
+	if tlsConfig == nil {
+		tlsConfig = &tls.Config{}
+	} else {
+		tlsConfig = tlsConfig.Clone()
+	}
+	if tlsConfig.ServerName == "" {
+		tlsConfig.ServerName = host
+	}
+
+	tlsConnection := tls.Client(connection, tlsConfig)
+	tlsStartedAt := time.Now()
+	if options.ConnectTimeout > 0 {
+		_ = tlsConnection.SetDeadline(connectionStartedAt.Add(options.ConnectTimeout))
+	}
+	if err = tlsConnection.Handshake(); err != nil {
+		_ = connection.Close()
+		fields["duration_ms"] = time.Since(tlsStartedAt).Milliseconds()
+		fields["server_name"] = tlsConfig.ServerName
+		logMQTTNetworkError("MQTT TLS handshake failed", err, fields)
+		return nil, err
+	}
+	_ = tlsConnection.SetDeadline(time.Time{})
+
+	state := tlsConnection.ConnectionState()
+	fields["cipher_suite"] = tls.CipherSuiteName(state.CipherSuite)
+	fields["duration_ms"] = time.Since(tlsStartedAt).Milliseconds()
+	fields["server_name"] = tlsConfig.ServerName
+	fields["tls_version"] = tls.VersionName(state.Version)
+	log.WithFields(fields).Info("MQTT TLS handshake established")
+	return tlsConnection, nil
+}
+
+func logMQTTDNSResolution(host string, timeout time.Duration, fields log.Fields) {
+	if host == "" || net.ParseIP(host) != nil {
+		return
+	}
+
+	lookupTimeout := timeout
+	if lookupTimeout <= 0 || lookupTimeout > 5*time.Second {
+		lookupTimeout = 5 * time.Second
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), lookupTimeout)
+	defer cancel()
+
+	startedAt := time.Now()
+	addresses, err := net.DefaultResolver.LookupHost(ctx, host)
+	dnsFields := cloneLogFields(fields)
+	dnsFields["duration_ms"] = time.Since(startedAt).Milliseconds()
+	if err != nil {
+		logMQTTNetworkError("MQTT broker DNS resolution failed", err, dnsFields)
+		return
+	}
+
+	dnsFields["resolved_addresses"] = addresses
+	log.WithFields(dnsFields).Info("MQTT broker DNS resolved")
+}
+
+func logMQTTNetworkError(message string, err error, fields log.Fields) {
+	errorFields := cloneLogFields(fields)
+	if networkError, ok := err.(net.Error); ok {
+		errorFields["network_timeout"] = networkError.Timeout()
+	}
+	if operationError, ok := err.(*net.OpError); ok {
+		errorFields["network"] = operationError.Net
+		errorFields["operation"] = operationError.Op
+	}
+	log.WithError(err).WithFields(errorFields).Error(message)
+}
+
+func cloneLogFields(fields log.Fields) log.Fields {
+	cloned := make(log.Fields, len(fields))
+	for key, value := range fields {
+		cloned[key] = value
+	}
+	return cloned
+}
+
+func isSecureMQTTScheme(scheme string) bool {
+	switch strings.ToLower(scheme) {
+	case "ssl", "tls", "mqtts", "mqtt+ssl", "tcps":
+		return true
+	default:
+		return false
+	}
+}
+
 func HasMQTTClientModified(configuration *models.Configuration) bool {
 	MTTURI := configuration.Config.MQTTURI
 	MTTUsername := configuration.Config.MQTTUsername
@@ -59,6 +234,7 @@ func HasMQTTClientModified(configuration *models.Configuration) bool {
 // - kerberos/{hubkey}/device/{devicekey}/motion: a motion signal
 
 func ConfigureMQTT(configDirectory string, configuration *models.Configuration, communication *models.Communication) mqtt.Client {
+	installRemoteAccessHook()
 
 	config := configuration.Config
 
@@ -116,6 +292,7 @@ func ConfigureMQTT(configDirectory string, configuration *models.Configuration,
 		// Some extra options to make sure the connection behaves
 		// properly. More information here: github.com/eclipse/paho.mqtt.golang.
 		//opts.SetCleanSession(true)
+		enableMQTTConnectionDiagnostics(opts, mqttURL)
 		opts.SetCleanSession(false)
 		opts.SetResumeSubs(true)
 		opts.SetStore(mqtt.NewMemoryStore())
@@ -203,7 +380,7 @@ func ConfigureMQTT(configDirectory string, configuration *models.Configuration,
 				"component":  "routers/mqtt",
 				"event":      "initial_connection_timeout",
 				"timeout_ms": (30 * time.Second).Milliseconds(),
-			}).Error("Timed out establishing initial MQTT connection")
+			}).Warn("Initial MQTT connection is still retrying")
 		}
 		return mqc
 	}
@@ -273,6 +450,7 @@ func MQTTListenerHandler(mqttClient mqtt.Client, hubKey string, configDirectory
 			// We will receive all messages from our hub, so we'll need to filter to the relevant device.
 			if message.Mid != "" && message.Timestamp != 0 && message.DeviceId == configuration.Config.Key {
 				var payload models.Payload
+				remoteAuthenticated := false
 
 				// Messages might be hidden, if so we'll need to decrypt them using the Kerberos Hub private key.
 				if message.Hidden && configuration.Config.HubEncryption == "true" {
@@ -289,8 +467,10 @@ func MQTTListenerHandler(mqttClient mqtt.Client, hubKey string, configDirectory
 								log.Error("routers.mqtt.main.MQTTListenerHandler(): error decrypting message: " + err.Error())
 								return
 							}
-							json.Unmarshal(visibleValue, &payload)
-							message.Payload = payload
+							if err := json.Unmarshal(visibleValue, &payload); err == nil {
+								message.Payload = payload
+								remoteAuthenticated = true
+							}
 						} else {
 							log.Error("routers.mqtt.main.MQTTListenerHandler(): error decrypting message, no private key provided.")
 						}
@@ -338,7 +518,9 @@ func MQTTListenerHandler(mqttClient mqtt.Client, hubKey string, configDirectory
 											log.Error("routers.mqtt.main.MQTTListenerHandler(): error decrypting message: " + err.Error())
 											return
 										}
-										json.Unmarshal(decryptedValue, &payload)
+										if err := json.Unmarshal(decryptedValue, &payload); err == nil {
+											remoteAuthenticated = true
+										}
 									} else {
 										log.Error("routers.mqtt.main.MQTTListenerHandler(): error decrypting message, assymetric keys do not match.")
 										return
@@ -396,6 +578,14 @@ 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 "remote-session-open":
+					go HandleRemoteSessionOpen(mqttClient, hubKey, payload, remoteAuthenticated, configuration)
+				case "remote-session-input":
+					go HandleRemoteSessionInput(mqttClient, hubKey, payload, remoteAuthenticated, configuration)
+				case "remote-session-resize":
+					go HandleRemoteSessionResize(payload, remoteAuthenticated)
+				case "remote-session-close":
+					go HandleRemoteSessionClose(payload, remoteAuthenticated)
 				}
 
 			}

machinery/src/routers/mqtt/main_test.go

@@ -1,12 +1,95 @@
 package mqtt
 
 import (
+	"crypto/tls"
+	"errors"
+	"net/http/httptest"
+	"net/url"
+	"os"
 	"testing"
 	"time"
 
+	mqtt "github.com/eclipse/paho.mqtt.golang"
 	"github.com/kerberos-io/agent/machinery/src/models"
 )
 
+func TestEnableMQTTConnectionDiagnostics(t *testing.T) {
+	tests := []struct {
+		name      string
+		brokerURL string
+		want      bool
+	}{
+		{name: "ActiveMQ TLS", brokerURL: "mqtt+ssl://broker.example:8883", want: true},
+		{name: "TCP", brokerURL: "tcp://broker.example:1883", want: true},
+		{name: "default TCP", brokerURL: "broker.example:1883", want: true},
+		{name: "WebSocket", brokerURL: "wss://broker.example/mqtt", want: false},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			options := mqtt.NewClientOptions()
+			enableMQTTConnectionDiagnostics(options, test.brokerURL)
+			if got := options.CustomOpenConnectionFn != nil; got != test.want {
+				t.Fatalf("CustomOpenConnectionFn configured = %t, want %t", got, test.want)
+			}
+		})
+	}
+}
+
+func TestOpenMQTTConnectionReturnsTCPError(t *testing.T) {
+	options := *mqtt.NewClientOptions().SetConnectTimeout(100 * time.Millisecond)
+	brokerURL, err := url.Parse("tcp://127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	connection, err := openMQTTConnection(brokerURL, options)
+	if connection != nil {
+		connection.Close()
+		t.Fatal("openMQTTConnection() returned a connection for an unavailable endpoint")
+	}
+	if err == nil {
+		t.Fatal("openMQTTConnection() returned no TCP error")
+	}
+}
+
+func TestOpenMQTTConnectionEstablishesActiveMQTLS(t *testing.T) {
+	server := httptest.NewTLSServer(nil)
+	defer server.Close()
+
+	serverURL, err := url.Parse(server.URL)
+	if err != nil {
+		t.Fatal(err)
+	}
+	brokerURL, err := url.Parse("mqtt+ssl://" + serverURL.Host)
+	if err != nil {
+		t.Fatal(err)
+	}
+	options := *mqtt.NewClientOptions().
+		SetConnectTimeout(time.Second).
+		SetTLSConfig(&tls.Config{InsecureSkipVerify: true}) // #nosec G402 -- local test server
+
+	connection, err := openMQTTConnection(brokerURL, options)
+	if err != nil {
+		t.Fatalf("openMQTTConnection() error = %v", err)
+	}
+	defer connection.Close()
+	if _, ok := connection.(*tls.Conn); !ok {
+		t.Fatalf("openMQTTConnection() connection type = %T, want *tls.Conn", connection)
+	}
+}
+
+func TestIsSecureMQTTScheme(t *testing.T) {
+	for _, scheme := range []string{"ssl", "tls", "mqtts", "mqtt+ssl", "tcps"} {
+		if !isSecureMQTTScheme(scheme) {
+			t.Errorf("isSecureMQTTScheme(%q) = false, want true", scheme)
+		}
+	}
+	if isSecureMQTTScheme("tcp") {
+		t.Error("isSecureMQTTScheme(\"tcp\") = true, want false")
+	}
+}
+
 func TestConfigureMQTTRequiresHubKey(t *testing.T) {
 	configuration := &models.Configuration{Config: models.Config{Key: "agent-key"}}
 
@@ -53,3 +136,73 @@ func TestEnqueueLatestAudioDoesNotBlockNilChannel(t *testing.T) {
 		t.Fatal("enqueueLatestAudio() blocked on a nil channel")
 	}
 }
+
+func TestRemoteAccessRequiresExplicitOptIn(t *testing.T) {
+	previous, present := os.LookupEnv(remoteAccessEnvironment)
+	t.Cleanup(func() {
+		if present {
+			_ = os.Setenv(remoteAccessEnvironment, previous)
+		} else {
+			_ = os.Unsetenv(remoteAccessEnvironment)
+		}
+	})
+
+	_ = os.Unsetenv(remoteAccessEnvironment)
+	if remoteAccessEnabled() {
+		t.Fatal("remoteAccessEnabled() = true without opt-in")
+	}
+	_ = os.Setenv(remoteAccessEnvironment, "true")
+	if !remoteAccessEnabled() {
+		t.Fatal("remoteAccessEnabled() = false after opt-in")
+	}
+}
+
+func TestNormalizeTerminalSize(t *testing.T) {
+	rows, columns := normalizeTerminalSize(0, 0)
+	if rows != 24 || columns != 80 {
+		t.Fatalf("normalizeTerminalSize(0, 0) = (%d, %d), want (24, 80)", rows, columns)
+	}
+
+	rows, columns = normalizeTerminalSize(500, 500)
+	if rows != 200 || columns != 400 {
+		t.Fatalf("normalizeTerminalSize(500, 500) = (%d, %d), want (200, 400)", rows, columns)
+	}
+}
+
+func TestDecodeRemotePayloadRejectsMissingSession(t *testing.T) {
+	_, err := decodeRemotePayload(models.Payload{Value: map[string]interface{}{
+		"kind": "shell",
+	}})
+	if err == nil {
+		t.Fatal("decodeRemotePayload() accepted a missing session id")
+	}
+}
+
+func TestRemoteSessionOpenRejectsUnprovenEncryption(t *testing.T) {
+	previous, present := os.LookupEnv(remoteAccessEnvironment)
+	t.Cleanup(func() {
+		if present {
+			_ = os.Setenv(remoteAccessEnvironment, previous)
+		} else {
+			_ = os.Unsetenv(remoteAccessEnvironment)
+		}
+	})
+	_ = os.Setenv(remoteAccessEnvironment, "true")
+
+	// The listener passes false when an envelope merely claims to be hidden but
+	// no ciphertext was successfully decrypted. The remote handler must reject it.
+	if err := validateRemoteAccess(false); !errors.Is(err, errRemoteUnauthenticated) {
+		t.Fatalf("validateRemoteAccess(false) error = %v, want %v", err, errRemoteUnauthenticated)
+	}
+}
+
+func TestRemoteSessionReservationIsIdempotent(t *testing.T) {
+	manager := newRemoteAccessManager()
+	session := &remoteSession{id: "session-1", kind: "logs"}
+	if _, err := manager.reserve(session); err != nil {
+		t.Fatalf("first reserve() failed: %v", err)
+	}
+	if _, err := manager.reserve(session); !errors.Is(err, errRemoteSessionExists) {
+		t.Fatalf("duplicate reserve() error = %v, want %v", err, errRemoteSessionExists)
+	}
+}

machinery/src/routers/mqtt/remote_access.go

@@ -0,0 +1,397 @@
+package mqtt
+
+import (
+	"context"
+	"encoding/base64"
+	"encoding/json"
+	"errors"
+	"io"
+	"os"
+	"os/exec"
+	"strconv"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/creack/pty"
+	paho "github.com/eclipse/paho.mqtt.golang"
+	"github.com/kerberos-io/agent/machinery/src/models"
+	log "github.com/sirupsen/logrus"
+)
+
+const (
+	remoteAccessEnvironment = "AGENT_REMOTE_ACCESS_ENABLED"
+	remoteHistoryLimit      = 500
+	remoteSessionLimit      = 5
+	remoteOutputChunkSize   = 4096
+	remoteInputLimit        = 64 * 1024
+	remoteSessionLifetime   = time.Hour
+)
+
+type remoteSession struct {
+	id     string
+	kind   string
+	pty    *os.File
+	cancel context.CancelFunc
+	logs   chan string
+	done   chan struct{}
+	timer  *time.Timer
+}
+
+type remoteAccessManager struct {
+	mu       sync.Mutex
+	sessions map[string]*remoteSession
+	history  []string
+}
+
+var (
+	remoteAccess             = newRemoteAccessManager()
+	remoteHookOnce           sync.Once
+	errRemoteDisabled        = errors.New("remote access is disabled on this agent")
+	errRemoteUnauthenticated = errors.New("remote access requires encrypted MQTT")
+	errRemoteSessionExists   = errors.New("remote session already exists")
+	errRemoteSessionLimit    = errors.New("remote session limit reached")
+)
+
+func newRemoteAccessManager() *remoteAccessManager {
+	return &remoteAccessManager{sessions: make(map[string]*remoteSession)}
+}
+
+func installRemoteAccessHook() {
+	remoteHookOnce.Do(func() {
+		log.AddHook(remoteAccess)
+	})
+}
+
+func (manager *remoteAccessManager) Levels() []log.Level {
+	return log.AllLevels
+}
+
+func (manager *remoteAccessManager) Fire(entry *log.Entry) error {
+	line, err := json.Marshal(map[string]interface{}{
+		"timestamp": entry.Time.Format(time.RFC3339Nano),
+		"level":     entry.Level.String(),
+		"message":   entry.Message,
+		"fields":    entry.Data,
+	})
+	if err != nil {
+		return nil
+	}
+	encoded := base64.StdEncoding.EncodeToString(append(line, '\n'))
+
+	manager.mu.Lock()
+	manager.history = append(manager.history, encoded)
+	if len(manager.history) > remoteHistoryLimit {
+		manager.history = manager.history[len(manager.history)-remoteHistoryLimit:]
+	}
+	for _, session := range manager.sessions {
+		if session.kind != "logs" {
+			continue
+		}
+		select {
+		case session.logs <- encoded:
+		default:
+		}
+	}
+	manager.mu.Unlock()
+	return nil
+}
+
+func remoteAccessEnabled() bool {
+	enabled, err := strconv.ParseBool(strings.TrimSpace(os.Getenv(remoteAccessEnvironment)))
+	return err == nil && enabled
+}
+
+func validateRemoteAccess(authenticated bool) error {
+	if !authenticated {
+		return errRemoteUnauthenticated
+	}
+	if !remoteAccessEnabled() {
+		return errRemoteDisabled
+	}
+	return nil
+}
+
+func decodeRemotePayload(payload models.Payload) (models.RemoteSessionPayload, error) {
+	data, err := json.Marshal(payload.Value)
+	if err != nil {
+		return models.RemoteSessionPayload{}, err
+	}
+	var request models.RemoteSessionPayload
+	if err := json.Unmarshal(data, &request); err != nil {
+		return models.RemoteSessionPayload{}, err
+	}
+	if request.SessionID == "" || len(request.SessionID) > 128 {
+		return models.RemoteSessionPayload{}, errors.New("invalid remote session id")
+	}
+	return request, nil
+}
+
+func normalizeTerminalSize(rows uint16, columns uint16) (uint16, uint16) {
+	if rows < 5 {
+		rows = 24
+	}
+	if rows > 200 {
+		rows = 200
+	}
+	if columns < 20 {
+		columns = 80
+	}
+	if columns > 400 {
+		columns = 400
+	}
+	return rows, columns
+}
+
+func HandleRemoteSessionOpen(client paho.Client, hubKey string, payload models.Payload, authenticated bool, configuration *models.Configuration) {
+	request, err := decodeRemotePayload(payload)
+	if err != nil {
+		return
+	}
+	if accessErr := validateRemoteAccess(authenticated); accessErr != nil {
+		publishRemoteStatus(client, hubKey, configuration, request.SessionID, request.Kind, "error", accessErr.Error())
+		return
+	}
+
+	switch request.Kind {
+	case "logs":
+		err = remoteAccess.openLogs(client, hubKey, configuration, request)
+	case "shell":
+		err = remoteAccess.openShell(client, hubKey, configuration, request)
+	default:
+		err = errors.New("unsupported remote session kind")
+	}
+	if err != nil {
+		if errors.Is(err, errRemoteSessionExists) {
+			publishRemoteStatus(client, hubKey, configuration, request.SessionID, request.Kind, "opened", "")
+			return
+		}
+		publishRemoteStatus(client, hubKey, configuration, request.SessionID, request.Kind, "error", err.Error())
+	}
+}
+
+func HandleRemoteSessionInput(client paho.Client, hubKey string, payload models.Payload, authenticated bool, configuration *models.Configuration) {
+	if validateRemoteAccess(authenticated) != nil {
+		return
+	}
+	request, err := decodeRemotePayload(payload)
+	if err != nil || len(request.Data) > remoteInputLimit*2 {
+		return
+	}
+	data, err := base64.StdEncoding.DecodeString(request.Data)
+	if err != nil || len(data) > remoteInputLimit {
+		return
+	}
+	remoteAccess.mu.Lock()
+	session := remoteAccess.sessions[request.SessionID]
+	remoteAccess.mu.Unlock()
+	if session == nil || session.kind != "shell" || session.pty == nil {
+		publishRemoteStatus(client, hubKey, configuration, request.SessionID, "shell", "error", "remote session is not open")
+		return
+	}
+	if _, err := session.pty.Write(data); err != nil {
+		publishRemoteStatus(client, hubKey, configuration, request.SessionID, "shell", "error", "failed to write terminal input")
+	}
+}
+
+func HandleRemoteSessionResize(payload models.Payload, authenticated bool) {
+	if validateRemoteAccess(authenticated) != nil {
+		return
+	}
+	request, err := decodeRemotePayload(payload)
+	if err != nil {
+		return
+	}
+	rows, columns := normalizeTerminalSize(request.Rows, request.Columns)
+	remoteAccess.mu.Lock()
+	session := remoteAccess.sessions[request.SessionID]
+	remoteAccess.mu.Unlock()
+	if session != nil && session.kind == "shell" && session.pty != nil {
+		_ = pty.Setsize(session.pty, &pty.Winsize{Rows: rows, Cols: columns})
+	}
+}
+
+func HandleRemoteSessionClose(payload models.Payload, authenticated bool) {
+	if !authenticated {
+		return
+	}
+	request, err := decodeRemotePayload(payload)
+	if err == nil {
+		remoteAccess.close(request.SessionID)
+	}
+}
+
+func (manager *remoteAccessManager) reserve(session *remoteSession) ([]string, error) {
+	manager.mu.Lock()
+	defer manager.mu.Unlock()
+	if _, exists := manager.sessions[session.id]; exists {
+		return nil, errRemoteSessionExists
+	}
+	if len(manager.sessions) >= remoteSessionLimit {
+		return nil, errRemoteSessionLimit
+	}
+	manager.sessions[session.id] = session
+	history := append([]string(nil), manager.history...)
+	return history, nil
+}
+
+func (manager *remoteAccessManager) expire(client paho.Client, hubKey string, configuration *models.Configuration, session *remoteSession) {
+	session.timer = time.AfterFunc(remoteSessionLifetime, func() {
+		manager.close(session.id)
+		if session.kind == "logs" {
+			publishRemoteStatus(client, hubKey, configuration, session.id, session.kind, "closed", "session lifetime reached")
+		}
+	})
+}
+
+func (manager *remoteAccessManager) openLogs(client paho.Client, hubKey string, configuration *models.Configuration, request models.RemoteSessionPayload) error {
+	session := &remoteSession{
+		id:   request.SessionID,
+		kind: "logs",
+		logs: make(chan string, 256),
+		done: make(chan struct{}),
+	}
+	history, err := manager.reserve(session)
+	if err != nil {
+		return err
+	}
+	manager.expire(client, hubKey, configuration, session)
+	tail := request.Tail
+	if tail <= 0 || tail > remoteHistoryLimit {
+		tail = 200
+	}
+	if len(history) > tail {
+		history = history[len(history)-tail:]
+	}
+
+	publishRemoteStatus(client, hubKey, configuration, session.id, session.kind, "opened", "")
+	go func() {
+		for _, line := range history {
+			publishRemoteOutput(client, hubKey, configuration, session.id, session.kind, line)
+		}
+		for {
+			select {
+			case line := <-session.logs:
+				publishRemoteOutput(client, hubKey, configuration, session.id, session.kind, line)
+			case <-session.done:
+				return
+			}
+		}
+	}()
+	return nil
+}
+
+func (manager *remoteAccessManager) openShell(client paho.Client, hubKey string, configuration *models.Configuration, request models.RemoteSessionPayload) error {
+	rows, columns := normalizeTerminalSize(request.Rows, request.Columns)
+	ctx, cancel := context.WithCancel(context.Background())
+	session := &remoteSession{
+		id:     request.SessionID,
+		kind:   "shell",
+		cancel: cancel,
+		done:   make(chan struct{}),
+	}
+	if _, err := manager.reserve(session); err != nil {
+		cancel()
+		return err
+	}
+	manager.expire(client, hubKey, configuration, session)
+	command := exec.CommandContext(ctx, "/bin/sh")
+	command.Env = append(os.Environ(), "TERM=xterm-256color", "HISTFILE=/dev/null")
+	terminal, err := pty.StartWithSize(command, &pty.Winsize{Rows: rows, Cols: columns})
+	if err != nil {
+		manager.remove(session.id)
+		cancel()
+		return err
+	}
+	session.pty = terminal
+
+	publishRemoteStatus(client, hubKey, configuration, session.id, session.kind, "opened", "")
+	go manager.forwardShell(client, hubKey, configuration, session, command)
+	return nil
+}
+
+func (manager *remoteAccessManager) forwardShell(client paho.Client, hubKey string, configuration *models.Configuration, session *remoteSession, command *exec.Cmd) {
+	buffer := make([]byte, remoteOutputChunkSize)
+	for {
+		count, err := session.pty.Read(buffer)
+		if count > 0 {
+			publishRemoteOutput(client, hubKey, configuration, session.id, session.kind, base64.StdEncoding.EncodeToString(buffer[:count]))
+		}
+		if err != nil {
+			if !errors.Is(err, io.EOF) && !errors.Is(err, os.ErrClosed) {
+				publishRemoteStatus(client, hubKey, configuration, session.id, session.kind, "error", "terminal stream closed unexpectedly")
+			}
+			break
+		}
+	}
+	_ = command.Wait()
+	manager.remove(session.id)
+	publishRemoteStatus(client, hubKey, configuration, session.id, session.kind, "closed", "")
+}
+
+func (manager *remoteAccessManager) close(sessionID string) {
+	manager.mu.Lock()
+	session := manager.sessions[sessionID]
+	delete(manager.sessions, sessionID)
+	manager.mu.Unlock()
+	if session == nil {
+		return
+	}
+	if session.cancel != nil {
+		session.cancel()
+	}
+	if session.pty != nil {
+		_ = session.pty.Close()
+	}
+	if session.logs != nil {
+		close(session.done)
+	}
+	if session.timer != nil {
+		session.timer.Stop()
+	}
+}
+
+func (manager *remoteAccessManager) remove(sessionID string) {
+	manager.mu.Lock()
+	session := manager.sessions[sessionID]
+	delete(manager.sessions, sessionID)
+	manager.mu.Unlock()
+	if session != nil && session.timer != nil {
+		session.timer.Stop()
+	}
+}
+
+func publishRemoteStatus(client paho.Client, hubKey string, configuration *models.Configuration, sessionID string, kind string, state string, errorMessage string) {
+	status := models.RemoteSessionStatus{
+		Timestamp: time.Now().Unix(),
+		SessionID: sessionID,
+		Kind:      kind,
+		State:     state,
+		Error:     errorMessage,
+	}
+	value, _ := json.Marshal(status)
+	var statusValue map[string]interface{}
+	_ = json.Unmarshal(value, &statusValue)
+	publishRemote(client, hubKey, configuration, "remote-session-status", statusValue, 1)
+}
+
+func publishRemoteOutput(client paho.Client, hubKey string, configuration *models.Configuration, sessionID string, kind string, data string) {
+	publishRemote(client, hubKey, configuration, "remote-session-output", map[string]interface{}{
+		"timestamp":  time.Now().Unix(),
+		"session_id": sessionID,
+		"kind":       kind,
+		"data":       data,
+	}, 0)
+}
+
+func publishRemote(client paho.Client, hubKey string, configuration *models.Configuration, action string, value map[string]interface{}, qos byte) {
+	message := models.Message{Payload: models.Payload{
+		Action:   action,
+		DeviceId: configuration.Config.Key,
+		Value:    value,
+	}}
+	payload, err := models.PackageMQTTMessage(configuration, message)
+	if err == nil {
+		client.Publish("kerberos/hub/"+hubKey, qos, false, payload)
+	}
+}