Merge pull request #299 from kerberos-io/feature/remote-recording

feature/remote-recording

.github/workflows/release-bump.yml

@@ -29,21 +29,21 @@ jobs:
 
   # Publish the platform image to the uug-ai GitHub Container Registry
   # (ghcr.io/uug-ai/agent-platform).
-  release:
-    needs: bump-release
-    uses: uug-ai/workflows/.github/workflows/release-create.yml@main
-    with:
-      organization: uug-ai
-      project: ${{ github.event.repository.name }}
-      tag: ${{ needs.bump-release.outputs.tag }}
-      docker_context: "."
-      create_gitops_pr: false
-      runner_matrix: >-
-        [
-          {"architecture":"amd64","runner":"ubuntu-24.04"},
-          {"architecture":"arm64","runner":"ubuntu-24.04-arm"}
-        ]
-    secrets: inherit
+  #release:
+  #  needs: bump-release
+  #  uses: uug-ai/workflows/.github/workflows/release-create.yml@main
+  #  with:
+  #    organization: uug-ai
+  #    project: ${{ github.event.repository.name }}
+  #    tag: ${{ needs.bump-release.outputs.tag }}
+  #    docker_context: "."
+  #    create_gitops_pr: false
+  #    runner_matrix: >-
+  #       [
+  #         {"architecture":"amd64","runner":"ubuntu-24.04"},
+  #         {"architecture":"arm64","runner":"ubuntu-24.04-arm"}
+  #      ]
+  #  secrets: inherit
 
   # Everything below mirrors the agent's own release-create.yml pipeline and
   # publishes the multi-arch image to the kerberos/agent Docker Hub repo, driven

README.md

@@ -203,7 +203,8 @@ Next to attaching the configuration file, it is also possible to override the co
 | `AGENT_REMOVE_AFTER_UPLOAD`                 | When enabled, recordings uploaded successfully to a storage will be removed from disk.          | "true"                         |
 | `AGENT_OFFLINE`                             | Makes sure no external connection is made.                                                      | "false"                        |
 | `AGENT_AUTO_CLEAN`                          | Cleans up the recordings directory.                                                             | "true"                         |
-| `AGENT_AUTO_CLEAN_MAX_SIZE`                 | If `AUTO_CLEAN` enabled, set the max size of the recordings directory (in MB).                  | "100"                          |
+| `AGENT_AUTO_CLEAN_MAX_SIZE`                 | If `AUTO_CLEAN` enabled, cap the recordings directory at this size (in MB). When unset/0, recordings may use the whole disk instead (see `AGENT_AUTO_CLEAN_MIN_FREE_SPACE`). | "100"                          |
+| `AGENT_AUTO_CLEAN_MIN_FREE_SPACE`           | When `AUTO_CLEAN` is enabled and no `MAX_SIZE` is set, keep at least this much free space (in MB) on the recordings disk before deleting the oldest (already-uploaded first) recordings. Defaults to 5% of the disk. | ""                             |
 | `AGENT_TIME`                                | Enable the timetable for Kerberos Agent                                                         | "false"                        |
 | `AGENT_TIMETABLE`                           | A (weekly) time table to specify when to make recordings "start1,end1,start2,end2;start1..      | ""                             |
 | `AGENT_REGION_POLYGON`                      | A single polygon set for motion detection: "x1,y1;x2,y2;x3,y3;...                               | ""                             |

machinery/data/config/config.json

@@ -7,7 +7,7 @@
 	"offline": "false",
 	"auto_clean": "true",
 	"remove_after_upload": "true",
