feat: add external frame processing

feature/add-encrypted-metadata-field

README.md

@@ -341,11 +341,26 @@ 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_FRAME_PROCESSING_ENABLED`            | Send keyframe-aligned JPEGs to an external Frame Processor over HTTP.                           | "false"                        |
+| `AGENT_FRAME_PROCESSING_ENDPOINT`           | Full Frame Processor HTTP endpoint, including `/v1/frames`.                                     | ""                             |
+| `AGENT_FRAME_PROCESSING_TOKEN`              | Environment-only bearer token used to authenticate frame submissions; never returned by config APIs. | ""                         |
+| `AGENT_FRAME_PROCESSING_PROFILE`            | Processing profile included with each frame.                                                    | "never-trigger"                |
+| `AGENT_FRAME_PROCESSING_ALLOW_REQUESTED_FRAMES` | Allow authenticated MQTT `capture-frame` commands; frame bytes are still submitted over HTTP. | "false"                    |
+| `AGENT_FRAME_PROCESSING_STREAM`             | Source stream: `auto`, `main`, or `sub`; `auto` prefers the substream when available.           | "auto"                         |
+| `AGENT_FRAME_PROCESSING_INTERVAL_SECONDS`   | Target period between submissions; the first keyframe at or after each deadline is selected.   | "10"                           |
+| `AGENT_FRAME_PROCESSING_WIDTH`              | Output JPEG width; aspect ratio is preserved when height is `0`.                               | "640"                          |
+| `AGENT_FRAME_PROCESSING_HEIGHT`             | Output JPEG height; `0` derives it from the source aspect ratio.                               | "0"                            |
+| `AGENT_FRAME_PROCESSING_JPEG_QUALITY`       | JPEG quality from 1 through 100.                                                                | "70"                           |
+| `AGENT_FRAME_PROCESSING_REQUEST_TIMEOUT_SECONDS` | Maximum duration of one HTTP submission.                                                   | "5"                            |
+| `AGENT_FRAME_PROCESSING_FRAME_TTL_SECONDS`  | Time after capture during which the Frame Processor may accept the frame.                       | "30"                           |
+| `AGENT_FRAME_PROCESSING_MAX_FRAME_BYTES`    | Maximum encoded JPEG size; values above 16 MiB are rejected.                                    | "4194304"                      |
+| `AGENT_FRAME_PROCESSING_PERIODIC_QUEUE_CAPACITY` | Bounded latest-wins periodic frame queue capacity, from 1 through 64.                       | "1"                            |
 | `AGENT_STUN_URI`                            | When using WebRTC, you'll need to provide a STUN server.                                        | "stun:turn-fra1.kerberos.io:3478"|
 | `AGENT_FORCE_TURN`                          | Force using a TURN server, by generating relay candidates only.                                 | "false"                        |
 | `AGENT_TURN_URI`                            | When using WebRTC, you'll need to provide a TURN server.                                        | "turn:turn-fra1.kerberos.io:3478"|
@@ -379,6 +394,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

machinery/data/config/config.json

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

machinery/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/gortsplib.go

@@ -1527,13 +1527,13 @@ func newDecoder(codecName string) (*Decoder, error) {
 
 	res := C.avcodec_open2(codecCtx, codec, nil)
 	if res < 0 {
-		C.avcodec_close(codecCtx)
+		C.avcodec_free_context(&codecCtx)
 		return nil, fmt.Errorf("avcodec_open2() failed")
 	}
 
 	srcFrame := C.av_frame_alloc()
 	if srcFrame == nil {
-		C.avcodec_close(codecCtx)
+		C.avcodec_free_context(&codecCtx)
 		return nil, fmt.Errorf("av_frame_alloc() failed")
 	}
 
@@ -1549,7 +1549,7 @@ func (d *Decoder) Close() {
 		C.av_frame_free(&d.srcFrame)
 	}
 	C.av_frame_free(&d.srcFrame)
-	C.avcodec_close(d.codecCtx)
+	C.avcodec_free_context(&d.codecCtx)
 }
 
 func (d *Decoder) decode(nalu []byte) (image.YCbCr, error) {

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

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

machinery/src/cloud/frameprocessing/frameprocessing_test.go

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

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

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

machinery/src/config/main.go

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

machinery/src/config/main_test.go

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

machinery/src/models/agent_run.go

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

machinery/src/models/agent_run_test.go

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

machinery/src/models/communication.go

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

machinery/src/models/config.go

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

machinery/src/models/frame_processing.go

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

machinery/src/models/frame_processing_test.go

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

machinery/src/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/packets/queue.go

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

machinery/src/packets/queue_latest_test.go

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

machinery/src/routers/http/health.go

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

machinery/src/routers/http/health_test.go

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

machinery/src/routers/mqtt/main.go

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

machinery/src/routers/mqtt/main_test.go

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