feat: add external frame processing

README.md

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

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/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/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/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,6 +1,7 @@
 package mqtt
 
 import (
+	"bytes"
 	"context"
 	"crypto/rsa"
 	"crypto/tls"
@@ -8,6 +9,7 @@ import (
 	"encoding/base64"
 	"encoding/json"
 	"encoding/pem"
+	"errors"
 	"fmt"
 	"io/ioutil"
 	"math/rand"
@@ -578,6 +580,8 @@ 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":
@@ -606,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

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