Merge pull request #295 from kerberos-io/feature/optimise-hls-upload

feature/optimise-hls-upload

README.md

@@ -231,9 +231,9 @@ Next to attaching the configuration file, it is also possible to override the co
 | `AGENT_MQTT_PASSWORD`                       | Password of the MQTT broker.                                                                    | ""                             |
 | `AGENT_REALTIME_PROCESSING`                 | If `AGENT_REALTIME_PROCESSING` set to `true`, the agent will send key frames to the topic       | ""                             |
 | `AGENT_REALTIME_PROCESSING_TOPIC`           | The topic to which keyframes will be sent in base64 encoded format.                             | ""                             |
-| `AGENT_STUN_URI`                            | When using WebRTC, you'll need to provide a STUN server.                                        | "stun:turn.kerberos.io:8443"   |
+| `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.kerberos.io:8443"   |
+| `AGENT_TURN_URI`                            | When using WebRTC, you'll need to provide a TURN server.                                        | "turn:turn-fra1.kerberos.io:3478"|
 | `AGENT_TURN_USERNAME`                       | TURN username used for WebRTC.                                                                  | "username1"                    |
 | `AGENT_TURN_PASSWORD`                       | TURN password used for WebRTC.                                                                  | "password1"                    |
 | `AGENT_CLOUD`                               | Store recordings in Kerberos Hub (s3), Kerberos Vault (kstorage), or Dropbox (dropbox).         | "s3"                           |

machinery/data/config/config.json

@@ -106,9 +106,9 @@
 	"mqtturi": "tcp://mqtt.kerberos.io:1883",
 	"mqtt_username": "",
 	"mqtt_password": "",
-	"stunuri": "stun:turn.kerberos.io:8443",
 	"turn_force": "false",
-	"turnuri": "turn:turn.kerberos.io:8443",
+	"stunuri": "stun:turn-fra1.kerberos.io:3478",
+	"turnuri": "turn:turn-fra1.kerberos.io:3478",
 	"turn_username": "username1",
 	"turn_password": "password1",
 	"heartbeaturi": "",

machinery/src/cloud/cloud.go

@@ -2,6 +2,7 @@ package cloud
 
 import (
 	"bytes"
+	"context"
 	"crypto/tls"
 	"encoding/base64"
 	"encoding/json"
@@ -21,6 +22,7 @@ import (
 	"time"
 
 	"github.com/kerberos-io/agent/machinery/src/capture"
+	"github.com/kerberos-io/agent/machinery/src/cloud/livesnapshot"
 	"github.com/kerberos-io/agent/machinery/src/encryption"
 	"github.com/kerberos-io/agent/machinery/src/log"
 	"github.com/kerberos-io/agent/machinery/src/models"
@@ -528,6 +530,7 @@ loop:
 						"onvif_events_list": %s,
 						"cameraConnected": "%s",
 						"hasBackChannel": "%s",
+						"livePreviewHttp": true,
 						"numberoffiles" : "33",
 						"timestamp" : 1564747908,
 						"cameratype" : "IPCamera",
@@ -684,7 +687,35 @@ func HandleLiveStreamSD(livestreamCursor *packets.QueueCursor, configuration *mo
 				hubKey = config.HubKey
 			}
 
-			lastLivestreamRequest := int64(0)
+			lastLivestreamRequestMQTT := int64(0)
+			lastLivestreamRequestHTTP := int64(0)
+
+			// HTTP transport (preferred when this agent is paired with a Kerberos
+			// Hub): ship preview frames to hub-api over HTTPS instead of pushing
+			// (large, base64) images through the MQTT broker. Viewers opt in per
+			// session via the "http" transport on their keepalive; the legacy MQTT
+			// push is kept for viewers (older frontends) that don't, and as a fallback.
+			region := ""
+			if config.S3 != nil {
+				region = config.S3.Region
+			}
+			var snapshotPublisher *livesnapshot.Publisher
+			if config.HubURI != "" && config.HubKey != "" {
+				snapshotPublisher = livesnapshot.NewPublisher(livesnapshot.PublisherConfig{
+					HubURI:        config.HubURI,
+					HubKey:        config.HubKey,
+					HubPrivateKey: config.HubPrivateKey,
+					Region:        region,
+					DeviceKey:     deviceId,
+				})
+				log.Log.Info("cloud.HandleLiveStreamSD(): HTTP preview transport ENABLED; frames go to " + strings.TrimRight(config.HubURI, "/") + "/storage/snapshot when a viewer requests it (kept off MQTT).")
+			} else {
+				log.Log.Info("cloud.HandleLiveStreamSD(): HTTP preview transport DISABLED (Hub not configured: HubURI/HubKey empty); preview frames are pushed over MQTT.")
+			}
+
+			// Track the transport actually used so we log only when it changes; the
+			// loop runs once per keyframe and logging every frame would be noise.
+			lastTransport := ""
 
 			var cursorError error
 			var pkt packets.Packet
@@ -695,20 +726,74 @@ func HandleLiveStreamSD(livestreamCursor *packets.QueueCursor, configuration *mo
 					continue
 				}
 				now := time.Now().Unix()
+				// Drain both viewer keepalive channels (non-blocking): one for the
+				// HTTP transport, one for the legacy MQTT push.
 				select {
 				case <-communication.HandleLiveSD:
-					lastLivestreamRequest = now
+					lastLivestreamRequestMQTT = now
 				default:
 				}
-				if now-lastLivestreamRequest > 3 {
+				select {
+				case <-communication.HandleLiveSDHTTP:
+					lastLivestreamRequestHTTP = now
+				default:
+				}
+
+				mqttViewerActive := now-lastLivestreamRequestMQTT <= 3
+				httpViewerActive := now-lastLivestreamRequestHTTP <= 3
+				if !mqttViewerActive && !httpViewerActive {
 					continue
 				}
-				log.Log.Info("cloud.HandleLiveStreamSD(): Sending base64 encoded images to MQTT.")
-				img, err := rtspClient.DecodePacket(pkt)
-				if err == nil {
-					imageResized, _ := utils.ResizeImage(&img, uint(config.Capture.IPCamera.BaseWidth), uint(config.Capture.IPCamera.BaseHeight))
-					bytes, _ := utils.ImageToBytes(imageResized)
 
+				img, err := rtspClient.DecodePacket(pkt)
+				if err != nil {
+					continue
+				}
+				imageResized, _ := utils.ResizeImage(&img, uint(config.Capture.IPCamera.BaseWidth), uint(config.Capture.IPCamera.BaseHeight))
+				bytes, _ := utils.ImageToBytes(imageResized)
+
+				// Prefer HTTP for viewers that asked for it. Only if that did not
+				// deliver (Hub not configured, or the upload failed) do we also push
+				// over MQTT, so a new frontend can still fall back to its MQTT path.
+				httpPushed := false
+				var httpErr error
+				if httpViewerActive && snapshotPublisher != nil {
+					ctx, cancel := context.WithTimeout(context.Background(), 4*time.Second)
+					httpErr = snapshotPublisher.PublishSnapshot(ctx, bytes)
+					if httpErr == nil {
+						httpPushed = true
+					}
+					cancel()
+				}
+
+				pushMQTT := mqttViewerActive || (httpViewerActive && !httpPushed)
+
+				// Log only when the effective transport changes, so an operator can
+				// tell at a glance whether a device's preview travels over HTTP or
+				// MQTT (and why it fell back) without per-frame log spam.
+				transport := ""
+				if httpPushed {
+					transport = "http"
+				} else if pushMQTT {
+					transport = "mqtt"
+				}
+				if transport != "" && transport != lastTransport {
+					if transport == "http" {
+						log.Log.Info("cloud.HandleLiveStreamSD(): delivering preview frames over HTTP for device " + deviceId + ".")
+					} else {
+						reason := "viewer requested MQTT (older frontend)"
+						if httpViewerActive && snapshotPublisher == nil {
+							reason = "viewer asked for HTTP but Hub is not configured"
+						} else if httpViewerActive && httpErr != nil {
+							reason = "HTTP upload failed, falling back: " + httpErr.Error()
+						}
+						log.Log.Info("cloud.HandleLiveStreamSD(): delivering preview frames over MQTT for device " + deviceId + " (" + reason + ").")
+					}
+					lastTransport = transport
+				}
+
+				if pushMQTT {
+					log.Log.Debug("cloud.HandleLiveStreamSD(): Sending base64 encoded images to MQTT.")
 					chunking := config.Capture.LiveviewChunking
 
 					if chunking == "true" {

machinery/src/cloud/livehls.go

@@ -1,6 +1,7 @@
 package cloud
 
 import (
+	"os"
 	"time"
 
 	mqtt "github.com/eclipse/paho.mqtt.golang"
@@ -29,7 +30,6 @@ const hlsViewerTimeoutSeconds = 8
 // spamming the control plane.
 const hlsReadyReannounceSeconds = 2
 
-
 // HandleLiveStreamHLS drives the live HLS producer. It mirrors HandleLiveStreamSD:
 // it reads the camera's packet stream from a Latest() cursor, and while a viewer
 // is active (kept alive via communication.HandleLiveHLS) it muxes the packets
@@ -38,6 +38,14 @@ const hlsReadyReannounceSeconds = 2
 //
 // A session is created lazily on the first keyframe seen while a viewer is active
 // and torn down once viewers go away, so an idle camera produces no live traffic.
+//
+// By default (AGENT_LIVE_HLS_PREWARM unset or != "false") the agent instead keeps
+// one long-lived session muxing continuously into a small in-memory ring buffer
+// while idle (uploading nothing) and, the moment a viewer arrives, flushes the
+// already-encoded init + most-recent segment(s) and starts uploading live. This
+// trades a little idle CPU for a near-instant "requesting stream", so viewers no
+// longer wait a full GOP for the first segment to be cut. Set
+// AGENT_LIVE_HLS_PREWARM=false to fall back to the lazy on-demand path above.
 func HandleLiveStreamHLS(livestreamCursor *packets.QueueCursor, configuration *models.Configuration, communication *models.Communication, mqttClient mqtt.Client, _ capture.RTSPClient) {
 
 	log.Log.Debug("cloud.HandleLiveStreamHLS(): started")
@@ -78,6 +86,29 @@ func HandleLiveStreamHLS(livestreamCursor *packets.QueueCursor, configuration *m
 	width := uint16(config.Capture.IPCamera.Width)
 	height := uint16(config.Capture.IPCamera.Height)
 
+	// prewarm keeps a single long-lived session muxing into an in-memory ring
+	// buffer while idle and flushes it the instant a viewer arrives, eliminating
+	// the per-request GOP wait. Enabled by default; set AGENT_LIVE_HLS_PREWARM=false
+	// to fall back to the lazy on-demand path.
+	prewarm := os.Getenv("AGENT_LIVE_HLS_PREWARM") != "false"
+	if prewarm {
+		log.Log.Info("cloud.HandleLiveStreamHLS(): live HLS prewarm ENABLED (set AGENT_LIVE_HLS_PREWARM=false to disable)")
+	} else {
+		log.Log.Info("cloud.HandleLiveStreamHLS(): live HLS prewarm DISABLED (AGENT_LIVE_HLS_PREWARM=false)")
+	}
+
+	// lowLatency enables LL-HLS: each segment is sliced into CMAF parts shipped the
+	// instant they close and advertised via #EXT-X-PART, taking glass-to-glass HLS
+	// latency from ~4-6s down to ~1-2s. Enabled by default; set
+	// AGENT_LIVE_HLS_LOW_LATENCY=false to fall back to whole-segment HLS.
+	partTargetMs := uint64(0)
+	if os.Getenv("AGENT_LIVE_HLS_LOW_LATENCY") != "false" {
+		partTargetMs = livehls.DefaultPartTargetMs
+		log.Log.Info("cloud.HandleLiveStreamHLS(): live HLS low-latency (LL-HLS) ENABLED (set AGENT_LIVE_HLS_LOW_LATENCY=false to disable)")
+	} else {
+		log.Log.Info("cloud.HandleLiveStreamHLS(): live HLS low-latency (LL-HLS) DISABLED (AGENT_LIVE_HLS_LOW_LATENCY=false)")
+	}
+
 	var session *livehls.Session
 	lastViewerRequest := int64(0)
 	lastReadyAnnounce := int64(0)
@@ -97,7 +128,10 @@ func HandleLiveStreamHLS(livestreamCursor *packets.QueueCursor, configuration *m
 			// fired when this session's first segment landed. Re-announce (throttled)
 			// so late/refreshed viewers learn the active session id; the frontend
 			// dedupes by session id, so this is a no-op for viewers already playing.
-			if session != nil && session.IsReady() && now-lastReadyAnnounce >= hlsReadyReannounceSeconds {
+			// UploadsActive() is always true for the on-demand path; for prewarm it
+			// suppresses a stale re-announce while idle (the flush-on-arrival path
+			// below announces once the buffer has actually been shipped).
+			if session != nil && session.IsReady() && session.UploadsActive() && now-lastReadyAnnounce >= hlsReadyReannounceSeconds {
 				publishHLSReady(configuration, mqttClient, hubKey, deviceId, session.SessionID())
 				lastReadyAnnounce = now
 			}
@@ -106,6 +140,55 @@ func HandleLiveStreamHLS(livestreamCursor *packets.QueueCursor, configuration *m
 
 		viewerActive := now-lastViewerRequest <= hlsViewerTimeoutSeconds
 
+		if prewarm {
+			// Keep one long-lived session muxing into the ring buffer. Create it on
+			// the first keyframe (so the buffer opens on a random-access point) and
+			// never tear it down for idleness; uploads, not muxing, are what we gate
+			// on viewer presence.
+			if session == nil {
+				if len(pkt.Data) == 0 || !pkt.IsVideo || !pkt.IsKeyFrame {
+					continue
+				}
+				session = livehls.NewSession(publisher, livehls.SessionOptions{
+					Codec:          pkt.Codec,
+					SPSNALUs:       config.Capture.IPCamera.SPSNALUs,
+					PPSNALUs:       config.Capture.IPCamera.PPSNALUs,
+					VPSNALUs:       config.Capture.IPCamera.VPSNALUs,
+					Width:          width,
+					Height:         height,
+					PartTargetMs:   partTargetMs,
+					StartBuffering: true,
+				})
+				session.SetOnReady(func(sessionID string) {
+					log.Log.Info("cloud.HandleLiveStreamHLS(): live HLS session ready, announcing " + sessionID)
+					publishHLSReady(configuration, mqttClient, hubKey, deviceId, sessionID)
+					lastReadyAnnounce = time.Now().Unix()
+				})
+				log.Log.Info("cloud.HandleLiveStreamHLS(): prewarming live HLS session " + session.SessionID())
+			}
+
+			if viewerActive {
+				// Activating flushes the cached init + buffered segment(s). onReady
+				// announces the first-ever readiness; on a later re-activation it has
+				// already fired, so announce here (throttled, so the first activation
+				// does not double up) once the buffer has actually been shipped.
+				if session.SetUploadsActive(true) && session.IsReady() && now-lastReadyAnnounce >= hlsReadyReannounceSeconds {
+					publishHLSReady(configuration, mqttClient, hubKey, deviceId, session.SessionID())
+					lastReadyAnnounce = now
+				}
+			} else {
+				// No viewer: keep muxing into the buffer but stop uploading.
+				session.SetUploadsActive(false)
+			}
+
+			if len(pkt.Data) > 0 && pkt.IsVideo {
+				if err := session.WritePacket(pkt); err != nil {
+					log.Log.Error("cloud.HandleLiveStreamHLS(): " + err.Error())
+				}
+			}
+			continue
+		}
+
 		if !viewerActive {
 			// No viewer: stop and discard the session so we stop shipping segments.
 			if session != nil {
@@ -127,12 +210,13 @@ func HandleLiveStreamHLS(livestreamCursor *packets.QueueCursor, configuration *m
 				continue
 			}
 			session = livehls.NewSession(publisher, livehls.SessionOptions{
-				Codec:    pkt.Codec,
-				SPSNALUs: config.Capture.IPCamera.SPSNALUs,
-				PPSNALUs: config.Capture.IPCamera.PPSNALUs,
-				VPSNALUs: config.Capture.IPCamera.VPSNALUs,
-				Width:    width,
-				Height:   height,
+				Codec:        pkt.Codec,
+				SPSNALUs:     config.Capture.IPCamera.SPSNALUs,
+				PPSNALUs:     config.Capture.IPCamera.PPSNALUs,
+				VPSNALUs:     config.Capture.IPCamera.VPSNALUs,
+				Width:        width,
+				Height:       height,
+				PartTargetMs: partTargetMs,
 			})
 			session.SetOnReady(func(sessionID string) {
 				log.Log.Info("cloud.HandleLiveStreamHLS(): live HLS session ready, announcing " + sessionID)

machinery/src/cloud/livehls/publisher.go

@@ -57,6 +57,11 @@ const (
 	headerLiveName      = "X-Kerberos-Live-Name"
 	headerLiveSequence  = "X-Kerberos-Live-Sequence"
 	headerLiveDuration  = "X-Kerberos-Live-Duration"
+	// Low-latency (LL-HLS) part headers. A part belongs to media segment
+	// X-Kerberos-Live-Sequence and is the X-Kerberos-Live-Part-th chunk within it;
+	// X-Kerberos-Live-Part-Independent flags a part that starts on a keyframe.
+	headerLivePart            = "X-Kerberos-Live-Part"
+	headerLivePartIndependent = "X-Kerberos-Live-Part-Independent"
 
 	// defaultPublishTimeout bounds a single segment upload. A live segment that
 	// cannot be delivered within roughly its own duration is stale, so the upload
@@ -136,12 +141,33 @@ func (p *Publisher) PublishSegment(ctx context.Context, sessionID string, seg vi
 	})
 }
 
+// PublishPart uploads one CMAF partial segment (LL-HLS). The part is named
+// seg-<segment>.<part>.m4s and carries its segment sequence, part index,
+// independence flag and duration in headers so hub-api can advertise it via
+// #EXT-X-PART and reconstruct the full segment by concatenating its parts.
+func (p *Publisher) PublishPart(ctx context.Context, sessionID string, part video.LivePart) error {
+	return p.post(ctx, postParams{
+		sessionID:   sessionID,
+		name:        fmt.Sprintf("seg-%d.%d.m4s", part.SegmentSeq, part.PartIndex),
+		sequence:    part.SegmentSeq,
+		durationMs:  part.DurationMs,
+		partIndex:   part.PartIndex,
+		independent: part.Independent,
+		hasPart:     true,
+		contentType: contentTypeSegment,
+		body:        part.Data,
+	})
+}
+
 type postParams struct {
 	sessionID   string
 	name        string
 	sequence    uint32
 	durationMs  uint64
 	hasSegment  bool
+	partIndex   uint32
+	independent bool
+	hasPart     bool
 	contentType string
 	body        []byte
 }
@@ -165,10 +191,18 @@ func (p *Publisher) post(ctx context.Context, params postParams) error {
 	req.Header.Set(headerStorageDevice, p.cfg.DeviceKey)
 	req.Header.Set(headerLiveSession, params.sessionID)
 	req.Header.Set(headerLiveName, params.name)
-	if params.hasSegment {
+	if params.hasSegment || params.hasPart {
 		req.Header.Set(headerLiveSequence, strconv.FormatUint(uint64(params.sequence), 10))
 		req.Header.Set(headerLiveDuration, strconv.FormatUint(params.durationMs, 10))
 	}
+	if params.hasPart {
+		req.Header.Set(headerLivePart, strconv.FormatUint(uint64(params.partIndex), 10))
+		independent := "0"
+		if params.independent {
+			independent = "1"
+		}
+		req.Header.Set(headerLivePartIndependent, independent)
+	}
 	req.Header.Set(headerHubPublicKey, p.cfg.HubKey)
 	req.Header.Set(headerHubPrivateKey, p.cfg.HubPrivateKey)
 	req.Header.Set(headerHubRegion, p.cfg.Region)

machinery/src/cloud/livehls/session.go

@@ -18,6 +18,12 @@ import (
 // large enough that per-segment HTTP overhead is negligible.
 const DefaultTargetSegmentMs = 2000
 
+// DefaultPartTargetMs is the nominal LL-HLS part length used when low latency is
+// enabled. ~300ms parts yield ~6-7 parts per 2s segment; with the playlist's
+// PART-HOLD-BACK at ~3x the part target this lands glass-to-glass latency around
+// 1-2s (versus ~4-6s for whole-segment HLS).
+const DefaultPartTargetMs = 300
+
 // Session ties a video.LiveSegmenter to a Publisher: it converts capture packets
 // into CMAF segments and ships each one to hub-api. Exactly one init segment is
 // delivered per session (re-attempted until it lands), after which media
@@ -43,6 +49,21 @@ type Session struct {
 	lastInitAt time.Time
 	readyFired bool
 	onReady    func(sessionID string)
+
+	// uploadsActive gates whether the init and completed segments are shipped to
+	// hub-api. It is true for the default on-demand path. The prewarm path starts
+	// it false so the session keeps muxing into bufferedSegments without producing
+	// any live traffic until a viewer actually arrives; see SetUploadsActive.
+	uploadsActive bool
+	// bufferedSegments is the in-memory ring buffer (the most recent
+	// prewarmMaxBufferedSegments segments) kept while uploadsActive is false, so a
+	// viewer that arrives can be served an already-encoded segment immediately
+	// instead of waiting a full GOP for the next one to be cut.
+	bufferedSegments []video.LiveSegment
+	// bufferedParts is the LL-HLS counterpart of bufferedSegments: while idle it
+	// retains the parts of the most recent (prewarmMaxBufferedSegments+1) segments,
+	// pruned a WHOLE segment at a time so a flushed segment is never partial.
+	bufferedParts []video.LivePart
 }
 
 // SessionOptions configures a live HLS session.
@@ -54,6 +75,16 @@ type SessionOptions struct {
 	Width           uint16   // encoded width (for the avcC fallback path)
 	Height          uint16   // encoded height
 	TargetSegmentMs uint64   // 0 => DefaultTargetSegmentMs
+	// PartTargetMs, when > 0, enables LL-HLS: each segment is additionally sliced
+	// into ~PartTargetMs CMAF parts that are published (and advertised via
+	// #EXT-X-PART) the instant they close, for ~1-2s glass-to-glass latency. 0
+	// keeps the classic whole-segment path.
+	PartTargetMs uint64
+	// StartBuffering starts the session in prewarm (buffer-only) mode: it muxes
+	// segments into an in-memory ring buffer but uploads nothing until
+	// SetUploadsActive(true) is called. Default false => uploads are live
+	// immediately (the on-demand path's behaviour).
+	StartBuffering bool
 }
 
 // NewSession builds a session with a fresh random id and wires the segmenter's
@@ -65,11 +96,16 @@ func NewSession(publisher *Publisher, opts SessionOptions) *Session {
 	}
 	seg := video.NewLiveSegmenter(opts.Codec, opts.SPSNALUs, opts.PPSNALUs, opts.VPSNALUs, target)
 	seg.SetDimensions(opts.Width, opts.Height)
+	if opts.PartTargetMs > 0 {
+		seg.EnableLowLatency(opts.PartTargetMs)
+	}
 
 	s := &Session{
 		id:        newSessionID(),
 		publisher: publisher,
 		segmenter: seg,
+		// Uploads are live by default; the prewarm path opts into buffer-only mode.
+		uploadsActive: !opts.StartBuffering,
 		newContext: func() (context.Context, context.CancelFunc) {
 			return context.WithTimeout(context.Background(), defaultPublishTimeout)
 		},
@@ -82,8 +118,13 @@ func NewSession(publisher *Publisher, opts SessionOptions) *Session {
 	seg.OnInit = func(initBytes []byte) error {
 		s.mu.Lock()
 		s.initBytes = append([]byte(nil), initBytes...)
+		active := s.uploadsActive
 		s.mu.Unlock()
-		s.publishInitIfNeeded()
+		// While prewarming we cache the init in memory but ship nothing; it is
+		// uploaded on the first SetUploadsActive(true) flush.
+		if active {
+			s.publishInitIfNeeded()
+		}
 		return nil
 	}
 
@@ -91,6 +132,15 @@ func NewSession(publisher *Publisher, opts SessionOptions) *Session {
 	// init segment has landed (a media segment is useless without it), and we fire
 	// OnReady after the first successfully shipped segment.
 	seg.OnSegment = func(segment video.LiveSegment) error {
+		s.mu.Lock()
+		active := s.uploadsActive
+		s.mu.Unlock()
+		if !active {
+			// Prewarm: retain the most recent segments in memory but upload nothing
+			// until a viewer arrives (SetUploadsActive flushes them).
+			s.bufferSegment(segment)
+			return nil
+		}
 		if !s.publishInitIfNeeded() {
 			log.Log.Warning("livehls.Session: dropping segment " +
 				fmt.Sprintf("%d", segment.SequenceNumber) + " because init has not been delivered yet")
@@ -109,6 +159,36 @@ func NewSession(publisher *Publisher, opts SessionOptions) *Session {
 		return nil
 	}
 
+	// In LL-HLS mode the segmenter emits parts (not whole segments); ship each one
+	// the instant it closes. Mirrors OnSegment: buffer while prewarming, otherwise
+	// publish after the init has landed and fire OnReady on the first part.
+	if opts.PartTargetMs > 0 {
+		seg.OnPart = func(part video.LivePart) error {
+			s.mu.Lock()
+			active := s.uploadsActive
+			s.mu.Unlock()
+			if !active {
+				s.bufferPart(part)
+				return nil
+			}
+			if !s.publishInitIfNeeded() {
+				log.Log.Warning("livehls.Session: dropping part " +
+					fmt.Sprintf("%d.%d", part.SegmentSeq, part.PartIndex) +
+					" because init has not been delivered yet")
+				return nil
+			}
+			ctx, cancel := s.newContext()
+			defer cancel()
+			if err := s.publisher.PublishPart(ctx, s.id, part); err != nil {
+				log.Log.Warning("livehls.Session: " + err.Error())
+				return nil
+			}
+			s.fireReadyOnce()
+			s.refreshInitIfStale()
+			return nil
+		}
+	}
+
 	return s
 }
 
@@ -136,6 +216,128 @@ func (s *Session) SetOnReady(fn func(sessionID string)) {
 	s.mu.Unlock()
 }
 
+// prewarmMaxBufferedSegments is how many of the most recent completed segments
+// the prewarm path keeps in memory while idle and flushes to a viewer on arrival.
+// One segment keeps startup instant (the viewer immediately gets a playable
+// segment) while starting as close to the live edge as possible, so the HLS view
+// tracks the WebRTC/live edge instead of opening several seconds behind; hls.js
+// then converges to the edge via maxLiveSyncPlaybackRate. Raising it trades
+// latency-from-live for a little more startup cushion.
+const prewarmMaxBufferedSegments = 1
+
+// SetUploadsActive toggles whether the session ships its init and segments to
+// hub-api, and reports whether this call flipped it from inactive to active.
+//
+// While uploads are inactive the session keeps muxing capture packets into an
+// in-memory ring buffer (the cached init plus the most recent
+// prewarmMaxBufferedSegments segments) but uploads nothing, so an idle camera
+// produces no live traffic. Switching from inactive to active immediately
+// flushes the cached init and buffered segments so a viewer can start almost
+// instantly instead of waiting a full GOP for the next segment to be cut.
+// Switching from active to inactive resets the init-published flag so the next
+// activation re-uploads the init (it may have aged out of the hub's short-TTL
+// live window while idle). All other transitions are no-ops. Driven from the
+// live-stream goroutine; not safe for concurrent use.
+func (s *Session) SetUploadsActive(active bool) bool {
+	s.mu.Lock()
+	if s.uploadsActive == active {
+		s.mu.Unlock()
+		return false
+	}
+	s.uploadsActive = active
+	if !active {
+		// Going idle: force the next activation to re-deliver the init segment,
+		// which may have expired from the hub live window while nobody was watching.
+		s.initPublished = false
+		s.mu.Unlock()
+		return false
+	}
+	// Inactive -> active: take the cached buffered segments/parts and flush them
+	// outside the lock (the publish calls take their own time and re-acquire the
+	// mutex).
+	buffered := s.bufferedSegments
+	bufferedParts := s.bufferedParts
+	s.bufferedSegments = nil
+	s.bufferedParts = nil
+	s.mu.Unlock()
+
+	// Deliver the init first; media segments are useless without it.
+	for i := range buffered {
+		if !s.publishInitIfNeeded() {
+			break
+		}
+		ctx, cancel := s.newContext()
+		if err := s.publisher.PublishSegment(ctx, s.id, buffered[i]); err != nil {
+			log.Log.Warning("livehls.Session: prewarm flush: " + err.Error())
+			cancel()
+			continue
+		}
+		cancel()
+		s.fireReadyOnce()
+		s.refreshInitIfStale()
+	}
+	// LL-HLS: flush the buffered parts in order (oldest first) so the viewer gets a
+	// playable, near-live window immediately.
+	for i := range bufferedParts {
+		if !s.publishInitIfNeeded() {
+			break
+		}
+		ctx, cancel := s.newContext()
+		if err := s.publisher.PublishPart(ctx, s.id, bufferedParts[i]); err != nil {
+			log.Log.Warning("livehls.Session: prewarm flush (part): " + err.Error())
+			cancel()
+			continue
+		}
+		cancel()
+		s.fireReadyOnce()
+		s.refreshInitIfStale()
+	}
+	return true
+}
+
+// UploadsActive reports whether the session is currently shipping segments (as
+// opposed to buffering them while prewarming). Always true for the on-demand
+// path.
+func (s *Session) UploadsActive() bool {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.uploadsActive
+}
+
+// bufferSegment appends a completed segment to the in-memory prewarm ring buffer,
+// discarding the oldest so at most prewarmMaxBufferedSegments are retained.
+func (s *Session) bufferSegment(seg video.LiveSegment) {
+	s.mu.Lock()
+	s.bufferedSegments = append(s.bufferedSegments, seg)
+	if overflow := len(s.bufferedSegments) - prewarmMaxBufferedSegments; overflow > 0 {
+		// Drop the oldest segment(s) and shrink the backing array so retained bytes
+		// stay bounded.
+		s.bufferedSegments = append([]video.LiveSegment(nil), s.bufferedSegments[overflow:]...)
+	}
+	s.mu.Unlock()
+}
+
+// bufferPart appends a part to the LL-HLS prewarm ring buffer, pruning whole
+// older segments (never individual parts) so the retained window always consists
+// of complete segments plus the in-progress one. Pruning on a part-0 boundary
+// keeps at most prewarmMaxBufferedSegments fully-buffered segments behind the
+// current one, which guarantees a flushed segment can be reconstructed in full.
+func (s *Session) bufferPart(part video.LivePart) {
+	s.mu.Lock()
+	s.bufferedParts = append(s.bufferedParts, part)
+	if part.PartIndex == 0 && part.SegmentSeq > uint32(prewarmMaxBufferedSegments) {
+		minSeg := part.SegmentSeq - uint32(prewarmMaxBufferedSegments)
+		kept := make([]video.LivePart, 0, len(s.bufferedParts))
+		for _, p := range s.bufferedParts {
+			if p.SegmentSeq >= minSeg {
+				kept = append(kept, p)
+			}
+		}
+		s.bufferedParts = kept
+	}
+	s.mu.Unlock()
+}
+
 // WritePacket feeds one capture packet into the segmenter. Non-video packets are
 // ignored (the spike is video-only). The decode timestamp is derived exactly as
 // the recording muxer does: DTS = PTS - compositionOffset, with the composition

machinery/src/cloud/livesnapshot/publisher.go

@@ -0,0 +1,151 @@
+// Package livesnapshot implements the agent-side producer for the live-view
+// "preview" (SD) mode over HTTP.
+//
+// Historically the preview pipeline shipped each resized keyframe (a base64
+// JPEG, often chunked) to viewers over the MQTT broker. MQTT is a control plane
+// for small messages, so pushing ~1 image/second of base64 image data per
+// watched camera congests the broker and delays genuine control traffic. This
+// package moves those frames off MQTT: the agent POSTs the latest resized JPEG
+// straight to hub-api over plain HTTPS (outbound only), and viewers fetch it
+// back with their session token. Only the tiny "a viewer is watching" keepalive
+// stays on MQTT.
+//
+// The wire contract (agent -> hub-api) deliberately mirrors the live HLS ingest
+// and the existing storage-upload convention (X-Kerberos-Storage-Device plus the
+// Hub public/private key auth headers). hub-api authenticates the agent and
+// stores the frame in an ephemeral, short-TTL per-device slot which it serves
+// straight back to authorized viewers; the frame never enters the vault or the
+// recordings collection.
+//
+// Like live HLS segments, a preview frame is worthless once stale: a frame that
+// fails to upload is superseded by the next one a second later, so the publisher
+// is fire-and-forget and drops on failure (logged) rather than retrying.
+package livesnapshot
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"net/http"
+	"strings"
+	"time"
+
+	"github.com/kerberos-io/agent/machinery/src/log"
+)
+
+const (
+	// snapshotIngestPath is the hub-api endpoint that accepts the latest preview
+	// frame and stores it in the device's ephemeral snapshot slot (mirrors the
+	// /storage/live live-HLS ingest convention).
+	snapshotIngestPath = "/storage/snapshot"
+
+	contentTypeJPEG = "image/jpeg"
+
+	// Header names for the snapshot ingest contract (shared with live HLS / storage).
+	headerHubPublicKey  = "X-Kerberos-Hub-PublicKey"
+	headerHubPrivateKey = "X-Kerberos-Hub-PrivateKey"
+	headerHubRegion     = "X-Kerberos-Hub-Region"
+	headerStorageDevice = "X-Kerberos-Storage-Device"
+
+	// defaultPublishTimeout bounds a single snapshot upload. Preview frames are
+	// produced roughly once a second from a single goroutine, so an upload that
+	// cannot land in a few seconds is abandoned rather than allowed to back up the
+	// preview loop behind a slow request.
+	defaultPublishTimeout = 4 * time.Second
+)
+
+// PublisherConfig carries the hub endpoint and credentials needed to ship
+// preview frames. It is populated from the agent's models.Config (the same
+// HubURI/HubKey/HubPrivateKey used by recordings and live HLS).
+type PublisherConfig struct {
+	HubURI        string // base hub-api URL, e.g. https://api.hub.example.com
+	HubKey        string // Hub public key (X-Kerberos-Hub-PublicKey)
+	HubPrivateKey string // Hub private key (X-Kerberos-Hub-PrivateKey)
+	Region        string // storage region (X-Kerberos-Hub-Region), may be empty
+	DeviceKey     string // device/camera key (X-Kerberos-Storage-Device)
+
+	// Timeout optionally overrides defaultPublishTimeout (used by tests).
+	Timeout time.Duration
+	// HTTPClient optionally injects a client (used by tests). When nil a
+	// redirect-credential-stripping client is created.
+	HTTPClient *http.Client
+}
+
+// Publisher ships the latest preview frame to hub-api over plain HTTP POST.
+//
+// It is safe for sequential use from a single live-stream goroutine. PublishSnapshot
+// is fire-and-forget: it returns an error for the caller to log, but the caller is
+// expected to continue (drop-on-fail) rather than retry.
+type Publisher struct {
+	cfg    PublisherConfig
+	client *http.Client
+}
+
+// NewPublisher builds a Publisher. The HTTP client strips the Hub credential
+// headers on a cross-host redirect (net/http does this for standard auth headers
+// but not custom-named ones), matching the recording/live-HLS upload clients.
+func NewPublisher(cfg PublisherConfig) *Publisher {
+	client := cfg.HTTPClient
+	if client == nil {
+		timeout := cfg.Timeout
+		if timeout <= 0 {
+			timeout = defaultPublishTimeout
+		}
+		client = &http.Client{
+			Timeout:       timeout,
+			CheckRedirect: stripHubCredentialsOnCrossHostRedirect,
+		}
+	}
+	return &Publisher{cfg: cfg, client: client}
+}
+
+// PublishSnapshot uploads a single resized preview frame (JPEG) as the device's
+// latest snapshot. It overwrites whatever frame was there before, so viewers
+// always fetch the most recent frame.
+func (p *Publisher) PublishSnapshot(ctx context.Context, jpeg []byte) error {
+	if p.cfg.HubURI == "" {
+		return fmt.Errorf("livesnapshot: HubURI not configured")
+	}
+	if len(jpeg) == 0 {
+		return fmt.Errorf("livesnapshot: empty snapshot body")
+	}
+
+	url := strings.TrimRight(p.cfg.HubURI, "/") + snapshotIngestPath
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(jpeg))
+	if err != nil {
+		return fmt.Errorf("livesnapshot: build request: %w", err)
+	}
+
+	req.Header.Set("Content-Type", contentTypeJPEG)
+	req.Header.Set(headerStorageDevice, p.cfg.DeviceKey)
+	req.Header.Set(headerHubPublicKey, p.cfg.HubKey)
+	req.Header.Set(headerHubPrivateKey, p.cfg.HubPrivateKey)
+	req.Header.Set(headerHubRegion, p.cfg.Region)
+
+	resp, err := p.client.Do(req)
+	if err != nil {
+		return fmt.Errorf("livesnapshot: upload snapshot: %w", err)
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
+		return fmt.Errorf("livesnapshot: upload snapshot rejected: %s", resp.Status)
+	}
+	log.Log.Debug("livesnapshot.Publisher.PublishSnapshot(): shipped preview frame for device " + p.cfg.DeviceKey)
+	return nil
+}
+
+// stripHubCredentialsOnCrossHostRedirect removes the Hub credential headers when
+// a redirect crosses to a different host. net/http strips standard sensitive
+// headers on a cross-host redirect but not custom-named ones, so without this the
+// Hub keys could leak to a redirect target.
+func stripHubCredentialsOnCrossHostRedirect(req *http.Request, via []*http.Request) error {
+	if len(via) == 0 {
+		return nil
+	}
+	if req.URL.Host != via[0].URL.Host {
+		req.Header.Del(headerHubPrivateKey)
+		req.Header.Del(headerHubPublicKey)
+	}
+	return nil
+}

machinery/src/components/kerberos.go

@@ -70,6 +70,7 @@ func Bootstrap(ctx context.Context, configDirectory string, configuration *model
 	communication.HandleUpload = make(chan string, 1)
 	communication.HandleHeartBeat = make(chan string, 1)
 	communication.HandleLiveSD = make(chan int64, 1)
+	communication.HandleLiveSDHTTP = make(chan int64, 1)
 	communication.HandleLiveHDKeepalive = make(chan string, 1)
 	communication.HandleLiveHDPeers = make(chan string, 1)
 	communication.HandleLiveHLS = make(chan int64, 1)
@@ -177,19 +178,10 @@ func RunAgent(configDirectory string, configuration *models.Configuration, commu
 	configuration.Config.Capture.IPCamera.Height = height
 
 	// Set the liveview width and height, this is used for the liveview and motion regions (drawing on the hub).
-	baseWidth := config.Capture.IPCamera.BaseWidth
-	baseHeight := config.Capture.IPCamera.BaseHeight
-	// If the liveview height is not set, we will calculate it based on the width and aspect ratio of the camera.
-	if baseWidth > 0 && baseHeight == 0 {
-		widthAspectRatio := float64(baseWidth) / float64(width)
-		configuration.Config.Capture.IPCamera.BaseHeight = int(float64(height) * widthAspectRatio)
-	} else if baseHeight > 0 && baseWidth > 0 {
-		configuration.Config.Capture.IPCamera.BaseHeight = baseHeight
-		configuration.Config.Capture.IPCamera.BaseWidth = baseWidth
-	} else {
-		configuration.Config.Capture.IPCamera.BaseHeight = height
-		configuration.Config.Capture.IPCamera.BaseWidth = width
-	}
+	// ResolveBaseDimensions gates the aspect-ratio compute on width/height > 0
+	// so a not-yet-probed stream can't poison the dimensions and crash resize.
+	configuration.Config.Capture.IPCamera.BaseWidth, configuration.Config.Capture.IPCamera.BaseHeight =
+		utils.ResolveBaseDimensions(config.Capture.IPCamera.BaseWidth, config.Capture.IPCamera.BaseHeight, width, height)
 
 	// Set the SPS and PPS values in the configuration.
 	configuration.Config.Capture.IPCamera.SPSNALUs = [][]byte{videoStream.SPS}
@@ -247,19 +239,8 @@ func RunAgent(configDirectory string, configuration *models.Configuration, commu
 
 		// If we have a substream, we need to set the width and height of the substream. (so we will override above information)
 		// Set the liveview width and height, this is used for the liveview and motion regions (drawing on the hub).
-		baseWidth := config.Capture.IPCamera.BaseWidth
-		baseHeight := config.Capture.IPCamera.BaseHeight
-		// If the liveview height is not set, we will calculate it based on the width and aspect ratio of the camera.
-		if baseWidth > 0 && baseHeight == 0 {
-			widthAspectRatio := float64(baseWidth) / float64(width)
-			configuration.Config.Capture.IPCamera.BaseHeight = int(float64(height) * widthAspectRatio)
-		} else if baseHeight > 0 && baseWidth > 0 {
-			configuration.Config.Capture.IPCamera.BaseHeight = baseHeight
-			configuration.Config.Capture.IPCamera.BaseWidth = baseWidth
-		} else {
-			configuration.Config.Capture.IPCamera.BaseHeight = height
-			configuration.Config.Capture.IPCamera.BaseWidth = width
-		}
+		configuration.Config.Capture.IPCamera.BaseWidth, configuration.Config.Capture.IPCamera.BaseHeight =
+			utils.ResolveBaseDimensions(config.Capture.IPCamera.BaseWidth, config.Capture.IPCamera.BaseHeight, width, height)
 	}
 
 	// We are creating a queue to store the RTSP frames in, these frames will be

machinery/src/models/communication.go

@@ -37,6 +37,7 @@ type Communication struct {
 	HandleUpload          chan string
 	HandleHeartBeat       chan string
 	HandleLiveSD          chan int64
+	HandleLiveSDHTTP      chan int64
 	HandleLiveHDKeepalive chan string
 	HandleLiveHDHandshake chan LiveHDHandshake
 	HandleLiveHDPeers     chan string

machinery/src/models/mqtt.go

@@ -171,6 +171,12 @@ type UpdateConfigPayload struct {
 // We received a request SD stream request
 type RequestSDStreamPayload struct {
 	Timestamp int64 `json:"timestamp"` // timestamp
+	// Transport selects how the agent should deliver the preview frames for this
+	// viewer. "http" asks the agent to POST frames to hub-api (keeping them off
+	// MQTT); empty/absent means the legacy MQTT image push. Older agents simply
+	// ignore this unknown field and keep doing MQTT, and older frontends never set
+	// it — so new/old agents and frontends interoperate in every combination.
+	Transport string `json:"transport,omitempty"`
 }
 
 // We received a live HLS stream request. Like SD it is a simple viewer

machinery/src/routers/mqtt/main.go

@@ -550,9 +550,20 @@ func HandleRequestSDStream(mqttClient mqtt.Client, hubKey string, payload models
 
 	if requestSDStreamPayload.Timestamp != 0 {
 		if communication.CameraConnected {
-			select {
-			case communication.HandleLiveSD <- time.Now().Unix():
-			default:
+			// A viewer that opted into the HTTP transport is signalled on a separate
+			// channel so the producer ships its frames to hub-api over HTTP instead of
+			// publishing them over MQTT. Any other (or absent) transport keeps the
+			// legacy MQTT image push, so older frontends behave exactly as before.
+			if requestSDStreamPayload.Transport == "http" {
+				select {
+				case communication.HandleLiveSDHTTP <- time.Now().Unix():
+				default:
+				}
+			} else {
+				select {
+				case communication.HandleLiveSD <- time.Now().Unix():
+				default:
+				}
 			}
 			log.Log.Info("routers.mqtt.main.HandleRequestSDStream(): received request to livestream.")
 		} else {

machinery/src/utils/main.go

@@ -427,12 +427,49 @@ func ResizeImage(img image.Image, newWidth uint, newHeight uint) (*image.Image,
 		return nil, errors.New("image is nil")
 	}
 
+	// Callers cast int->uint, so a negative or poisoned int (e.g. MinInt from
+	// `int(float * +Inf)` when the source width is 0) wraps to a near-MaxUint
+	// value here and crashes nfnt/resize's allocator with "makeslice: len out
+	// of range". Clamp anything past a sane camera ceiling to 0 ("auto" in
+	// nfnt — preserves aspect from the source).
+	const maxDim uint = 8192
+	if newWidth > maxDim {
+		newWidth = 0
+	}
+	if newHeight > maxDim {
+		newHeight = 0
+	}
+
 	// resize to width 640 using Lanczos resampling
 	// and preserve aspect ratio
 	m := resize.Resize(newWidth, newHeight, img, resize.Lanczos3)
 	return &m, nil
 }
 
+// ResolveBaseDimensions resolves the liveview/motion base dimensions for a
+// stream given the (optionally configured) base width/height and the camera's
+// probed source width/height. It returns the width and height that should be
+// stored on the configuration.
+//
+// The aspect-ratio branch is gated on width>0 && height>0: a not-yet-probed
+// stream has width=height=0, which previously made the ratio +Inf and
+// int(float * +Inf) yield MinInt. That poisoned value, later cast to uint at
+// the ResizeImage call sites, wrapped to ~MaxUint and crashed resize with
+// "makeslice: len out of range". When the source isn't probed yet we fall back
+// to the source dimensions (0,0 -> "auto") instead.
+func ResolveBaseDimensions(baseWidth, baseHeight, width, height int) (int, int) {
+	if baseWidth > 0 && baseHeight == 0 && width > 0 && height > 0 {
+		// Derive the height from the configured width and the source aspect ratio.
+		widthAspectRatio := float64(baseWidth) / float64(width)
+		return baseWidth, int(float64(height) * widthAspectRatio)
+	} else if baseHeight > 0 && baseWidth > 0 {
+		// Both base dimensions are configured; honor them as-is.
+		return baseWidth, baseHeight
+	}
+	// Nothing usable configured (or source not probed yet): use source dimensions.
+	return width, height
+}
+
 func ResizeHeightWithAspectRatio(newWidth int, width int, height int) (int, int) {
 	if newWidth <= 0 || width <= 0 || height <= 0 {
 		return width, height

machinery/src/utils/resize_test.go

@@ -0,0 +1,124 @@
+package utils
+
+import (
+	"image"
+	"math"
+	"testing"
+)
+
+func TestResolveBaseDimensions(t *testing.T) {
+	tests := []struct {
+		name                  string
+		baseWidth, baseHeight int
+		width, height         int
+		wantWidth, wantHeight int
+	}{
+		{
+			name:      "base width set, height derived from aspect ratio",
+			baseWidth: 640, baseHeight: 0,
+			width: 1920, height: 1080,
+			wantWidth: 640, wantHeight: 360,
+		},
+		{
+			name:      "both base dimensions configured are honored",
+			baseWidth: 640, baseHeight: 480,
+			width: 1920, height: 1080,
+			wantWidth: 640, wantHeight: 480,
+		},
+		{
+			name:      "no base configured falls back to source dimensions",
+			baseWidth: 0, baseHeight: 0,
+			width: 1920, height: 1080,
+			wantWidth: 1920, wantHeight: 1080,
+		},
+		{
+			// Regression: a not-yet-probed stream has width=height=0. The old
+			// aspect-ratio branch divided by zero (float * +Inf -> MinInt) and
+			// poisoned BaseHeight, later crashing resize with makeslice panic.
+			name:      "unprobed stream (width=0) does not poison dimensions",
+			baseWidth: 640, baseHeight: 0,
+			width: 0, height: 0,
+			wantWidth: 0, wantHeight: 0,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			gotWidth, gotHeight := ResolveBaseDimensions(tt.baseWidth, tt.baseHeight, tt.width, tt.height)
+			if gotWidth != tt.wantWidth || gotHeight != tt.wantHeight {
+				t.Fatalf("ResolveBaseDimensions(%d,%d,%d,%d) = (%d,%d), want (%d,%d)",
+					tt.baseWidth, tt.baseHeight, tt.width, tt.height,
+					gotWidth, gotHeight, tt.wantWidth, tt.wantHeight)
+			}
+		})
+	}
+}
+
+func TestResolveBaseDimensionsNeverNegative(t *testing.T) {
+	// Whatever the inputs, the resolved dimensions must never be negative,
+	// otherwise the uint cast at the resize call sites wraps to ~MaxUint.
+	for _, c := range [][4]int{
+		{640, 0, 0, 0},
+		{640, 0, 0, 1080},
+		{640, 0, 1920, 0},
+		{0, 0, 0, 0},
+	} {
+		w, h := ResolveBaseDimensions(c[0], c[1], c[2], c[3])
+		if w < 0 || h < 0 {
+			t.Fatalf("ResolveBaseDimensions(%v) produced negative dims (%d,%d)", c, w, h)
+		}
+	}
+}
+
+func TestResizeImageClampsPoisonedDimensions(t *testing.T) {
+	src := image.NewRGBA(image.Rect(0, 0, 320, 240))
+
+	// uint(math.MinInt) is the value produced when a poisoned int (from
+	// int(float * +Inf)) is cast to uint at a call site. It must not panic
+	// nfnt/resize's allocator; it should fall back to source-aspect resize.
+	// Compute via a runtime int so the conversion doesn't overflow at compile time.
+	minInt := math.MinInt
+	poison := uint(minInt)
+
+	resized, err := ResizeImage(src, poison, poison)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if resized == nil {
+		t.Fatalf("expected an image, got nil")
+	}
+	b := (*resized).Bounds()
+	if b.Dx() != 320 || b.Dy() != 240 {
+		t.Fatalf("poisoned dims should fall back to source size, got %dx%d", b.Dx(), b.Dy())
+	}
+}
+
+func TestResizeImageClampsAboveCameraCeiling(t *testing.T) {
+	src := image.NewRGBA(image.Rect(0, 0, 320, 240))
+
+	// A width beyond any sane camera resolution is treated as "auto" (0).
+	resized, err := ResizeImage(src, 100000, 0)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	b := (*resized).Bounds()
+	if b.Dx() != 320 || b.Dy() != 240 {
+		t.Fatalf("oversized width should fall back to source size, got %dx%d", b.Dx(), b.Dy())
+	}
+}
+
+func TestResizeImageNormalResizeStillWorks(t *testing.T) {
+	src := image.NewRGBA(image.Rect(0, 0, 1920, 1080))
+
+	resized, err := ResizeImage(src, 640, 0)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	b := (*resized).Bounds()
+	if b.Dx() != 640 {
+		t.Fatalf("expected width 640, got %d", b.Dx())
+	}
+	if b.Dy() != 360 {
+		t.Fatalf("expected aspect-preserved height 360, got %d", b.Dy())
+	}
+}

machinery/src/video/livehls.go

@@ -81,8 +81,30 @@ type LiveSegmenter struct {
 	// OnInit is invoked exactly once with the encoded init segment bytes before
 	// the first media segment is emitted. Optional.
 	OnInit func(initBytes []byte) error
-	// OnSegment is invoked once per completed media segment. Optional.
+	// OnSegment is invoked once per completed media segment. Optional. It is left
+	// unused in low-latency mode (see OnPart).
 	OnSegment func(seg LiveSegment) error
+
+	// --- Low-latency (LL-HLS) partial-segment mode ---
+	//
+	// When partTargetMs > 0 the segmenter additionally slices each segment into
+	// ~partTargetMs CMAF "parts" (chunks) and emits them via OnPart the instant
+	// each one closes, instead of waiting for the whole segment. The classic
+	// per-segment OnSegment path above is left untouched (and unused) in this mode.
+	// Each part is one mp4ff fragment (moof+mdat); part 0 of a segment also carries
+	// the CMAF styp, so concatenating a segment's parts yields one valid segment.
+	partTargetMs uint64
+	// partFrag is the open part's fragment; partIndex is its 0-based index within
+	// the current segment; fragSeq is the globally monotonic moof sequence number
+	// shared across all parts (MSE wants increasing moof sequence numbers).
+	partFrag        *mp4ff.Fragment
+	partIndex       uint32
+	fragSeq         uint32
+	partSampleCount int
+	partDurationMs  uint64
+	partIndependent bool
+	// OnPart is invoked once per completed CMAF part when partTargetMs > 0.
+	OnPart func(part LivePart) error
 }
 
 // LiveSegment is one independently-decodable CMAF media segment.
@@ -97,6 +119,24 @@ type LiveSegment struct {
 	Data []byte
 }
 
+// LivePart is one CMAF partial segment (chunk) of a media segment, emitted in
+// low-latency mode the instant it closes - before the whole segment is done - so
+// the playlist can advertise it via #EXT-X-PART for near-live playback.
+type LivePart struct {
+	// SegmentSeq is the parent media segment's sequence number (the N in
+	// seg-N.K.m4s); PartIndex is K within that segment (0-based).
+	SegmentSeq uint32
+	PartIndex  uint32
+	// Independent is true when the part begins with a keyframe (its first sample is
+	// an IDR), i.e. it is independently decodable (#EXT-X-PART INDEPENDENT=YES).
+	Independent bool
+	// DurationMs is the summed sample duration of the part (for #EXT-X-PART).
+	DurationMs uint64
+	// Data of part 0 is styp+moof+mdat; later parts are bare moof+mdat, so
+	// concatenating a segment's parts in order yields one valid CMAF segment.
+	Data []byte
+}
+
 // Sample-entry flags matching the recording muxer so live and archived fragments
 // describe random access points identically.
 //
@@ -137,6 +177,16 @@ func (ls *LiveSegmenter) SetDimensions(width, height uint16) {
 	ls.height = height
 }
 
+// EnableLowLatency switches the segmenter into LL-HLS mode, additionally slicing
+// each segment into ~partTargetMs CMAF parts emitted via OnPart as they close.
+// partTargetMs is clamped to a sane floor. Call before the first WriteSample.
+func (ls *LiveSegmenter) EnableLowLatency(partTargetMs uint64) {
+	if partTargetMs < 100 {
+		partTargetMs = 100
+	}
+	ls.partTargetMs = partTargetMs
+}
+
 // InitSegment returns the encoded init segment bytes, building them on demand.
 // Useful for tests and for serving the #EXT-X-MAP target without waiting for the
 // first media segment.
@@ -238,6 +288,12 @@ func (ls *LiveSegmenter) WriteSample(isKeyframe bool, annexB []byte, ptsMs uint6
 		return fmt.Errorf("livehls: convert AnnexB: %w", err)
 	}
 
+	// Low-latency mode slices each segment into parts; the classic per-segment path
+	// below is left exactly as-is for the default (non-LL) configuration.
+	if ls.partTargetMs > 0 {
+		return ls.writeSampleLL(isKeyframe, lengthPrefixed, ptsMs, compositionOffsetMs)
+	}
+
 	// The previous sample's duration is the gap to this sample's PTS. Commit it
 	// to the (still open) current fragment before we consider rolling segments,
 	// because the pending sample always precedes this one in decode order.
@@ -350,9 +406,23 @@ func (ls *LiveSegmenter) emitSegment() error {
 	return nil
 }
 
-// Close flushes the final pending sample and emits the last open segment. Call
-// once when the live session ends so no trailing media is lost.
+// Close flushes the final pending sample and emits the last open segment (or, in
+// low-latency mode, the last open part). Call once when the live session ends so
+// no trailing media is lost.
 func (ls *LiveSegmenter) Close() error {
+	if ls.partTargetMs > 0 {
+		if ls.pending != nil {
+			dur := ls.lastDurationMs
+			if dur == 0 {
+				dur = liveFallbackDurationMs
+			}
+			ls.pending.Sample.Dur = uint32(dur)
+			if err := ls.commitPendingPart(); err != nil {
+				return err
+			}
+		}
+		return ls.closePart()
+	}
 	if ls.pending != nil {
 		dur := ls.lastDurationMs
 		if dur == 0 {
@@ -365,3 +435,152 @@ func (ls *LiveSegmenter) Close() error {
 	}
 	return ls.emitSegment()
 }
+
+// writeSampleLL is the low-latency counterpart of the per-segment staging in
+// WriteSample: it commits the previous sample into the open part, rolls the part
+// (every ~partTargetMs) and the segment (at keyframes, every ~targetSegmentMs),
+// then stages the current sample. Parts are emitted via OnPart as they close.
+func (ls *LiveSegmenter) writeSampleLL(isKeyframe bool, lengthPrefixed []byte, ptsMs uint64, compositionOffsetMs int32) error {
+	if ls.pending != nil {
+		dur := ls.lastDurationMs
+		if ptsMs > ls.pending.DecodeTime {
+			dur = ptsMs - ls.pending.DecodeTime
+		}
+		if dur == 0 {
+			dur = liveFallbackDurationMs
+		}
+		ls.lastDurationMs = dur
+		ls.pending.Sample.Dur = uint32(dur)
+		if err := ls.commitPendingPart(); err != nil {
+			return err
+		}
+	}
+
+	// Roll the segment at keyframes once enough media accumulated; otherwise roll a
+	// part once it reaches the part target. The two are mutually exclusive: a
+	// keyframe cut also closes the current part.
+	cut := false
+	if isKeyframe {
+		cut = !ls.started || (ptsMs-ls.segStartPTS) >= ls.targetSegmentMs
+	}
+	switch {
+	case cut:
+		if ls.started {
+			if err := ls.closePart(); err != nil {
+				return err
+			}
+		}
+		ls.openSegmentLL(ptsMs)
+	case ls.started && ls.partDurationMs >= ls.partTargetMs:
+		if err := ls.closePart(); err != nil {
+			return err
+		}
+		ls.openPartLL()
+	}
+
+	flags := liveNonSyncSampleFlags
+	if isKeyframe {
+		flags = liveSyncSampleFlags
+	}
+	ls.pending = &mp4ff.FullSample{
+		Sample: mp4ff.Sample{
+			Flags:                 flags,
+			Size:                  uint32(len(lengthPrefixed)),
+			CompositionTimeOffset: compositionOffsetMs,
+		},
+		DecodeTime: ptsMs,
+		Data:       lengthPrefixed,
+	}
+	return nil
+}
+
+// commitPendingPart appends the staged sample to the open part fragment, marking
+// the part independent when its first sample is a keyframe.
+func (ls *LiveSegmenter) commitPendingPart() error {
+	if ls.pending == nil {
+		return nil
+	}
+	if ls.partFrag == nil {
+		// No open part yet (pending staged before the first keyframe cut). The cut
+		// path always opens a part before staging, so this only guards against logic
+		// drift; drop rather than panic.
+		ls.pending = nil
+		return nil
+	}
+	first := ls.partSampleCount == 0
+	if err := ls.partFrag.AddFullSampleToTrack(*ls.pending, ls.videoTrackID); err != nil {
+		return fmt.Errorf("livehls: AddFullSampleToTrack: %w", err)
+	}
+	if first && ls.pending.Sample.Flags == liveSyncSampleFlags {
+		ls.partIndependent = true
+	}
+	ls.partSampleCount++
+	ls.partDurationMs += uint64(ls.pending.Sample.Dur)
+	ls.segDurationMs += uint64(ls.pending.Sample.Dur)
+	ls.pending = nil
+	return nil
+}
+
+// openSegmentLL starts a fresh media segment at a keyframe by opening its part 0.
+func (ls *LiveSegmenter) openSegmentLL(startPTS uint64) {
+	ls.seqNr++
+	ls.partIndex = 0
+	ls.segStartPTS = startPTS
+	ls.segDurationMs = 0
+	ls.started = true
+	ls.openPartFragment()
+}
+
+// openPartLL starts the next part within the current segment.
+func (ls *LiveSegmenter) openPartLL() {
+	ls.partIndex++
+	ls.openPartFragment()
+}
+
+// openPartFragment allocates a fresh single-track fragment (one moof+mdat) for
+// the next part, with a globally monotonic moof sequence number.
+func (ls *LiveSegmenter) openPartFragment() {
+	ls.fragSeq++
+	frag, err := mp4ff.CreateFragment(ls.fragSeq, ls.videoTrackID)
+	if err != nil {
+		log.Log.Error("LiveSegmenter.openPartFragment(): CreateFragment failed: " + err.Error())
+		return
+	}
+	ls.partFrag = frag
+	ls.partSampleCount = 0
+	ls.partDurationMs = 0
+	ls.partIndependent = false
+}
+
+// closePart encodes the open part and hands it to OnPart. Part 0 of a segment
+// carries the CMAF styp; later parts are bare moof+mdat, so a segment's parts
+// concatenate into one valid segment. Empty parts are skipped.
+func (ls *LiveSegmenter) closePart() error {
+	if ls.partFrag == nil || ls.partSampleCount == 0 {
+		return nil
+	}
+	var buf bytes.Buffer
+	if ls.partIndex == 0 {
+		seg := mp4ff.NewMediaSegment() // includes a CMAF styp box by default
+		seg.AddFragment(ls.partFrag)
+		if err := seg.Encode(&buf); err != nil {
+			return fmt.Errorf("livehls: encode part %d.%d: %w", ls.seqNr, ls.partIndex, err)
+		}
+	} else {
+		if err := ls.partFrag.Encode(&buf); err != nil {
+			return fmt.Errorf("livehls: encode part %d.%d: %w", ls.seqNr, ls.partIndex, err)
+		}
+	}
+	out := LivePart{
+		SegmentSeq:  ls.seqNr,
+		PartIndex:   ls.partIndex,
+		Independent: ls.partIndependent,
+		DurationMs:  ls.partDurationMs,
+		Data:        buf.Bytes(),
+	}
+	ls.partFrag = nil
+	if ls.OnPart != nil {
+		return ls.OnPart(out)
+	}
+	return nil
+}

machinery/src/video/livehls_test.go

@@ -369,3 +369,186 @@ func TestLiveSegmenterWritesHLSBundle(t *testing.T) {
 	t.Logf("wrote HLS bundle to %s (%d segments)\n%s", outDir, len(segments), playlist)
 }
 
+// boxTypeAt returns the 4CC box type at the front of a top-level box blob (the
+// 4 bytes following the 32-bit size), or "" if the blob is too short.
+func boxTypeAt(b []byte) string {
+	if len(b) < 8 {
+		return ""
+	}
+	return string(b[4:8])
+}
+
+// TestLiveSegmenterLowLatencyParts runs the segmenter in LL-HLS mode over the
+// same synthetic stream and asserts that:
+//   - each ~2s segment is sliced into multiple CMAF parts (more parts than
+//     segments overall);
+//   - part 0 of every segment carries the CMAF styp and is INDEPENDENT (begins
+//     with the segment keyframe); later parts are bare moof+mdat (no styp);
+//   - moof sequence numbers are globally monotonic across all parts (MSE needs
+//     increasing moof sequence numbers);
+//   - concatenating a segment's parts in order yields exactly the same bytes the
+//     classic per-segment path would emit, decoding into one independent CMAF
+//     segment whose first sample is a sync sample with the expected tfdt;
+//   - every sample and keyframe of the input is preserved end to end.
+func TestLiveSegmenterLowLatencyParts(t *testing.T) {
+	const (
+		frameDurMs = uint64(40) // 25 fps
+		gopFrames  = 25         // keyframe every 1000 ms
+		numGOPs    = 6
+		numFrames  = gopFrames * numGOPs // 150 frames, 6000 ms
+		targetMs   = uint64(2000)        // 2s segments => 2 GOPs each
+		partMs     = uint64(300)         // ~300 ms parts => ~6-7 parts/segment
+	)
+
+	seg := NewLiveSegmenter("H264", [][]byte{liveTestSPS}, [][]byte{liveTestPPS}, nil, targetMs)
+	seg.SetDimensions(640, 480)
+	seg.EnableLowLatency(partMs)
+
+	var initBytes []byte
+	var initCalls int
+	var parts []LivePart
+	seg.OnInit = func(b []byte) error {
+		initCalls++
+		initBytes = append([]byte(nil), b...)
+		return nil
+	}
+	seg.OnPart = func(p LivePart) error {
+		parts = append(parts, p)
+		return nil
+	}
+
+	for i := 0; i < numFrames; i++ {
+		isKey := i%gopFrames == 0
+		if err := seg.WriteSample(isKey, makeAnnexBFrame(isKey), uint64(i)*frameDurMs, 0); err != nil {
+			t.Fatalf("WriteSample(frame=%d): %v", i, err)
+		}
+	}
+	if err := seg.Close(); err != nil {
+		t.Fatalf("Close: %v", err)
+	}
+
+	if initCalls != 1 {
+		t.Fatalf("OnInit called %d times, want 1", initCalls)
+	}
+	if len(parts) == 0 {
+		t.Fatal("no parts produced in low-latency mode")
+	}
+
+	// --- Parts are globally moof-monotonic, and group into 3 segments whose part
+	// indices are contiguous from 0. ---
+	bySeg := map[uint32][]LivePart{}
+	var order []uint32
+	var lastMoof uint32
+	for i, p := range parts {
+		if _, seen := bySeg[p.SegmentSeq]; !seen {
+			order = append(order, p.SegmentSeq)
+		}
+		bySeg[p.SegmentSeq] = append(bySeg[p.SegmentSeq], p)
+
+		// Decode the part to read its moof sequence number and confirm the styp
+		// convention (part 0 => styp present, later parts => bare moof+mdat).
+		front := boxTypeAt(p.Data)
+		if p.PartIndex == 0 {
+			if front != "styp" {
+				t.Errorf("seg %d part 0: leading box=%q, want styp", p.SegmentSeq, front)
+			}
+			if !p.Independent {
+				t.Errorf("seg %d part 0: Independent=false, want true (starts on keyframe)", p.SegmentSeq)
+			}
+		} else if front != "moof" {
+			t.Errorf("seg %d part %d: leading box=%q, want moof (no styp on later parts)", p.SegmentSeq, p.PartIndex, front)
+		}
+
+		parsed, err := mp4ff.DecodeFile(bytes.NewReader(p.Data))
+		if err != nil {
+			t.Fatalf("seg %d part %d: decode: %v", p.SegmentSeq, p.PartIndex, err)
+		}
+		if len(parsed.Segments) != 1 || len(parsed.Segments[0].Fragments) != 1 {
+			t.Fatalf("seg %d part %d: want exactly one fragment", p.SegmentSeq, p.PartIndex)
+		}
+		moof := parsed.Segments[0].Fragments[0].Moof.Mfhd.SequenceNumber
+		if i > 0 && moof <= lastMoof {
+			t.Errorf("part %d: moof sequence=%d not greater than previous %d", i, moof, lastMoof)
+		}
+		lastMoof = moof
+	}
+
+	if len(order) != 3 {
+		t.Fatalf("got %d segments, want 3", len(order))
+	}
+	if len(parts) <= len(order) {
+		t.Fatalf("got %d parts for %d segments, expected each segment to be sliced into multiple parts", len(parts), len(order))
+	}
+	for _, segSeq := range order {
+		for idx, p := range bySeg[segSeq] {
+			if p.PartIndex != uint32(idx) {
+				t.Errorf("seg %d: part index %d out of order (want %d)", segSeq, p.PartIndex, idx)
+			}
+		}
+	}
+
+	// --- Concatenating a segment's parts must reconstruct one independent CMAF
+	// segment that decodes against the init segment. ---
+	wantTFDT := map[uint32]uint64{1: 0, 2: 2000, 3: 4000}
+	var totalSamples, totalSync int
+	for _, segSeq := range order {
+		segParts := bySeg[segSeq]
+		var full []byte
+		var wantPartDur uint64
+		for _, p := range segParts {
+			full = append(full, p.Data...)
+			wantPartDur += p.DurationMs
+		}
+		standalone := append(append([]byte(nil), initBytes...), full...)
+		parsed, err := mp4ff.DecodeFile(bytes.NewReader(standalone))
+		if err != nil {
+			t.Fatalf("seg %d: decode concatenated parts: %v", segSeq, err)
+		}
+		if len(parsed.Segments) != 1 {
+			t.Fatalf("seg %d: parsed %d media segments, want 1", segSeq, len(parsed.Segments))
+		}
+		mseg := parsed.Segments[0]
+		if mseg.Styp == nil {
+			t.Errorf("seg %d: reconstructed segment missing CMAF styp", segSeq)
+		}
+		if len(mseg.Fragments) != len(segParts) {
+			t.Errorf("seg %d: %d fragments, want %d (one per part)", segSeq, len(mseg.Fragments), len(segParts))
+		}
+		firstTraf := mseg.Fragments[0].Moof.Traf
+		if got := firstTraf.Tfdt.BaseMediaDecodeTime(); got != wantTFDT[segSeq] {
+			t.Errorf("seg %d: first fragment tfdt=%d, want %d", segSeq, got, wantTFDT[segSeq])
+		}
+		var segDur uint64
+		var firstSample mp4ff.Sample
+		var haveFirst bool
+		for _, fr := range mseg.Fragments {
+			for _, trun := range fr.Moof.Traf.Truns {
+				for _, smp := range trun.Samples {
+					if !haveFirst {
+						firstSample = smp
+						haveFirst = true
+					}
+					totalSamples++
+					if isSyncSample(smp) {
+						totalSync++
+					}
+					segDur += uint64(smp.Dur)
+				}
+			}
+		}
+		if !isSyncSample(firstSample) {
+			t.Errorf("seg %d: first sample is not a sync sample", segSeq)
+		}
+		if segDur != wantPartDur {
+			t.Errorf("seg %d: summed sample dur=%d, summed part dur=%d", segSeq, segDur, wantPartDur)
+		}
+	}
+
+	if totalSamples != numFrames {
+		t.Errorf("total samples across parts=%d, want %d", totalSamples, numFrames)
+	}
+	if totalSync != numGOPs {
+		t.Errorf("total sync samples=%d, want %d (one per GOP)", totalSync, numGOPs)
+	}
+}
+