-	"max_directory_size": 100,
+	"max_directory_size": 0,
 	"timezone": "Africa/Ceuta",
 	"capture": {
 		"name": "",

machinery/go.mod

@@ -24,7 +24,7 @@ require (
 	github.com/gorilla/websocket v1.5.3
 	github.com/kellydunn/golang-geo v0.7.0
 	github.com/kerberos-io/joy4 v1.0.64
-	github.com/kerberos-io/onvif v1.0.0
+	github.com/kerberos-io/onvif v1.2.1
 	github.com/minio/minio-go/v6 v6.0.57
 	github.com/nfnt/resize v0.0.0-20180221191011-83c6a9932646
 	github.com/op/go-logging v0.0.0-20160315200505-970db520ece7
@@ -76,7 +76,6 @@ require (
 	github.com/golang/snappy v0.0.4 // indirect
 	github.com/google/uuid v1.6.0 // indirect
 	github.com/grpc-ecosystem/grpc-gateway/v2 v2.26.3 // indirect
-	github.com/icholy/digest v0.1.23 // indirect
 	github.com/josharian/intern v1.0.0 // indirect
 	github.com/json-iterator/go v1.1.12 // indirect
 	github.com/juju/errors v1.0.0 // indirect

machinery/go.sum

@@ -755,8 +755,6 @@ github.com/iancoleman/strcase v0.3.0/go.mod h1:iwCmte+B7n89clKwxIoIXy/HfoL7AsD47
 github.com/ianlancetaylor/demangle v0.0.0-20181102032728-5e5cf60278f6/go.mod h1:aSSvb/t6k1mPoxDqO4vJh6VOCGPwU4O0C2/Eqndh1Sc=
 github.com/ianlancetaylor/demangle v0.0.0-20200824232613-28f6c0f3b639/go.mod h1:aSSvb/t6k1mPoxDqO4vJh6VOCGPwU4O0C2/Eqndh1Sc=
 github.com/ianlancetaylor/demangle v0.0.0-20220319035150-800ac71e25c2/go.mod h1:aYm2/VgdVmcIU8iMfdMvDMsRAQjcfZSKFby6HOFvi/w=
-github.com/icholy/digest v0.1.23 h1:4hX2pIloP0aDx7RJW0JewhPPy3R8kU+vWKdxPsCCGtY=
-github.com/icholy/digest v0.1.23/go.mod h1:QNrsSGQ5v7v9cReDI0+eyjsXGUoRSUZQHeQ5C4XLa0Y=
 github.com/jessevdk/go-flags v1.4.0/go.mod h1:4FA24M0QyGHXBuZZK/XkWh8h0e1EYbRYJSGM75WSRxI=
 github.com/josharian/intern v1.0.0 h1:vlS4z54oSdjm0bgjRigI+G1HpF+tI+9rE5LLzOg8HmY=
 github.com/josharian/intern v1.0.0/go.mod h1:5DoeVV0s6jJacbCEi61lwdGj/aVlrQvzHFFd8Hwg//Y=
@@ -776,8 +774,8 @@ github.com/kellydunn/golang-geo v0.7.0 h1:A5j0/BvNgGwY6Yb6inXQxzYwlPHc6WVZR+Mrar
 github.com/kellydunn/golang-geo v0.7.0/go.mod h1:YYlQPJ+DPEzrHx8kT3oPHC/NjyvCCXE+IuKGKdrjrcU=
 github.com/kerberos-io/joy4 v1.0.64 h1:gTUSotHSOhp9mNqEecgq88tQHvpj7TjmrvPUsPm0idg=
 github.com/kerberos-io/joy4 v1.0.64/go.mod h1:nZp4AjvKvTOXRrmDyAIOw+Da+JA5OcSo/JundGfOlFU=
-github.com/kerberos-io/onvif v1.0.0 h1:pLJrK6skPkK+5Bj4XfqHUkQ2I+p5pwELnp+kQTJWXiQ=
-github.com/kerberos-io/onvif v1.0.0/go.mod h1:P1kUcCfeotJSlL1jwGseH6NSnCwWiuJLl3gAzafnLbA=
+github.com/kerberos-io/onvif v1.2.1 h1:+vxyHPylt0ufK8gv7FL+KzhJUeComMGrTmP5KxT2YEc=
+github.com/kerberos-io/onvif v1.2.1/go.mod h1:XSgEQXmEDjUQTbdXvsaRJt6Az8YPGj7L+j5iXKEGijU=
 github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8=
 github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
 github.com/klauspost/asmfmt v1.3.2/go.mod h1:AG8TuvYojzulgDAMCnYn50l/5QV3Bs/tp6j0HLHbNSE=
@@ -2109,7 +2107,6 @@ gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C
 gopkg.in/yaml.v3 v3.0.0-20200615113413-eeeca48fe776/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
 gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
-gotest.tools/v3 v3.5.1 h1:EENdUnS3pdur5nybKYIh2Vfgc8IUNBjxDPSjtiJcOzU=
 gotest.tools/v3 v3.5.1/go.mod h1:isy3WKz7GK6uNw/sbHzfKBLvlvXwUyV06n6brMxxopU=
 honnef.co/go/tools v0.0.0-20190102054323-c2f93a96b099/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
 honnef.co/go/tools v0.0.0-20190106161140-3f1c8253044a/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=

machinery/src/capture/cleanup_test.go

@@ -0,0 +1,219 @@
+package capture
+
+import (
+	"os"
+	"path/filepath"
+	"runtime"
+	"testing"
+	"time"
+
+	"github.com/kerberos-io/agent/machinery/src/models"
+)
+
+// writeRecording creates a file under recordingsDir and sets its modtime so the
+// tests can control the "oldest" ordering deterministically.
+func writeRecording(t *testing.T, recordingsDir, name string, ageMinutes int) {
+	t.Helper()
+	full := filepath.Join(recordingsDir, name)
+	if err := os.WriteFile(full, []byte("data"), 0o644); err != nil {
+		t.Fatalf("write recording %s: %v", name, err)
+	}
+	mod := time.Now().Add(-time.Duration(ageMinutes) * time.Minute)
+	if err := os.Chtimes(full, mod, mod); err != nil {
+		t.Fatalf("chtimes %s: %v", name, err)
+	}
+}
+
+// markPending creates the upload marker in cloudDir for the given recording,
+// marking it as still queued for upload.
+func markPending(t *testing.T, cloudDir, name string) {
+	t.Helper()
+	if err := os.WriteFile(filepath.Join(cloudDir, name), nil, 0o644); err != nil {
+		t.Fatalf("write marker %s: %v", name, err)
+	}
+}
+
+func newCleanupDirs(t *testing.T) (string, string) {
+	t.Helper()
+	base := t.TempDir()
+	recordingsDir := filepath.Join(base, "data", "recordings")
+	cloudDir := filepath.Join(base, "data", "cloud")
+	if err := os.MkdirAll(recordingsDir, 0o755); err != nil {
+		t.Fatalf("mkdir recordings: %v", err)
+	}
+	if err := os.MkdirAll(cloudDir, 0o755); err != nil {
+		t.Fatalf("mkdir cloud: %v", err)
+	}
+	return recordingsDir, cloudDir
+}
+
+// The core regression: when the oldest recording is still pending upload but a
+// newer one has already been uploaded, cleanup must delete the uploaded (safe)
+// one and leave the pending recording on disk so it can still be uploaded.
+func TestPickRecordingToCleanup_PrefersUploaded(t *testing.T) {
+	recordingsDir, cloudDir := newCleanupDirs(t)
+
+	// oldest is still pending upload (marker present).
+	writeRecording(t, recordingsDir, "oldest_pending.mp4", 30)
+	markPending(t, cloudDir, "oldest_pending.mp4")
+	// newer one has already been uploaded (no marker).
+	writeRecording(t, recordingsDir, "newer_uploaded.mp4", 10)
+
+	name, pending, err := pickRecordingToCleanup(recordingsDir, cloudDir)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if pending {
+		t.Fatalf("expected a safe (already-uploaded) deletion, got pending=true")
+	}
+	if name != "newer_uploaded.mp4" {
+		t.Fatalf("cleanup picked %q, want the uploaded recording newer_uploaded.mp4", name)
+	}
+}
+
+// Among several already-uploaded recordings, the oldest uploaded one is chosen.
+func TestPickRecordingToCleanup_OldestUploadedFirst(t *testing.T) {
+	recordingsDir, cloudDir := newCleanupDirs(t)
+
+	writeRecording(t, recordingsDir, "old_uploaded.mp4", 40)
+	writeRecording(t, recordingsDir, "mid_uploaded.mp4", 20)
+	// pending one must be ignored even though it is not the oldest.
+	writeRecording(t, recordingsDir, "pending.mp4", 30)
+	markPending(t, cloudDir, "pending.mp4")
+
+	name, pending, err := pickRecordingToCleanup(recordingsDir, cloudDir)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if pending {
+		t.Fatalf("expected pending=false, got true")
+	}
+	if name != "old_uploaded.mp4" {
+		t.Fatalf("cleanup picked %q, want old_uploaded.mp4", name)
+	}
+}
+
+// Last resort: when every recording is still pending upload, cleanup returns the
+// oldest one with pending=true so the caller can drop it (and its marker) to keep
+// the disk bounded.
+func TestPickRecordingToCleanup_AllPendingFallsBackToOldest(t *testing.T) {
+	recordingsDir, cloudDir := newCleanupDirs(t)
+
+	writeRecording(t, recordingsDir, "a_old.mp4", 50)
+	markPending(t, cloudDir, "a_old.mp4")
+	writeRecording(t, recordingsDir, "b_new.mp4", 5)
+	markPending(t, cloudDir, "b_new.mp4")
+
+	name, pending, err := pickRecordingToCleanup(recordingsDir, cloudDir)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if !pending {
+		t.Fatalf("expected pending=true when every recording is queued for upload")
+	}
+	if name != "a_old.mp4" {
+		t.Fatalf("cleanup picked %q, want the oldest pending a_old.mp4", name)
+	}
+}
+
+// An empty recordings directory yields os.ErrNotExist so the caller does nothing.
+func TestPickRecordingToCleanup_Empty(t *testing.T) {
+	recordingsDir, cloudDir := newCleanupDirs(t)
+
+	if _, _, err := pickRecordingToCleanup(recordingsDir, cloudDir); err != os.ErrNotExist {
+		t.Fatalf("expected os.ErrNotExist for an empty directory, got %v", err)
+	}
+}
+
+// writeSizedRecording writes a recording of an exact byte size so tests can
+// exercise the megabyte-based directory-cap threshold.
+func writeSizedRecording(t *testing.T, dir, name string, size int) {
+	t.Helper()
+	if err := os.WriteFile(filepath.Join(dir, name), make([]byte, size), 0o644); err != nil {
+		t.Fatalf("write sized recording %s: %v", name, err)
+	}
+}
+
+// When AGENT_AUTO_CLEAN_MAX_SIZE (MaxDirectorySize) is set, cleanup triggers once
+// the recordings directory grows past that many megabytes.
+func TestRecordingsNeedCleanup_FixedCap(t *testing.T) {
+	recordingsDir, _ := newCleanupDirs(t)
+	// ~2 MB of recordings on disk.
+	writeSizedRecording(t, recordingsDir, "big.mp4", 2*1000*1000)
+
+	over := &models.Configuration{Config: models.Config{MaxDirectorySize: 1}}
+	need, err := recordingsNeedCleanup(recordingsDir, over)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if !need {
+		t.Fatalf("expected cleanup when 2MB of recordings exceed the 1MB cap")
+	}
+
+	under := &models.Configuration{Config: models.Config{MaxDirectorySize: 100}}
+	need, err = recordingsNeedCleanup(recordingsDir, under)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if need {
+		t.Fatalf("expected no cleanup when 2MB of recordings stay under the 100MB cap")
+	}
+}
+
+// With no fixed cap (the default), cleanup is driven by the free space left on
+// the recordings filesystem versus the reserve.
+func TestRecordingsNeedCleanup_DefaultDiskReserve(t *testing.T) {
+	if runtime.GOOS != "linux" {
+		t.Skip("disk usage stats are only implemented on linux")
+	}
+	recordingsDir, _ := newCleanupDirs(t)
+
+	totalMB, availableMB, err := diskUsageMB(recordingsDir)
+	if err != nil {
+		t.Fatalf("diskUsageMB: %v", err)
+	}
+	if totalMB <= 0 || availableMB <= 0 {
+		t.Skipf("unexpected disk stats total=%dMB available=%dMB", totalMB, availableMB)
+	}
+
+	// A reserve larger than the whole disk means free space is always below it.
+	over := &models.Configuration{Config: models.Config{MinFreeSpace: totalMB + availableMB}}
+	need, err := recordingsNeedCleanup(recordingsDir, over)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if !need {
+		t.Fatalf("expected cleanup when free space (%dMB) is below the reserve", availableMB)
+	}
+
+	// A 1 MB reserve leaves plenty of free space, so nothing should be cleaned.
+	under := &models.Configuration{Config: models.Config{MinFreeSpace: 1}}
+	need, err = recordingsNeedCleanup(recordingsDir, under)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if need {
+		t.Fatalf("expected no cleanup when free space (%dMB) exceeds the 1MB reserve", availableMB)
+	}
+}
+
+// The default 5% reserve must never truncate to 0MB on small disks, otherwise
+// cleanup would only trigger once the disk is completely full.
+func TestDefaultReserveMB(t *testing.T) {
+	cases := []struct {
+		totalMB int64
+		want    int64
+	}{
+		{totalMB: 0, want: 1},   // no/unknown disk size still reserves 1MB
+		{totalMB: 10, want: 1},  // 5% = 0MB -> floored to 1MB
+		{totalMB: 19, want: 1},  // 5% = 0MB -> floored to 1MB
+		{totalMB: 20, want: 1},  // 5% = exactly 1MB
+		{totalMB: 100, want: 5}, // 5% = 5MB
+		{totalMB: 1000, want: 50},
+	}
+	for _, c := range cases {
+		if got := defaultReserveMB(c.totalMB); got != c.want {
+			t.Errorf("defaultReserveMB(%d) = %d, want %d", c.totalMB, got, c.want)
+		}
+	}
+}

machinery/src/capture/disk_linux.go

@@ -0,0 +1,23 @@
+//go:build linux
+
+package capture
+
+import "syscall"
+
+// diskUsageMB returns the total capacity and the currently available space (both
+// in megabytes, decimal) of the filesystem that contains path. Auto-clean uses
+// it to default its cleanup threshold to the real disk capacity instead of a
+// fixed size, so recordings can grow to fill the disk while keeping a reserve
+// free. Linux is the agent's deployment target (amd64/arm64 containers).
+func diskUsageMB(path string) (totalMB int64, availableMB int64, err error) {
+	var stat syscall.Statfs_t
+	if err = syscall.Statfs(path, &stat); err != nil {
+		return 0, 0, err
+	}
+	blockSize := int64(stat.Bsize)
+	totalMB = int64(stat.Blocks) * blockSize / 1000 / 1000
+	// Bavail is the free space available to unprivileged users, which is the
+	// space we can actually keep writing recordings into.
+	availableMB = int64(stat.Bavail) * blockSize / 1000 / 1000
+	return totalMB, availableMB, nil
+}

machinery/src/capture/disk_other.go

@@ -0,0 +1,13 @@
+//go:build !linux
+
+package capture
+
+import "errors"
+
+// diskUsageMB is only implemented on Linux (the agent's deployment target). On
+// other platforms (e.g. local macOS/Windows dev builds) auto-clean falls back to
+// its historical fixed-size directory cap, so this reports the capability as
+// unavailable.
+func diskUsageMB(path string) (totalMB int64, availableMB int64, err error) {
+	return 0, 0, errors.New("disk usage stats are not supported on this platform")
+}

machinery/src/capture/main.go

@@ -9,6 +9,7 @@ import (
 	"strconv"
 	"time"
 
+	mqtt "github.com/eclipse/paho.mqtt.golang"
 	"github.com/gin-gonic/gin"
 	"github.com/kerberos-io/agent/machinery/src/conditions"
 	"github.com/kerberos-io/agent/machinery/src/encryption"
@@ -20,45 +21,213 @@ import (
 	"go.opentelemetry.io/otel/trace"
 )
 
-func CleanupRecordingDirectory(configDirectory string, configuration *models.Configuration) {
-	autoClean := configuration.Config.AutoClean
-	if autoClean == "true" {
-		maxSize := configuration.Config.MaxDirectorySize
-		if maxSize == 0 {
-			maxSize = 300
-		}
-		// Total size of the recording directory.
-		recordingsDirectory := configDirectory + "/data/recordings"
-		size, err := utils.DirSize(recordingsDirectory)
-		if err == nil {
-			sizeInMB := size / 1000 / 1000
-			if sizeInMB >= maxSize {
-				// Remove the oldest recording
-				oldestFile, err := utils.FindOldestFile(recordingsDirectory)
-				if err == nil {
-					err := os.Remove(recordingsDirectory + "/" + oldestFile.Name())
-					log.Log.Info("HandleRecordStream: removed oldest file as part of cleanup - " + recordingsDirectory + "/" + oldestFile.Name())
-					if err != nil {
-						log.Log.Info("HandleRecordStream: something went wrong, " + err.Error())
-					}
-				} else {
-					log.Log.Info("HandleRecordStream: something went wrong, " + err.Error())
-				}
-			}
-		} else {
-			log.Log.Info("HandleRecordStream: something went wrong, " + err.Error())
-		}
-
+// publishRecordingState notifies the hub (and ultimately the live-view UI) that
+// this camera started ("recording": true) or stopped ("recording": false)
+// recording, so the frontend can show a "recording" indicator while the agent
+// is recording (e.g. a motion clip triggered manually from the live view or by
+// motion detection). It is a best-effort broadcast: when no hub/MQTT is
+// configured (or the agent is offline) it is a no-op, and a missed message is
+// self-healed by the frontend's safety timeout.
+func publishRecordingState(mqttClient mqtt.Client, hubKey string, configuration *models.Configuration, recording bool) {
+	if mqttClient == nil || hubKey == "" || configuration.Config.Offline == "true" {
+		return
+	}
+	message := models.Message{
+		Payload: models.Payload{
+			Action:   "recording",
+			DeviceId: configuration.Config.Key,
+			Value: map[string]interface{}{
+				"timestamp": time.Now().Unix(),
+				"recording": recording,
+			},
+		},
+	}
+	payload, err := models.PackageMQTTMessage(configuration, message)
+	if err == nil {
+		mqttClient.Publish("kerberos/hub/"+hubKey, 2, false, payload)
 	} else {
-		log.Log.Info("HandleRecordStream: Autoclean disabled, nothing to do here.")
+		log.Log.Error("capture.main.publishRecordingState(): failed to package MQTT message: " + err.Error())
 	}
 }
 
-func HandleRecordStream(queue *packets.Queue, configDirectory string, configuration *models.Configuration, communication *models.Communication, rtspClient RTSPClient) {
+func CleanupRecordingDirectory(configDirectory string, configuration *models.Configuration) {
+	autoClean := configuration.Config.AutoClean
+	if autoClean != "true" {
+		log.Log.Info("HandleRecordStream: Autoclean disabled, nothing to do here.")
+		return
+	}
+
+	recordingsDirectory := configDirectory + "/data/recordings"
+	cloudDirectory := configDirectory + "/data/cloud"
+
+	// Decide whether we still need to free up space. See recordingsNeedCleanup
+	// for the two modes: an explicit fixed directory cap
+	// (AGENT_AUTO_CLEAN_MAX_SIZE) or, by default, letting recordings use the whole
+	// disk while keeping a free-space reserve.
+	needsCleanup, err := recordingsNeedCleanup(recordingsDirectory, configuration)
+	if err != nil {
+		log.Log.Info("HandleRecordStream: something went wrong, " + err.Error())
+		return
+	}
+	if !needsCleanup {
+		return
+	}
+
+	// Remove the oldest recording, but PREFER recordings that have already been
+	// uploaded (i.e. no longer have a pending marker in data/cloud). This stops
+	// auto-clean from deleting recordings that are still queued for upload. That
+	// previously caused silent data loss: during a network outage the upload
+	// backlog grows, cleanup deletes the oldest (still un-uploaded) recording to
+	// stay under MaxDirectorySize, and when connectivity returns the upload loop
+	// finds the marker but the file is gone -> the recording is dropped and never
+	// reaches the vault.
+	//
+	// Only when EVERY recording on disk is still pending upload do we fall back to
+	// deleting the oldest pending one, as a last resort to keep the disk bounded
+	// (otherwise a long outage would fill the disk and stop new recordings).
+	name, pending, err := pickRecordingToCleanup(recordingsDirectory, cloudDirectory)
+	if err != nil {
+		log.Log.Info("HandleRecordStream: something went wrong, " + err.Error())
+		return
+	}
+
+	if err := os.Remove(recordingsDirectory + "/" + name); err != nil {
+		log.Log.Info("HandleRecordStream: something went wrong, " + err.Error())
+		return
+	}
+
+	if pending {
+		// Data-loss event: the whole recordings directory is an un-uploaded
+		// backlog (e.g. a prolonged network outage), so we had to drop a recording
+		// that was never uploaded to keep recording new footage. Also remove the
+		// now-dangling upload marker so the upload loop doesn't keep trying to
+		// upload a file that no longer exists.
+		log.Log.Warning("HandleRecordStream: removed oldest recording as part of cleanup, but it was STILL PENDING UPLOAD (disk full of un-uploaded recordings) - " + recordingsDirectory + "/" + name)
+		if err := os.Remove(cloudDirectory + "/" + name); err != nil && !os.IsNotExist(err) {
+			log.Log.Info("HandleRecordStream: could not remove dangling upload marker " + name + ", " + err.Error())
+		}
+	} else {
+		log.Log.Info("HandleRecordStream: removed oldest file as part of cleanup - " + recordingsDirectory + "/" + name)
+	}
+}
+
+// recordingsNeedCleanup reports whether auto-clean should free up space in the
+// recordings directory. There are two modes:
+//
+//   - AGENT_AUTO_CLEAN_MAX_SIZE (MaxDirectorySize, MB) set: cap the size of the
+//     recordings directory itself (the historical behaviour).
+//   - MaxDirectorySize == 0 (the default): recordings may use the WHOLE disk.
+//     Cleanup only triggers once the free space on the recordings filesystem
+//     drops to/below a reserve. The reserve is AGENT_AUTO_CLEAN_MIN_FREE_SPACE
+//     (MinFreeSpace, MB) when set, otherwise 5% of the disk's total capacity.
+//
+// If disk stats can't be read (e.g. non-Linux dev builds) it falls back to the
+// historical fixed 300 MB directory cap so behaviour stays bounded.
+func recordingsNeedCleanup(recordingsDirectory string, configuration *models.Configuration) (bool, error) {
+	maxSize := configuration.Config.MaxDirectorySize
+
+	// Explicit fixed cap on the recordings directory size.
+	if maxSize > 0 {
+		size, err := utils.DirSize(recordingsDirectory)
+		if err != nil {
+			return false, err
+		}
+		return size/1000/1000 >= maxSize, nil
+	}
+
+	// Default: allow recordings to use the full disk, keeping a reserve free.
+	totalMB, availableMB, err := diskUsageMB(recordingsDirectory)
+	if err != nil {
+		// Disk stats unavailable: fall back to the historical 300 MB cap.
+		size, derr := utils.DirSize(recordingsDirectory)
+		if derr != nil {
+			return false, derr
+		}
+		return size/1000/1000 >= 300, nil
+	}
+
+	reserveMB := configuration.Config.MinFreeSpace
+	if reserveMB <= 0 {
+		reserveMB = defaultReserveMB(totalMB)
+	}
+
+	return availableMB <= reserveMB, nil
+}
+
+// defaultReserveMB returns the free-space reserve (MB) to keep on the recordings
+// disk when AGENT_AUTO_CLEAN_MIN_FREE_SPACE is not set: 5% of the disk total,
+// but never below 1MB. On very small disks 5% truncates to 0MB, which would
+// disable the reserve entirely (cleanup only once availableMB <= 0), so we floor
+// it at 1MB to preserve the intended "keep some space free" behaviour.
+func defaultReserveMB(totalMB int64) int64 {
+	reserveMB := totalMB * 5 / 100
+	if reserveMB < 1 {
+		reserveMB = 1
+	}
+	return reserveMB
+}
+
+// pickRecordingToCleanup chooses which recording to delete to free space in the
+// recordings directory. It returns the oldest recording that has already been
+// uploaded (no pending marker with the same name in cloudDirectory). Only when
+// every recording is still pending upload does it return the oldest recording
+// overall with pending=true, signalling the caller that it is about to drop an
+// un-uploaded recording as a last resort.
+func pickRecordingToCleanup(recordingsDirectory, cloudDirectory string) (string, bool, error) {
+	entries, err := os.ReadDir(recordingsDirectory)
+	if err != nil {
+		return "", false, err
+	}
+
+	var oldestSafeName, oldestAnyName string
+	var oldestSafeTime, oldestAnyTime time.Time
+
+	for _, entry := range entries {
+		if entry.IsDir() {
+			continue
+		}
+		info, err := entry.Info()
+		if err != nil || !info.Mode().IsRegular() {
+			continue
+		}
+		modTime := info.ModTime()
+
+		if oldestAnyName == "" || modTime.Before(oldestAnyTime) {
+			oldestAnyName = entry.Name()
+			oldestAnyTime = modTime
+		}
+
+		// A recording is still pending upload if a marker with the same name
+		// exists in the cloud directory. Skip those when picking a safe candidate.
+		if _, statErr := os.Stat(cloudDirectory + "/" + entry.Name()); statErr == nil {
+			continue
+		}
+
+		if oldestSafeName == "" || modTime.Before(oldestSafeTime) {
+			oldestSafeName = entry.Name()
+			oldestSafeTime = modTime
+		}
+	}
+
+	if oldestSafeName != "" {
+		return oldestSafeName, false, nil
+	}
+	if oldestAnyName != "" {
+		return oldestAnyName, true, nil
+	}
+	return "", false, os.ErrNotExist
+}
+
+func HandleRecordStream(queue *packets.Queue, configDirectory string, configuration *models.Configuration, communication *models.Communication, rtspClient RTSPClient, mqttClient mqtt.Client) {
 
 	config := configuration.Config
+	hubKey := config.HubKey
 	loc, _ := time.LoadLocation(config.Timezone)
 
+	// Start each capture session with manual recording off, so a leftover
+	// request from before a restart/reconnect doesn't silently persist.
+	communication.IsRecordingManual.UnSet()
+
 	if config.Capture.Recording == "false" {
 		log.Log.Info("capture.main.HandleRecordStream(): disabled, we will not record anything.")
 	} else {
@@ -223,6 +392,9 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
 
 					recordingStatus = "idle"
 
+					// Notify the hub / live-view UI that this camera stopped recording.
+					publishRecordingState(mqttClient, hubKey, configuration, false)
+
 					// Clean up the recording directory if necessary.
 					CleanupRecordingDirectory(configDirectory, configuration)
 				}
@@ -299,6 +471,9 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
 					writeSampleToMP4(mp4Video, videoTrack, audioTrack, pkt)
 					recordingStatus = "started"
 
+					// Notify the hub / live-view UI that this camera started recording.
+					publishRecordingState(mqttClient, hubKey, configuration, true)
+
 				} else if start {
 
 					writeSampleToMP4(mp4Video, videoTrack, audioTrack, pkt)
@@ -376,6 +551,9 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
 
 					recordingStatus = "idle"
 
+					// Notify the hub / live-view UI that this camera stopped recording.
+					publishRecordingState(mqttClient, hubKey, configuration, false)
+
 					// Clean up the recording directory if necessary.
 					CleanupRecordingDirectory(configDirectory, configuration)
 				}
@@ -492,6 +670,14 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
 					default:
 					}
 
+					// While a manual recording is active, keep it alive: refresh the
+					// motion timestamp every iteration so the post-recording timeout
+					// never fires. The clip still rolls over at maxRecordingPeriod and
+					// is restarted below, until the viewer stops the manual recording.
+					if communication.IsRecordingManual.IsSet() {
+						motionTimestamp = now
+					}
+
 					if start && (motionTimestamp+postRecording-now < 0 || now-startRecording > maxRecordingPeriod-500) && nextPkt.IsKeyFrame {
 						log.Log.Info("capture.main.HandleRecordStream(motiondetection): timestamp+postRecording-now < 0  - " + strconv.FormatInt(motionTimestamp+postRecording-now, 10) + " < 0")
 						log.Log.Info("capture.main.HandleRecordStream(motiondetection): now-startRecording > maxRecordingPeriod-500 - " + strconv.FormatInt(now-startRecording, 10) + " > " + strconv.FormatInt(maxRecordingPeriod-500, 10))
@@ -523,6 +709,9 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
 							log.Log.Debug("capture.main.HandleRecordStream(continuous): no AAC audio codec detected, skipping audio track.")
 						}
 						start = true
+
+						// Notify the hub / live-view UI that this camera started recording.
+						publishRecordingState(mqttClient, hubKey, configuration, true)
 					}
 					if start {
 						writeSampleToMP4(mp4Video, videoTrack, audioTrack, pkt)
@@ -557,6 +746,19 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
 				mp4Video.Close(&config)
 				log.Log.Info("capture.main.HandleRecordStream(motiondetection): file save: " + name)
 
+				// Notify the hub / live-view UI that this camera stopped recording.
+				publishRecordingState(mqttClient, hubKey, configuration, false)
+
+				// If the viewer still has a manual recording running, this clip just
+				// rolled over at the max length — immediately kick off the next
+				// segment so recording stays continuous until they stop it.
+				if communication.IsRecordingManual.IsSet() {
+					select {
+					case communication.HandleMotion <- models.MotionDataPartial{Timestamp: time.Now().Unix(), NumberOfChanges: 100000000}:
+					default:
+					}
+				}
+
 				// Update the name of the recording with the duration.
 				// We will update the name of the recording with the duration in milliseconds.
 				if mp4Video.VideoTotalDuration > 0 {

machinery/src/cloud/cloud.go

@@ -874,7 +874,7 @@ func HandleLiveStreamSD(livestreamCursor *packets.QueueCursor, configuration *mo
 	log.Log.Debug("cloud.HandleLiveStreamSD(): finished")
 }
 
-func HandleLiveStreamHD(livestreamCursor *packets.QueueCursor, configuration *models.Configuration, communication *models.Communication, mqttClient mqtt.Client, rtspClient capture.RTSPClient) {
+func HandleLiveStreamHD(configuration *models.Configuration, communication *models.Communication, mqttClient mqtt.Client, rtspClient capture.RTSPClient, rtspSubClient capture.RTSPClient, subStreamEnabled bool) {
 
 	config := configuration.Config
 
@@ -888,23 +888,51 @@ func HandleLiveStreamHD(livestreamCursor *packets.QueueCursor, configuration *mo
 			// Create per-peer broadcasters instead of shared tracks.
 			// Each viewer gets its own track with independent, non-blocking writes
 			// so a slow/congested peer cannot stall the others.
-			streams, _ := rtspClient.GetStreams()
-			videoBroadcaster := webrtc.NewVideoBroadcaster(streams)
-			audioBroadcaster := webrtc.NewAudioBroadcaster(streams)
+			//
+			// Both the main (high-resolution) and sub (low-resolution) streams are
+			// exposed as separate broadcasters that are always forwarding, so a
+			// viewer can pick the resolution it needs per peer connection without
+			// the agent re-negotiating the RTSP source.
+			mainStreams, _ := rtspClient.GetStreams()
+			mainVideoBroadcaster := webrtc.NewVideoBroadcaster(mainStreams)
+			mainAudioBroadcaster := webrtc.NewAudioBroadcaster(mainStreams)
 
-			if videoBroadcaster == nil && audioBroadcaster == nil {
-				log.Log.Error("cloud.HandleLiveStreamHD(): failed to create both video and audio broadcasters")
+			if mainVideoBroadcaster == nil && mainAudioBroadcaster == nil {
+				log.Log.Error("cloud.HandleLiveStreamHD(): failed to create both video and audio broadcasters for the main stream")
 				return
 			}
 
-			go webrtc.WriteToTrack(livestreamCursor, configuration, communication, mqttClient, videoBroadcaster, audioBroadcaster, rtspClient)
+			go webrtc.WriteToTrack(communication.Queue.Latest(), configuration, communication, mqttClient, mainVideoBroadcaster, mainAudioBroadcaster, rtspClient)
+
+			// Sub stream broadcasters, only when a distinct sub stream is available.
+			var subVideoBroadcaster *webrtc.TrackBroadcaster
+			var subAudioBroadcaster *webrtc.TrackBroadcaster
+			if subStreamEnabled && rtspSubClient != nil && communication.SubQueue != nil {
+				subStreams, _ := rtspSubClient.GetStreams()
+				subVideoBroadcaster = webrtc.NewVideoBroadcaster(subStreams)
+				subAudioBroadcaster = webrtc.NewAudioBroadcaster(subStreams)
+				go webrtc.WriteToTrack(communication.SubQueue.Latest(), configuration, communication, mqttClient, subVideoBroadcaster, subAudioBroadcaster, rtspSubClient)
+			}
+			subBroadcastersReady := subVideoBroadcaster != nil || subAudioBroadcaster != nil
 
 			if config.Capture.ForwardWebRTC == "true" {
 
 			} else {
 				log.Log.Info("cloud.HandleLiveStreamHD(): Waiting for peer connections.")
 				for handshake := range communication.HandleLiveHDHandshake {
-					log.Log.Info("cloud.HandleLiveStreamHD(): setting up a peer connection.")
+					// Route each viewer to the main or sub broadcasters based on the
+					// quality it requested; "auto" prefers the sub stream when one is
+					// available, matching the historical default.
+					useSub := models.SelectSubStreamForQuality(config, handshake.Payload.Quality, subStreamEnabled && subBroadcastersReady)
+					videoBroadcaster := mainVideoBroadcaster
+					audioBroadcaster := mainAudioBroadcaster
+					streamLabel := "main"
+					if useSub {
+						videoBroadcaster = subVideoBroadcaster
+						audioBroadcaster = subAudioBroadcaster
+						streamLabel = "sub"
+					}
+					log.Log.Info("cloud.HandleLiveStreamHD(): setting up a peer connection on the " + streamLabel + " stream (quality=" + handshake.Payload.Quality + ").")
 					go webrtc.InitializeWebRTCConnection(configuration, communication, mqttClient, videoBroadcaster, audioBroadcaster, handshake)
 				}
 			}

machinery/src/cloud/kerberos_vault.go

@@ -6,6 +6,7 @@ import (
 	"io"
 	"net/http"
 	"os"
+	"strconv"
 	"time"
 
 	"github.com/kerberos-io/agent/machinery/src/log"
@@ -199,17 +200,59 @@ func setVaultHeaders(h http.Header, vault models.KStorage, publicKey, deviceKey,
 }
 
 // newVaultHTTPClient builds an HTTP client honouring the AGENT_TLS_INSECURE
-// escape hatch. A timeout of 0 disables the client-level timeout, which is
-// required for streaming large upload bodies.
+// escape hatch. A timeout of 0 disables the *overall* client timeout, which is
+// required for streaming large upload bodies without capping the total transfer
+// time. Transport-level timeouts are still applied so that a lost network
+// connection (for example the internet being disconnected) fails reasonably
+// fast and the upload is retried, instead of the request hanging until the OS
+// TCP timeout (which can be many minutes) and blocking the whole upload loop.
 func newVaultHTTPClient(timeout time.Duration) *http.Client {
-	client := &http.Client{}
+	// Start from a clone of the default transport so we keep its sane dial and
+	// TLS-handshake timeouts, connection pooling and HTTP/2 support even when the
+	// AGENT_TLS_INSECURE escape hatch is enabled (a bare http.Transport would have
+	// no dial/handshake timeouts at all).
+	transport := http.DefaultTransport.(*http.Transport).Clone()
+
+	// ResponseHeaderTimeout bounds how long we wait for the vault's response
+	// headers *after* the request body has been fully written. It does not limit
+	// the time spent streaming the (potentially large) upload body, so big
+	// recordings still upload fine, but a vault/network that disappears while we
+	// wait for the acknowledgement is detected and the upload is retried instead
+	// of hanging indefinitely.
+	transport.ResponseHeaderTimeout = vaultResponseHeaderTimeout()
+
 	if os.Getenv("AGENT_TLS_INSECURE") == "true" {
-		client.Transport = &http.Transport{
-			TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
+		if transport.TLSClientConfig == nil {
+			transport.TLSClientConfig = &tls.Config{}
 		}
+		transport.TLSClientConfig.InsecureSkipVerify = true
 	}
+
+	client := &http.Client{Transport: transport}
 	if timeout > 0 {
 		client.Timeout = timeout
 	}
 	return client
 }
+
+// vaultResponseHeaderTimeout returns the maximum time to wait for a vault's
+// response headers after the request body has been written. It defaults to 5
+// minutes — generous enough for the vault to persist/finalize a chunk or a full
+// recording to its storage provider — and can be tuned with the
+// AGENT_VAULT_RESPONSE_HEADER_TIMEOUT_SECONDS environment variable. A value of 0
+// (or a negative/invalid value) disables the timeout.
+func vaultResponseHeaderTimeout() time.Duration {
+	const def = 5 * time.Minute
+	v := os.Getenv("AGENT_VAULT_RESPONSE_HEADER_TIMEOUT_SECONDS")
+	if v == "" {
+		return def
+	}
+	n, err := strconv.Atoi(v)
+	if err != nil {
+		return def
+	}
+	if n <= 0 {
+		return 0
+	}
+	return time.Duration(n) * time.Second
+}

machinery/src/cloud/livehls.go

@@ -1,11 +1,11 @@
 package cloud
 
 import (
+	"os"
 	"time"
 
 	mqtt "github.com/eclipse/paho.mqtt.golang"
 
-	"github.com/kerberos-io/agent/machinery/src/capture"
 	"github.com/kerberos-io/agent/machinery/src/cloud/livehls"
 	"github.com/kerberos-io/agent/machinery/src/log"
 	"github.com/kerberos-io/agent/machinery/src/models"
@@ -29,7 +29,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,7 +37,15 @@ 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.
-func HandleLiveStreamHLS(livestreamCursor *packets.QueueCursor, configuration *models.Configuration, communication *models.Communication, mqttClient mqtt.Client, _ capture.RTSPClient) {
+//
+// 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(configuration *models.Configuration, communication *models.Communication, mqttClient mqtt.Client, subStreamEnabled bool) {
 
 	log.Log.Debug("cloud.HandleLiveStreamHLS(): started")
 
@@ -73,10 +80,39 @@ func HandleLiveStreamHLS(livestreamCursor *packets.QueueCursor, configuration *m
 		DeviceKey:     deviceId,
 	})
 
+	// The live session can be served from the main (high-resolution) or sub
+	// (low-resolution) stream and switched on demand. requestedQuality tracks the
+	// latest tier asked for over the keepalive; source holds the cursor plus the
+	// encoded parameter sets/dimensions for the stream currently being muxed.
 	// Encoded dimensions are only needed for the avcC fallback path (an SPS that
-	// mp4ff's strict parser rejects); the main stream dimensions are a safe value.
-	width := uint16(config.Capture.IPCamera.Width)
-	height := uint16(config.Capture.IPCamera.Height)
+	// mp4ff's strict parser rejects).
+	requestedQuality := models.StreamQualityAuto
+	useSub := models.SelectSubStreamForQuality(config, requestedQuality, subStreamEnabled)
+	source := buildHLSSource(config, communication, useSub)
+	log.Log.Info("cloud.HandleLiveStreamHLS(): serving live HLS from the " + source.label + " stream")
+
+	// 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)
@@ -86,26 +122,97 @@ func HandleLiveStreamHLS(livestreamCursor *packets.QueueCursor, configuration *m
 	var pkt packets.Packet
 
 	for cursorError == nil {
-		pkt, cursorError = livestreamCursor.ReadPacket()
+		pkt, cursorError = source.cursor.ReadPacket()
 
 		now := time.Now().Unix()
 		select {
-		case <-communication.HandleLiveHLS:
+		case q := <-communication.HandleLiveHLS:
 			lastViewerRequest = now
+			if q != "" {
+				requestedQuality = q
+			}
 			// A keepalive may come from a viewer that just connected or hard-
 			// refreshed and therefore missed the one-shot readiness announcement
 			// 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
 			}
 		default:
 		}
 
+		// Switch the source stream when the requested quality now maps to the other
+		// stream. Tearing the current session down makes the producer rebuild the
+		// init segment and announce a fresh session id from the new stream, which the
+		// viewer re-attaches to.
+		if wantSub := models.SelectSubStreamForQuality(config, requestedQuality, subStreamEnabled); wantSub != useSub {
+			useSub = wantSub
+			if session != nil {
+				_ = session.Close()
+				session = nil
+			}
+			source = buildHLSSource(config, communication, useSub)
+			lastReadyAnnounce = 0
+			log.Log.Info("cloud.HandleLiveStreamHLS(): switched live HLS to the " + source.label + " stream (quality=" + requestedQuality + ")")
+			continue
+		}
+
 		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:       source.sps,
+					PPSNALUs:       source.pps,
+					VPSNALUs:       source.vps,
+					Width:          source.width,
+					Height:         source.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 +234,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:     source.sps,
+				PPSNALUs:     source.pps,
+				VPSNALUs:     source.vps,
+				Width:        source.width,
+				Height:       source.height,
+				PartTargetMs: partTargetMs,
 			})
 			session.SetOnReady(func(sessionID string) {
 				log.Log.Info("cloud.HandleLiveStreamHLS(): live HLS session ready, announcing " + sessionID)
@@ -175,3 +283,45 @@ func publishHLSReady(configuration *models.Configuration, mqttClient mqtt.Client
 		log.Log.Error("cloud.HandleLiveStreamHLS(): failed to package receive-hls-ready message: " + err.Error())
 	}
 }
+
+// hlsStreamSource bundles everything the live HLS producer needs to mux one of
+// the camera's streams: the packet cursor it reads from plus the encoded
+// parameter sets and dimensions used to build that stream's init segment.
+type hlsStreamSource struct {
+	cursor *packets.QueueCursor
+	sps    [][]byte
+	pps    [][]byte
+	vps    [][]byte
+	width  uint16
+	height uint16
+	label  string
+}
+
+// buildHLSSource resolves the packet cursor and encoded parameter sets/dimensions
+// for the selected stream. useSub picks the sub (low-resolution) stream when one
+// is available; otherwise the main (high-resolution) stream is used. A fresh
+// Latest() cursor is created so muxing resumes from the live edge of the chosen
+// stream after a switch.
+func buildHLSSource(config models.Config, communication *models.Communication, useSub bool) hlsStreamSource {
+	cam := config.Capture.IPCamera
+	if useSub && communication.SubQueue != nil {
+		return hlsStreamSource{
+			cursor: communication.SubQueue.Latest(),
+			sps:    cam.SubSPSNALUs,
+			pps:    cam.SubPPSNALUs,
+			vps:    cam.SubVPSNALUs,
+			width:  uint16(cam.SubWidth),
+			height: uint16(cam.SubHeight),
+			label:  "sub",
+		}
+	}
+	return hlsStreamSource{
+		cursor: communication.Queue.Latest(),
+		sps:    cam.SPSNALUs,
+		pps:    cam.PPSNALUs,
+		vps:    cam.VPSNALUs,
+		width:  uint16(cam.Width),
+		height: uint16(cam.Height),
+		label:  "main",
+	}
+}

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

@@ -46,12 +46,12 @@ func resumableUploadsEnabled() bool {
 // explicit size is configured. Splitting the upload into chunks keeps each HTTP
 // request small enough for intermediary proxies/load balancers and checkpoints
 // progress frequently, so an interruption resumes with minimal re-upload.
-const tusDefaultChunkSize int64 = 1 << 20 // 1 MiB
+const tusDefaultChunkSize int64 = 8 << 20 // 8 MiB (>= S3 multipart minimum part size)
 
 const tusProgressBucketPercent int64 = 10
 
 // tusChunkSize returns the number of bytes to send per PATCH request. It
-// defaults to tusDefaultChunkSize (1 MiB) and can be overridden with the
+// defaults to tusDefaultChunkSize (8 MiB) and can be overridden with the
 // AGENT_TUS_CHUNK_SIZE_BYTES environment variable. A value of 0 (or negative)
 // disables chunking and sends the remaining bytes in a single PATCH.
 func tusChunkSize() int64 {
@@ -154,10 +154,20 @@ func runTusUpload(baseURL, metadata, fileName, label, slot string, setHeaders tu
 	const maxAttempts = 4
 	restartedAfterComplete := false
 
+	// lastStatus holds the HTTP status code of the most recent tus request. A
+	// value of 0 means the request failed at the transport level (no HTTP
+	// response at all, e.g. the vault was unreachable or the connection dropped
+	// because the internet went down). It lets the final "gave up" return report
+	// whether the vault actually answered, so the caller only advances its
+	// retry/back-off policy on a definitive response and transient network errors
+	// never consume the retry budget (matching the legacy single-POST behaviour).
+	lastStatus := 0
+
 	for attempt := 0; attempt < maxAttempts; attempt++ {
 		// (1) Ensure we have an active upload URL, creating one if needed.
 		if uploadURL == "" {
 			created, status, cerr := tusCreate(client, baseURL, size, metadata, setHeaders, fileName)
+			lastStatus = status
 			if cerr != nil {
 				if status == http.StatusNotFound || status == http.StatusMethodNotAllowed || status == http.StatusNotImplemented {
 					// The vault does not implement tus; let the caller fall back.
@@ -173,6 +183,7 @@ func runTusUpload(baseURL, metadata, fileName, label, slot string, setHeaders tu
 
 		// (2) Query the current server-side offset.
 		offset, status, herr := tusHead(client, uploadURL, setHeaders)
+		lastStatus = status
 		if herr != nil {
 			if status == http.StatusNotFound || status == http.StatusGone {
 				// The upload expired/was removed server-side; start over.
@@ -220,6 +231,7 @@ func runTusUpload(baseURL, metadata, fileName, label, slot string, setHeaders tu
 				patchLen = chunkSize
 			}
 			newOffset, status, respBody, perr := tusPatch(client, uploadURL, offset, patchLen, file, setHeaders)
+			lastStatus = status
 			if perr != nil {
 				if status >= 400 {
 					// Definitive rejection (e.g. provider push failed during finalize).
@@ -258,7 +270,13 @@ func runTusUpload(baseURL, metadata, fileName, label, slot string, setHeaders tu
 		return true, true, true, lastBody, nil
 	}
 
-	return false, true, true, "resumable upload did not complete after retries", errors.New(label + ": resumable upload did not complete after retries")
+	// Every attempt failed. Only report responded=true when the vault actually
+	// answered on the last attempt (lastStatus > 0). If every attempt failed at
+	// the transport level (lastStatus == 0, e.g. the internet was disconnected),
+	// report responded=false so the caller keeps the recording queued and retries
+	// later instead of consuming its retry budget and entering the long back-off
+	// timeout.
+	return false, lastStatus > 0, true, "resumable upload did not complete after retries", errors.New(label + ": resumable upload did not complete after retries")
 }
 
 // uploadVaultResumable uploads a recording directly to a Kerberos Vault using
@@ -535,10 +553,15 @@ func removeTusResumeState(path string) {
 	_ = os.Remove(path)
 }
 
+// tusBackoffBaseDelay is the base delay used by tusBackoff for the exponential
+// back-off between resume attempts. It is a package variable (rather than a
+// constant) so tests can shrink it to keep them fast.
+var tusBackoffBaseDelay = 500 * time.Millisecond
+
 // tusBackoff sleeps for an exponentially increasing duration (capped) between
 // resume attempts to avoid hammering a temporarily unavailable vault.
 func tusBackoff(attempt int) {
-	delay := time.Duration(500*(1<<uint(attempt))) * time.Millisecond
+	delay := tusBackoffBaseDelay * time.Duration(1<<uint(attempt))
 	if delay > 3*time.Second {
 		delay = 3 * time.Second
 	}

machinery/src/cloud/tus_client_test.go

@@ -5,6 +5,7 @@ import (
 	"encoding/base64"
 	"fmt"
 	"io"
+	"net"
 	"net/http"
 	"net/http/httptest"
 	"os"
@@ -13,6 +14,7 @@ import (
 	"strings"
 	"sync"
 	"testing"
+	"time"
 
 	"github.com/kerberos-io/agent/machinery/src/models"
 )
@@ -350,6 +352,47 @@ func TestUploadVaultResumable_Unsupported(t *testing.T) {
 	}
 }
 
+// TestUploadVaultResumable_NetworkErrorKeepsRetryBudget verifies that when the
+// vault is unreachable (mimicking the internet being disconnected) the resumable
+// upload reports responded=false. That is what stops the caller
+// (UploadKerberosVault) from consuming its retry budget and entering the long
+// back-off timeout on a transient network outage, so the recording keeps being
+// retried until connectivity returns.
+func TestUploadVaultResumable_NetworkErrorKeepsRetryBudget(t *testing.T) {
+	// Bind then immediately release a loopback port so every connection to it is
+	// refused, producing a transport-level error (no HTTP response).
+	ln, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("listen: %v", err)
+	}
+	addr := ln.Addr().String()
+	if cerr := ln.Close(); cerr != nil {
+		t.Fatalf("close listener: %v", cerr)
+	}
+
+	// Keep the between-attempt back-off tiny so the test stays fast.
+	oldDelay := tusBackoffBaseDelay
+	tusBackoffBaseDelay = time.Millisecond
+	defer func() { tusBackoffBaseDelay = oldDelay }()
+
+	fileName := "1564859471_6-474162_oprit_577-283-727-375_1153_27.mp4"
+	withRecording(t, fileName, bytes.Repeat([]byte("n"), 2048))
+
+	uploaded, responded, supported, _, err := uploadVaultResumable(testVault("http://"+addr), "pk", "dev", fileName, "test", "primary")
+	if uploaded {
+		t.Fatal("expected uploaded=false when the vault is unreachable")
+	}
+	if !supported {
+		t.Fatal("a transport error is not a missing tus endpoint; expected supported=true")
+	}
+	if responded {
+		t.Fatal("expected responded=false for a pure network error so the retry budget is preserved")
+	}
+	if err == nil {
+		t.Fatal("expected an error when the vault is unreachable")
+	}
+}
+
 func TestUploadVaultResumable_FinalizeRetry(t *testing.T) {
 	srv := newFakeTus()
 	srv.failFinalize = 1

machinery/src/components/kerberos.go

@@ -73,8 +73,9 @@ func Bootstrap(ctx context.Context, configDirectory string, configuration *model
 	communication.HandleLiveSDHTTP = make(chan int64, 1)
 	communication.HandleLiveHDKeepalive = make(chan string, 1)
 	communication.HandleLiveHDPeers = make(chan string, 1)
-	communication.HandleLiveHLS = make(chan int64, 1)
+	communication.HandleLiveHLS = make(chan string, 1)
 	communication.IsConfiguring = abool.New()
+	communication.IsRecordingManual = abool.New()
 
 	cameraSettings := &models.Camera{}
 
@@ -237,6 +238,13 @@ func RunAgent(configDirectory string, configuration *models.Configuration, commu
 		configuration.Config.Capture.IPCamera.SubWidth = width
 		configuration.Config.Capture.IPCamera.SubHeight = height
 
+		// Capture the sub stream parameter sets separately from the main stream so
+		// the live HLS muxer can build a correct init segment when a viewer asks for
+		// the sub (low-resolution) stream on demand.
+		configuration.Config.Capture.IPCamera.SubSPSNALUs = [][]byte{videoSubStream.SPS}
+		configuration.Config.Capture.IPCamera.SubPPSNALUs = [][]byte{videoSubStream.PPS}
+		configuration.Config.Capture.IPCamera.SubVPSNALUs = [][]byte{videoSubStream.VPS}
+
 		// 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).
 		configuration.Config.Capture.IPCamera.BaseWidth, configuration.Config.Capture.IPCamera.BaseHeight =
@@ -287,29 +295,22 @@ func RunAgent(configDirectory string, configuration *models.Configuration, commu
 	}
 
 	// Handle livestream HLS (adaptive segments over HTTP via hub-api -> vault).
-	// Uses the sub stream when available (lower bitrate, browser-friendly), else
-	// the main stream. Like SD it is viewer-keepalive gated and produces no
-	// traffic while nobody is watching.
-	if subStreamEnabled {
-		livestreamHLSCursor := subQueue.Latest()
-		go cloud.HandleLiveStreamHLS(livestreamHLSCursor, configuration, communication, mqttClient, rtspSubClient)
-	} else {
-		livestreamHLSCursor := queue.Latest()
-		go cloud.HandleLiveStreamHLS(livestreamHLSCursor, configuration, communication, mqttClient, rtspClient)
-	}
+	// The producer can serve either the main (high-resolution) or sub
+	// (low-resolution) stream and switches between them on demand based on the
+	// quality the viewer requests; "auto" prefers the sub stream when available.
+	// Like SD it is viewer-keepalive gated and produces no traffic while nobody is
+	// watching.
+	go cloud.HandleLiveStreamHLS(configuration, communication, mqttClient, subStreamEnabled)
 
-	// Handle livestream HD (high resolution over WEBRTC)
+	// Handle livestream HD (high resolution over WEBRTC). Both the main and sub
+	// stream are exposed as separate broadcasters so a viewer can request the
+	// high (main) or low (sub) resolution per peer connection; "auto" prefers the
+	// sub stream when available.
 	communication.HandleLiveHDHandshake = make(chan models.LiveHDHandshake, 100)
-	if subStreamEnabled {
-		livestreamHDCursor := subQueue.Latest()
-		go cloud.HandleLiveStreamHD(livestreamHDCursor, configuration, communication, mqttClient, rtspSubClient)
-	} else {
-		livestreamHDCursor := queue.Latest()
-		go cloud.HandleLiveStreamHD(livestreamHDCursor, configuration, communication, mqttClient, rtspClient)
-	}
+	go cloud.HandleLiveStreamHD(configuration, communication, mqttClient, rtspClient, rtspSubClient, subStreamEnabled)
 
 	// Handle recording, will write an mp4 to disk.
-	go capture.HandleRecordStream(queue, configDirectory, configuration, communication, rtspClient)
+	go capture.HandleRecordStream(queue, configDirectory, configuration, communication, rtspClient, mqttClient)
 
 	// Handle processing of motion
 	communication.HandleMotion = make(chan models.MotionDataPartial, 10)

machinery/src/config/main.go

@@ -327,6 +327,12 @@ func applyAgentEnvVars(configuration *models.Configuration, prefix string, apply
 					configuration.Config.MaxDirectorySize = size
 				}
 				break
+			case "AGENT_AUTO_CLEAN_MIN_FREE_SPACE":
+				size, err := strconv.ParseInt(value, 10, 64)
+				if err == nil {
+					configuration.Config.MinFreeSpace = size
+				}
+				break
 
 			/* Camera configuration */
 			case "AGENT_CAPTURE_IPCAMERA_RTSP":

machinery/src/models/communication.go

@@ -41,14 +41,23 @@ type Communication struct {
 	HandleLiveHDKeepalive chan string
 	HandleLiveHDHandshake chan LiveHDHandshake
 	HandleLiveHDPeers     chan string
-	HandleLiveHLS         chan int64
-	HandleONVIF           chan OnvifAction
-	IsConfiguring         *abool.AtomicBool
-	Queue                 *packets.Queue
-	SubQueue              *packets.Queue
-	Image                 string
-	CameraConnected       bool
-	MainStreamConnected   bool
-	SubStreamConnected    bool
-	HasBackChannel        bool
+	// HandleLiveHLS is the live HLS viewer keepalive. It carries the requested
+	// quality tier ("auto"|"high"|"low"; empty => auto) so the producer can switch
+	// the live session between the main and sub stream on demand.
+	HandleLiveHLS chan string
+	HandleONVIF   chan OnvifAction
+	IsConfiguring *abool.AtomicBool
+	// IsRecordingManual is set while a viewer has requested a manual recording
+	// from the live view (the record button). While set, the motion-based
+	// recorder keeps recording (it does not auto-close on the post-recording
+	// timeout) until the viewer stops it again. It is independent of motion
+	// detection so it also works when nothing is moving.
+	IsRecordingManual   *abool.AtomicBool
+	Queue               *packets.Queue
+	SubQueue            *packets.Queue
+	Image               string
+	CameraConnected     bool
+	MainStreamConnected bool
+	SubStreamConnected  bool
+	HasBackChannel      bool
 }

machinery/src/models/config.go

@@ -21,6 +21,7 @@ type Config struct {
 	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"`
@@ -99,8 +100,14 @@ type IPCamera struct {
 	SPSNALUs      [][]byte `json:"sps_nalus,omitempty" bson:"sps_nalus,omitempty"`
 	PPSNALUs      [][]byte `json:"pps_nalus,omitempty" bson:"pps_nalus,omitempty"`
 	VPSNALUs      [][]byte `json:"vps_nalus,omitempty" bson:"vps_nalus,omitempty"`
-	SampleRate    int      `json:"sample_rate,omitempty" bson:"sample_rate,omitempty"`
-	Channels      int      `json:"channels,omitempty" bson:"channels,omitempty"`
+	// Sub stream parameter sets, captured separately from the main stream so the
+	// live HLS muxer can build a correct init segment when a viewer switches the
+	// live view to the sub (low-resolution) stream.
+	SubSPSNALUs [][]byte `json:"sub_sps_nalus,omitempty" bson:"sub_sps_nalus,omitempty"`
+	SubPPSNALUs [][]byte `json:"sub_pps_nalus,omitempty" bson:"sub_pps_nalus,omitempty"`
+	SubVPSNALUs [][]byte `json:"sub_vps_nalus,omitempty" bson:"sub_vps_nalus,omitempty"`
+	SampleRate  int      `json:"sample_rate,omitempty" bson:"sample_rate,omitempty"`
+	Channels    int      `json:"channels,omitempty" bson:"channels,omitempty"`
 }
 
 // USBCamera configuration, such as the device path (/dev/video*)

machinery/src/models/mqtt.go

@@ -150,6 +150,10 @@ type AudioPayload struct {
 // We received a recording request, we'll send it to the motion handler.
 type RecordPayload struct {
 	Timestamp int64 `json:"timestamp"` // timestamp of the recording request.
+	// Recording toggles a manual recording from the live view: true starts a
+	// recording (and keeps it running), false stops it. Older clients that only
+	// send a timestamp default to false; the live view always sets it explicitly.
+	Recording bool `json:"recording"`
 }
 
 // We received a preset position request, we'll request it through onvif and send it back.
@@ -179,11 +183,26 @@ type RequestSDStreamPayload struct {
 	Transport string `json:"transport,omitempty"`
 }
 
+// Stream quality tiers a viewer can request for the live (HD) view. The agent
+// maps these onto the camera's main (high-resolution) or sub (low-resolution)
+// RTSP stream, so a viewer can pick the resolution it needs instead of the agent
+// always preferring the sub stream. Empty/unknown values are treated as "auto"
+// for backward compatibility: older frontends that never set a quality keep the
+// previous behaviour (sub stream when available, otherwise main).
+const (
+	StreamQualityAuto = "auto" // agent decides based on availability/resolution
+	StreamQualityHigh = "high" // main stream (highest resolution)
+	StreamQualityLow  = "low"  // sub stream (lowest resolution)
+)
+
 // We received a live HLS stream request. Like SD it is a simple viewer
 // keepalive: the agent owns the live HLS session, so the request only needs to
-// signal "a viewer is watching" to keep the segment pipeline alive.
+// signal "a viewer is watching" to keep the segment pipeline alive. Quality lets
+// the viewer ask for the main (high) or sub (low) stream on demand; the agent
+// switches the live session's source stream when it changes.
 type RequestHLSStreamPayload struct {
-	Timestamp int64 `json:"timestamp"` // timestamp
+	Timestamp int64  `json:"timestamp"`         // timestamp
+	Quality   string `json:"quality,omitempty"` // "auto" | "high" | "low" (empty => auto)
 }
 
 // We received a request HD stream request
@@ -192,6 +211,7 @@ type RequestHDStreamPayload struct {
 	HubKey             string `json:"hub_key"`             // hub key
 	SessionID          string `json:"session_id"`          // session id
 	SessionDescription string `json:"session_description"` // session description
+	Quality            string `json:"quality,omitempty"`   // "auto" | "high" | "low" (empty => auto)
 }
 
 // We received a receive HD candidates request

machinery/src/models/stream.go

@@ -0,0 +1,40 @@
+package models
+
+// SelectSubStreamForQuality decides whether the live (HD) view should be served
+// from the sub (secondary) RTSP stream for the requested quality tier.
+//
+// It is resolution-aware: "high" picks whichever configured stream has the
+// higher resolution and "low" whichever has the lower resolution, regardless of
+// which one is wired as main vs sub. "auto" — the default, also used for the
+// empty/unknown value sent by older frontends that never set a quality — keeps
+// the historical behaviour of preferring the sub stream when one is available
+// (lower bitrate, browser friendly), falling back to the main stream otherwise.
+//
+// When no sub stream is configured the main stream is always used.
+func SelectSubStreamForQuality(config Config, quality string, subStreamEnabled bool) bool {
+	if !subStreamEnabled {
+		return false
+	}
+
+	cam := config.Capture.IPCamera
+	mainPixels := cam.Width * cam.Height
+	subPixels := cam.SubWidth * cam.SubHeight
+
+	switch quality {
+	case StreamQualityHigh:
+		// Highest resolution available. If the sub stream is (unusually) larger,
+		// use it; otherwise use the main stream. When dimensions are not yet known
+		// (0), default to the main stream for "high".
+		return subPixels > mainPixels
+	case StreamQualityLow:
+		// Lowest resolution available. If the main stream is (unusually) the
+		// smaller of the two, use it; otherwise use the sub stream. When the sub
+		// dimensions are unknown, still prefer the sub stream for "low".
+		if mainPixels > 0 && subPixels > 0 && mainPixels < subPixels {
+			return false
+		}
+		return true
+	default: // StreamQualityAuto, empty, or any unknown value
+		return true
+	}
+}

machinery/src/models/stream_test.go

@@ -0,0 +1,53 @@
+package models
+
+import "testing"
+
+func cfgWithDims(mainW, mainH, subW, subH int) Config {
+	c := Config{}
+	c.Capture.IPCamera.Width = mainW
+	c.Capture.IPCamera.Height = mainH
+	c.Capture.IPCamera.SubWidth = subW
+	c.Capture.IPCamera.SubHeight = subH
+	return c
+}
+
+func TestSelectSubStreamForQuality(t *testing.T) {
+	tests := []struct {
+		name             string
+		config           Config
+		quality          string
+		subStreamEnabled bool
+		wantSub          bool
+	}{
+		// No sub stream configured -> always the main stream.
+		{"no sub, auto", cfgWithDims(1920, 1080, 0, 0), StreamQualityAuto, false, false},
+		{"no sub, high", cfgWithDims(1920, 1080, 0, 0), StreamQualityHigh, false, false},
+		{"no sub, low", cfgWithDims(1920, 1080, 0, 0), StreamQualityLow, false, false},
+
+		// Typical config: main is the bigger stream, sub the smaller one.
+		{"auto prefers sub", cfgWithDims(1920, 1080, 640, 480), StreamQualityAuto, true, true},
+		{"empty prefers sub", cfgWithDims(1920, 1080, 640, 480), "", true, true},
+		{"unknown prefers sub", cfgWithDims(1920, 1080, 640, 480), "potato", true, true},
+		{"high picks main", cfgWithDims(1920, 1080, 640, 480), StreamQualityHigh, true, false},
+		{"low picks sub", cfgWithDims(1920, 1080, 640, 480), StreamQualityLow, true, true},
+
+		// Dimensions not probed yet (0): high defaults to main, low/auto to sub.
+		{"unknown dims, high", cfgWithDims(0, 0, 0, 0), StreamQualityHigh, true, false},
+		{"unknown dims, low", cfgWithDims(0, 0, 0, 0), StreamQualityLow, true, true},
+		{"unknown dims, auto", cfgWithDims(0, 0, 0, 0), StreamQualityAuto, true, true},
+
+		// Inverted config: sub is (unusually) the higher-resolution stream.
+		{"inverted high picks sub", cfgWithDims(640, 480, 1920, 1080), StreamQualityHigh, true, true},
+		{"inverted low picks main", cfgWithDims(640, 480, 1920, 1080), StreamQualityLow, true, false},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := SelectSubStreamForQuality(tt.config, tt.quality, tt.subStreamEnabled)
+			if got != tt.wantSub {
+				t.Errorf("SelectSubStreamForQuality(quality=%q, subEnabled=%v) = %v, want %v",
+					tt.quality, tt.subStreamEnabled, got, tt.wantSub)
+			}
+		})
+	}
+}

machinery/src/routers/mqtt/main.go

@@ -375,11 +375,29 @@ func HandleRecording(mqttClient mqtt.Client, hubKey string, payload models.Paylo
 	var recordPayload models.RecordPayload
 	json.Unmarshal(jsonData, &recordPayload)
 
-	if recordPayload.Timestamp != 0 {
-		motionDataPartial := models.MotionDataPartial{
-			Timestamp: recordPayload.Timestamp,
+	timestamp := recordPayload.Timestamp
+	if timestamp == 0 {
+		timestamp = time.Now().Unix()
+	}
+
+	if recordPayload.Recording {
+		// Start a manual recording from the live view (record button). Keep it
+		// running until the viewer stops it again — the motion recorder honours
+		// communication.IsRecordingManual and won't auto-close on the
+		// post-recording timeout while it's set. We also inject a motion event
+		// so the recording starts immediately, even when nothing is moving.
+		log.Log.Info("routers.mqtt.main.HandleRecording(): manual recording started.")
+		communication.IsRecordingManual.Set()
+		select {
+		case communication.HandleMotion <- models.MotionDataPartial{Timestamp: timestamp, NumberOfChanges: 100000000}:
+		default:
+			log.Log.Warning("routers.mqtt.main.HandleRecording(): motion channel full, manual recording start not queued.")
 		}
-		communication.HandleMotion <- motionDataPartial
+	} else {
+		// Stop the manual recording; the motion recorder closes the clip once the
+		// post-recording window elapses.
+		log.Log.Info("routers.mqtt.main.HandleRecording(): manual recording stopped.")
+		communication.IsRecordingManual.UnSet()
 	}
 }
 
@@ -585,8 +603,11 @@ func HandleRequestHLSStream(mqttClient mqtt.Client, hubKey string, payload model
 
 	if requestHLSStreamPayload.Timestamp != 0 {
 		if communication.CameraConnected {
+			// Forward the requested quality ("auto"|"high"|"low"; empty => auto) so
+			// the producer can switch the live session between the main and sub
+			// stream on demand. The send doubles as the viewer keepalive.
 			select {
-			case communication.HandleLiveHLS <- time.Now().Unix():
+			case communication.HandleLiveHLS <- requestHLSStreamPayload.Quality:
 			default:
 			}
 			log.Log.Info("routers.mqtt.main.HandleRequestHLSStream(): received request to livestream over HLS.")

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)
+	}
+}
+