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

@@ -6,7 +6,6 @@ import (
 
 	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"
@@ -46,7 +45,7 @@ const hlsReadyReannounceSeconds = 2
 // trades a little idle CPU for a near-instant "requesting stream", so viewers no
 // longer wait a full GOP for the first segment to be cut. Set
 // AGENT_LIVE_HLS_PREWARM=false to fall back to the lazy on-demand path above.
-func HandleLiveStreamHLS(livestreamCursor *packets.QueueCursor, configuration *models.Configuration, communication *models.Communication, mqttClient mqtt.Client, _ capture.RTSPClient) {
+func HandleLiveStreamHLS(configuration *models.Configuration, communication *models.Communication, mqttClient mqtt.Client, subStreamEnabled bool) {
 
 	log.Log.Debug("cloud.HandleLiveStreamHLS(): started")
 
@@ -81,10 +80,16 @@ 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
@@ -117,12 +122,15 @@ 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)
@@ -138,6 +146,22 @@ func HandleLiveStreamHLS(livestreamCursor *packets.QueueCursor, configuration *m
 		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 {
@@ -151,11 +175,11 @@ func HandleLiveStreamHLS(livestreamCursor *packets.QueueCursor, configuration *m
 				}
 				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,
+					SPSNALUs:       source.sps,
+					PPSNALUs:       source.pps,
+					VPSNALUs:       source.vps,
+					Width:          source.width,
+					Height:         source.height,
 					PartTargetMs:   partTargetMs,
 					StartBuffering: true,
 				})
@@ -211,11 +235,11 @@ func HandleLiveStreamHLS(livestreamCursor *packets.QueueCursor, configuration *m
 			}
 			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,
+				SPSNALUs:     source.sps,
+				PPSNALUs:     source.pps,
+				VPSNALUs:     source.vps,
+				Width:        source.width,
+				Height:       source.height,
 				PartTargetMs: partTargetMs,
 			})
 			session.SetOnReady(func(sessionID string) {
@@ -259,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/tus_client.go

@@ -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.")