fix(playback): trust completed segment writes

fix(playback): bound sparse seek cleanup
fix(playback): finish segment acknowledgements
2026-08-11 15:02:38 -04:00 · 2026-08-11 14:55:27 -04:00 · 2026-08-11 14:45:33 -04:00 · 2026-08-11 14:30:55 -04:00 · 2026-08-11 14:21:33 -04:00 · 2026-08-11 13:42:43 -04:00
28 changed files with 1949 additions and 105 deletions
@@ -33,6 +33,7 @@ import (
 	"github.com/Silo-Server/silo-server/internal/settingsresolve"
 	"github.com/Silo-Server/silo-server/internal/streamtoken"
 	"github.com/Silo-Server/silo-server/internal/subtitles"
+	"github.com/Silo-Server/silo-server/internal/transcodeproxy"
 	"github.com/Silo-Server/silo-server/internal/userstore"
 	"github.com/Silo-Server/silo-server/internal/watchstate"
 	"github.com/Silo-Server/silo-server/internal/watchsync"
@@ -229,10 +230,11 @@ func NewPlaybackHandler(sessionMgr SessionManagerInterface, opts ...FilePathReso
 	h.tm.Config = func() playback.TranscodeRuntimeConfig {
 		c := h.playbackConfig()
 		return playback.TranscodeRuntimeConfig{
-			TranscodeDir: c.TranscodeDir,
-			FFmpegPath:   c.FFmpegPath,
-			HWAccel:      c.HWAccel,
-			HWDevice:     c.HWDevice,
+			TranscodeDir:            c.TranscodeDir,
+			FFmpegPath:              c.FFmpegPath,
+			HWAccel:                 c.HWAccel,
+			HWDevice:                c.HWDevice,
+			SegmentRetentionSeconds: c.SegmentRetentionSeconds,
 		}
 	}
 	h.tm.StartThrottler = func(ctx context.Context, ts *playback.TranscodeSession) {
@@ -1374,7 +1376,7 @@ func (h *PlaybackHandler) HandleGetTranscodeSegment(w http.ResponseWriter, r *ht
 	h.touchSessionActivity(sessionID)

 	segmentName := chi.URLParam(r, "name")
-	segmentPath, err := transcodeSession.GetSegment(segmentName)
+	segmentLease, err := transcodeSession.OpenSegment(segmentName)
 	if err != nil && errors.Is(err, playback.ErrSegmentNotFound) {
 		segNum, parseErr := playback.ParseSegmentNumber(segmentName)
 		if parseErr == nil {
@@ -1411,7 +1413,7 @@ func (h *PlaybackHandler) HandleGetTranscodeSegment(w http.ResponseWriter, r *ht
 					"session", sessionID,
 					"playback_session_id", sessionID,
 				)
-				segmentPath, err = transcodeSession.WaitForSegment(segmentName, decision.WaitTimeout)
+				segmentLease, err = transcodeSession.WaitForOpenSegment(segmentName, decision.WaitTimeout)
 				if err != nil && errors.Is(err, playback.ErrSegmentNotFound) {
 					slog.InfoContext(r.Context(), "transcode segment wait timeout", "component", "api",
 						"segment", segmentName,
@@ -1473,7 +1475,7 @@ func (h *PlaybackHandler) HandleGetTranscodeSegment(w http.ResponseWriter, r *ht
 					); restartErr == nil {
 						// Throttler + exit monitor re-arm via the session's
 						// restart hook.
-						segmentPath, err = transcodeSession.WaitForSegment(segmentName, 30*time.Second)
+						segmentLease, err = transcodeSession.WaitForOpenSegment(segmentName, 30*time.Second)
 						if err == nil && strings.EqualFold(transcodeSession.Opts().TargetCodecVideo, "copy") {
 							// Copy-mode seeks can resume as soon as the target segment
 							// exists, but that sometimes leaves the player one segment
@@ -1481,7 +1483,9 @@ func (h *PlaybackHandler) HandleGetTranscodeSegment(w http.ResponseWriter, r *ht
 							// for a single lookahead fragment when available so the
 							// first resumed playback window is less brittle.
 							nextSegmentName := fmt.Sprintf("seg_%05d.m4s", segNum+1)
-							_, _ = transcodeSession.WaitForSegment(nextSegmentName, 1200*time.Millisecond)
+							if nextSegment, nextErr := transcodeSession.WaitForOpenSegment(nextSegmentName, 1200*time.Millisecond); nextErr == nil {
+								_ = nextSegment.Close()
+							}
 						}
 					}
 				}
@@ -1489,7 +1493,7 @@ func (h *PlaybackHandler) HandleGetTranscodeSegment(w http.ResponseWriter, r *ht
 		} else if transcodeSession.IsRunning() {
 			// Non-numbered segment (e.g., init.mp4 for fMP4 HLS).
 			// Wait briefly — the init segment is written almost immediately.
-			segmentPath, err = transcodeSession.WaitForSegment(segmentName, 10*time.Second)
+			segmentLease, err = transcodeSession.WaitForOpenSegment(segmentName, 10*time.Second)
 		}
 	}
 	if err != nil {
@@ -1501,14 +1505,17 @@ func (h *PlaybackHandler) HandleGetTranscodeSegment(w http.ResponseWriter, r *ht
 		return
 	}

-	// Report segment download for throttle tracking.
-	if segNum, parseErr := playback.ParseSegmentNumber(segmentName); parseErr == nil {
-		transcodeSession.ReportSegmentDownloaded(segNum)
-	}
-
 	w.Header().Set("Cache-Control", "no-store, max-age=0")
 	w.Header().Set("Pragma", "no-cache")
-	http.ServeFile(w, r, segmentPath)
+	defer func() { _ = segmentLease.Close() }()
+	sw := httpstream.NewRollingDeadlineWriter(w)
+	http.ServeContent(sw, r, segmentLease.Info.Name(), segmentLease.Info.ModTime(), segmentLease.File)
+	if r.Method == http.MethodGet &&
+		sw.CompletedFullResponse(segmentLease.Info.Size()) {
+		if segNum, parseErr := playback.ParseSegmentNumber(segmentName); parseErr == nil {
+			transcodeSession.ReportSegmentDownloadedForGeneration(segNum, segmentLease.Generation)
+		}
+	}
 }

 // buildProxyManifestURL signs a stream token carrying the session's full
@@ -1531,6 +1538,9 @@ func (h *PlaybackHandler) buildProxyManifestURL(card playback.RecipeCard, proxyN
 func (h *PlaybackHandler) proxyToTranscodeNode(w http.ResponseWriter, r *http.Request, transcodeNodeURL, path string) {
 	sessionID := chi.URLParam(r, "session_id")
 	targetURL := transcodeNodeURL + path
+	isSegmentRoute := strings.Contains(path, "/segment/")
+	_, segmentParseErr := playback.ParseSegmentNumber(filepath.Base(path))
+	isMediaSegment := segmentParseErr == nil
 	// Capture the signed stream token ("st") before stripping it from the URL.
 	// We forward it out-of-band as a header so the node can reconstruct after a
 	// self-restart, while keeping it out of the forwarded/logged URL.
@@ -1550,6 +1560,14 @@ func (h *PlaybackHandler) proxyToTranscodeNode(w http.ResponseWriter, r *http.Re
 		return
 	}
 	req.Header.Set("Authorization", "Bearer "+h.JWTSecret)
+	if isSegmentRoute {
+		// The node's immediate transport peer is this API process, so receiving a
+		// complete response there does not prove that the browser received it.
+		// Suppress node-local accounting and acknowledge only after the downstream
+		// writer completes. Forward range validators so both hops serve the same
+		// representation.
+		transcodeproxy.PrepareRequest(req, r)
+	}
 	// Best-effort forward of the stream token as a header so the node's
 	// reconstruct path (X-Silo-Stream-Token) can rebuild after a self-restart.
 	// Verify at the API boundary and confirm it belongs to this session; an
@@ -1573,7 +1591,7 @@ func (h *PlaybackHandler) proxyToTranscodeNode(w http.ResponseWriter, r *http.Re
 		http.Error(w, "transcode node unavailable", http.StatusBadGateway)
 		return
 	}
-	defer resp.Body.Close()
+	defer func() { _ = resp.Body.Close() }()

 	// The node strips "st" from the request query (kept out of node URLs/logs),
 	// so the segment/init URIs in the manifest it builds carry no token. Without
@@ -1603,16 +1621,22 @@ func (h *PlaybackHandler) proxyToTranscodeNode(w http.ResponseWriter, r *http.Re
 		return
 	}

-	for k, vv := range resp.Header {
-		for _, v := range vv {
-			w.Header().Add(k, v)
-		}
-	}
+	generation := resp.Header.Get(transcodeproxy.GenerationHeader)
+	transcodeproxy.CopyResponseHeaders(w.Header(), resp.Header)
 	// Proxied transcode output can stream past the server's absolute
 	// WriteTimeout; roll the write deadline with progress instead.
 	sw := httpstream.NewRollingDeadlineWriter(w)
 	sw.WriteHeader(resp.StatusCode)
-	io.Copy(sw, resp.Body)
+	if _, copyErr := io.Copy(sw, resp.Body); copyErr != nil {
+		return
+	}
+	fullSize := transcodeproxy.FullRepresentationSize(resp)
+	if isMediaSegment && generation != "" && r.Method == http.MethodGet &&
+		sw.CompletedFullResponse(fullSize) {
+		if ackErr := transcodeproxy.Acknowledge(r.Context(), http.DefaultClient, transcodeNodeURL+path, h.JWTSecret, generation); ackErr != nil {
+			slog.WarnContext(r.Context(), "acknowledge transcode segment completion", "component", "api", "error", ackErr, "playback_session_id", sessionID)
+		}
+	}
 }

 // maybeStartThrottler reads throttle settings and starts the throttler if enabled.
@@ -0,0 +1,19 @@
+package handlers
+
+import (
+	"testing"
+
+	"github.com/Silo-Server/silo-server/internal/config"
+	"github.com/Silo-Server/silo-server/internal/playback"
+)
+
+func TestPlaybackHandlerTranscodeRuntimeConfigIncludesSegmentRetention(t *testing.T) {
+	handler := NewPlaybackHandler(playback.NewSessionManager(0, 0))
+	handler.PlaybackConfig = func() config.PlaybackConfig {
+		return config.PlaybackConfig{SegmentRetentionSeconds: 300}
+	}
+
+	if got := handler.tm.Config().SegmentRetentionSeconds; got != 300 {
+		t.Fatalf("configured segment retention = %d, want 300", got)
+	}
+}
@@ -0,0 +1,127 @@
+package handlers
+
+import (
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/Silo-Server/silo-server/internal/transcodeproxy"
+)
+
+func TestProxyToTranscodeNodeAcknowledgesOnlyFullDownstreamResponse(t *testing.T) {
+	const (
+		body       = "complete segment"
+		generation = "17"
+	)
+	tests := []struct {
+		name        string
+		rangeHeader string
+		wantStatus  int
+		wantAck     int32
+	}{
+		{name: "ordinary get", wantStatus: http.StatusOK, wantAck: 1},
+		{name: "whole-file range", rangeHeader: "bytes=0-", wantStatus: http.StatusPartialContent, wantAck: 1},
+		{name: "partial range", rangeHeader: "bytes=1-", wantStatus: http.StatusPartialContent},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			var acknowledgements atomic.Int32
+			seenRange := make(chan string, 1)
+			node := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				switch r.Method {
+				case http.MethodGet:
+					if r.Header.Get(transcodeproxy.RequestHeader) != "1" {
+						t.Error("segment request omitted proxy marker")
+					}
+					seenRange <- r.Header.Get("Range")
+					w.Header().Set(transcodeproxy.GenerationHeader, generation)
+					http.ServeContent(w, r, "seg_00007.ts", time.Time{}, strings.NewReader(body))
+				case http.MethodPost:
+					if r.Header.Get(transcodeproxy.GenerationHeader) != generation {
+						t.Errorf("ack generation = %q, want %q", r.Header.Get(transcodeproxy.GenerationHeader), generation)
+					}
+					acknowledgements.Add(1)
+					w.WriteHeader(http.StatusNoContent)
+				default:
+					http.Error(w, "unexpected method", http.StatusMethodNotAllowed)
+				}
+			}))
+			defer node.Close()
+
+			handler := &PlaybackHandler{JWTSecret: "proxy-test-secret"}
+			req := withPlaybackRouteParam(
+				httptest.NewRequest(http.MethodGet, "/playback/transcode/public/segment/seg_00007.ts", nil),
+				"session_id",
+				"public",
+			)
+			req.Header.Set("Range", tt.rangeHeader)
+			rr := httptest.NewRecorder()
+			handler.proxyToTranscodeNode(rr, req, node.URL, "/transcode/remote/segment/seg_00007.ts")
+
+			if rr.Code != tt.wantStatus {
+				t.Fatalf("status = %d, want %d; body = %q", rr.Code, tt.wantStatus, rr.Body.String())
+			}
+			if got := <-seenRange; got != tt.rangeHeader {
+				t.Fatalf("forwarded Range = %q, want %q", got, tt.rangeHeader)
+			}
+			if got := rr.Header().Get(transcodeproxy.GenerationHeader); got != "" {
+				t.Fatalf("internal generation leaked downstream: %q", got)
+			}
+			if got := acknowledgements.Load(); got != tt.wantAck {
+				t.Fatalf("acknowledgements = %d, want %d", got, tt.wantAck)
+			}
+		})
+	}
+}
+
+func TestProxyToTranscodeNodeDoesNotAcknowledgeFailedDownstreamWrite(t *testing.T) {
+	const body = "complete segment"
+	var acknowledgements atomic.Int32
+	node := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.Method == http.MethodPost {
+			acknowledgements.Add(1)
+			w.WriteHeader(http.StatusNoContent)
+			return
+		}
+		w.Header().Set(transcodeproxy.GenerationHeader, "21")
+		http.ServeContent(w, r, "seg_00007.ts", time.Time{}, strings.NewReader(body))
+	}))
+	defer node.Close()
+
+	handler := &PlaybackHandler{JWTSecret: "proxy-test-secret"}
+	req := withPlaybackRouteParam(
+		httptest.NewRequest(http.MethodGet, "/playback/transcode/public/segment/seg_00007.ts", nil),
+		"session_id",
+		"public",
+	)
+	w := &failingProxyResponseWriter{header: make(http.Header), remaining: 5}
+	handler.proxyToTranscodeNode(w, req, node.URL, "/transcode/remote/segment/seg_00007.ts")
+
+	if got := acknowledgements.Load(); got != 0 {
+		t.Fatalf("failed downstream transfer produced %d acknowledgement(s)", got)
+	}
+}
+
+type failingProxyResponseWriter struct {
+	header    http.Header
+	status    int
+	remaining int
+}
+
+func (w *failingProxyResponseWriter) Header() http.Header { return w.header }
+
+func (w *failingProxyResponseWriter) WriteHeader(status int) { w.status = status }
+
+func (w *failingProxyResponseWriter) Write(p []byte) (int, error) {
+	if w.remaining <= 0 {
+		return 0, io.ErrClosedPipe
+	}
+	n := min(len(p), w.remaining)
+	w.remaining -= n
+	return n, io.ErrClosedPipe
+}
@@ -890,7 +890,7 @@ func (h *PlaybackHandler) prepareLocalTransportV3(r *http.Request, session *play
 	sourceProfile, sourceBitDepth := sourceVideoTranscodeFactsV3(file, result)
 	seekSeconds, startSegment := configureHLSTimelineV3(result.Plan, videoCodec, 2, sourceMetadata.DurationSeconds)
 	unlock := h.tm.LockSessionLifecycle(session.ID)
-	opts := playback.TranscodeOpts{InputPath: file.FilePath, OutputDir: outputDir, OutputSubdir: outputSubdir, SessionID: session.ID, SourceVideoCodec: sourceMetadata.VideoCodec, SourceVideoProfile: sourceProfile, SourceVideoBitDepth: sourceBitDepth, SoftwareVideoDecode: sourceMetadata.SoftwareVideoDecode, VideoBitstreamFilter: videoBitstreamFilterForPlanV3(result.Plan), SeekSeconds: seekSeconds, StartSegmentNumber: startSegment, TargetResolution: result.TargetResolution, TargetCodecVideo: videoCodec, TargetCodecAudio: result.TargetAudioCodec, TargetAudioChannels: result.TargetAudioChannels, TargetAudioBitrateKbps: result.TargetAudioBitrateKbps, TargetBitrateKbps: result.TargetBitrateKbps, SegmentDuration: 2, FFmpegPath: cfg.FFmpegPath, HWAccel: cfg.HWAccel, HWDevice: cfg.HWDevice, AudioTrackIndex: plannedAudioTrackIndexV3(result, session.AudioTrackIndex), SubtitleTrackIndex: result.SubtitleTransportTrackIndex, SubtitleBurnIn: result.SubtitleBurnIn, SubtitleCodec: result.SubtitleCodec, TotalDuration: sourceMetadata.DurationSeconds, FastStart: true, NodeType: playbackNodeIntegratedV3, ExecutionMode: playbackNodeIntegratedV3, FFmpegLogSink: h.FFmpegLogSink}
+	opts := playback.TranscodeOpts{InputPath: file.FilePath, OutputDir: outputDir, OutputSubdir: outputSubdir, SessionID: session.ID, SourceVideoCodec: sourceMetadata.VideoCodec, SourceVideoProfile: sourceProfile, SourceVideoBitDepth: sourceBitDepth, SoftwareVideoDecode: sourceMetadata.SoftwareVideoDecode, VideoBitstreamFilter: videoBitstreamFilterForPlanV3(result.Plan), SeekSeconds: seekSeconds, StartSegmentNumber: startSegment, TargetResolution: result.TargetResolution, TargetCodecVideo: videoCodec, TargetCodecAudio: result.TargetAudioCodec, TargetAudioChannels: result.TargetAudioChannels, TargetAudioBitrateKbps: result.TargetAudioBitrateKbps, TargetBitrateKbps: result.TargetBitrateKbps, SegmentDuration: 2, SegmentRetentionSeconds: cfg.SegmentRetentionSeconds, FFmpegPath: cfg.FFmpegPath, HWAccel: cfg.HWAccel, HWDevice: cfg.HWDevice, AudioTrackIndex: plannedAudioTrackIndexV3(result, session.AudioTrackIndex), SubtitleTrackIndex: result.SubtitleTransportTrackIndex, SubtitleBurnIn: result.SubtitleBurnIn, SubtitleCodec: result.SubtitleCodec, TotalDuration: sourceMetadata.DurationSeconds, FastStart: true, NodeType: playbackNodeIntegratedV3, ExecutionMode: playbackNodeIntegratedV3, FFmpegLogSink: h.FFmpegLogSink}
 	ts, err := h.startLocalPlaybackTransport(r.Context(), opts)
 	if err != nil {
 		unlock()
@@ -13,8 +13,11 @@ import (
 	"github.com/robfig/cron/v3"
 )

-const cloudflareURLMode = "cloudflare_token"
-const chapterThumbnailSoftwareToneMapKey = "playback.chapter_thumbnail_software_tone_map_enabled"
+const (
+	cloudflareURLMode                  = "cloudflare_token"
+	playbackSegmentRetentionSettingKey = "playback.segment_retention_seconds"
+	chapterThumbnailSoftwareToneMapKey = "playback.chapter_thumbnail_software_tone_map_enabled"
+)

 // adminSettingDefaults is the effective value shown by the Admin UI when no
 // row exists in server_settings. Keep these values aligned with the runtime
@@ -51,6 +54,7 @@ var adminSettingDefaults = map[string]string{

 	"playback.ffmpeg_path":                     "/usr/lib/jellyfin-ffmpeg/ffmpeg",
 	"playback.transcode_dir":                   DefaultTranscodeDir,
+	playbackSegmentRetentionSettingKey:         "600",
 	"playback.hw_accel":                        "auto",
 	"playback.transcode_enabled":               "true",
 	"playback.local_transcode_fallback":        "true",
@@ -301,6 +305,16 @@ func NormalizeAdminSetting(key, raw string) (string, error) {
 		return normalizeAdminInt(key, value, 1, 99)
 	case "transcode_throttle_seconds":
 		return normalizeAdminInt(key, value, 60, 86400)
+	case playbackSegmentRetentionSettingKey:
+		normalized, err := normalizeAdminInt(key, value, 0, 86400)
+		if err != nil {
+			return "", err
+		}
+		seconds, _ := strconv.Atoi(normalized)
+		if seconds != 0 && seconds < 120 {
+			return "", fmt.Errorf("%s must be 0 or between 120 and 86400", key)
+		}
+		return normalized, nil
 	case "ai.max_concurrent_jobs", "subtitle_ai.max_concurrent_jobs":
 		return normalizeAdminInt(key, value, 1, 1024)
 	case "subtitle_ai.batch_size":
@@ -225,6 +225,7 @@ func TestNormalizeAdminSettingRejectsInvalidValues(t *testing.T) {
 		{key: "theme.catalog_url", value: "http://raw.githubusercontent.com/Silo-Server/silo-themes/main/catalog.json"},
 		{key: "theme.catalog_url", value: "https://example.com/catalog.json"},
 		{key: "redis.url", value: "not-a-url"},
+		{key: "playback.segment_retention_seconds", value: "119"},
 	}
 	for _, tc := range tests {
 		t.Run(tc.key, func(t *testing.T) {
@@ -235,6 +236,20 @@ func TestNormalizeAdminSettingRejectsInvalidValues(t *testing.T) {
 	}
 }

+func TestNormalizeAdminSettingAcceptsSegmentRetentionBounds(t *testing.T) {
+	for _, value := range []string{"0", "120", "86400"} {
+		t.Run(value, func(t *testing.T) {
+			got, err := NormalizeAdminSetting("playback.segment_retention_seconds", value)
+			if err != nil {
+				t.Fatalf("NormalizeAdminSetting: %v", err)
+			}
+			if got != value {
+				t.Fatalf("normalized retention = %q, want %q", got, value)
+			}
+		})
+	}
+}
+
 func TestNormalizeAdminSettingAcceptsApprovedThemeCatalogURL(t *testing.T) {
 	got, err := NormalizeAdminSetting(
 		"theme.catalog_url",
@@ -157,9 +157,10 @@ func (c MatcherConfig) TVSeriesRootQueueEnabled() bool {

 // PlaybackConfig holds transcoding and playback settings.
 type PlaybackConfig struct {
-	FFmpegPath   string `yaml:"ffmpeg_path"`
-	TranscodeDir string `yaml:"transcode_dir"`
-	HWAccel      string `yaml:"hw_accel"`
+	FFmpegPath              string `yaml:"ffmpeg_path"`
+	TranscodeDir            string `yaml:"transcode_dir"`
+	SegmentRetentionSeconds int    `yaml:"segment_retention_seconds"`
+	HWAccel                 string `yaml:"hw_accel"`
 	// HWDevice is the GPU render device for hardware transcodes. A single
 	// path pins every GPU workload to that device; a comma-separated list
 	// (e.g. "/dev/dri/renderD128,/dev/dri/renderD129") balances workloads
@@ -479,6 +480,7 @@ func setDefaults() *configRaw {
 		Playback: PlaybackConfig{
 			FFmpegPath:                   "/usr/lib/jellyfin-ffmpeg/ffmpeg",
 			TranscodeDir:                 DefaultTranscodeDir,
+			SegmentRetentionSeconds:      600,
 			HWAccel:                      "auto",
 			ChapterThumbnailWorkers:      1,
 			ChapterThumbnailExecution:    "local",
@@ -317,6 +317,14 @@ func LoadFromDB(m map[string]string) (*Config, error) {
 	// Playback
 	cfg.Playback.FFmpegPath = stringOr(m, "playback.ffmpeg_path", "/usr/lib/jellyfin-ffmpeg/ffmpeg")
 	cfg.Playback.TranscodeDir = stringOr(m, "playback.transcode_dir", DefaultTranscodeDir)
+	segmentRetentionSeconds, err := intOr(m, playbackSegmentRetentionSettingKey, 600)
+	if err != nil {
+		return nil, err
+	}
+	if segmentRetentionSeconds != 0 && (segmentRetentionSeconds < 120 || segmentRetentionSeconds > 86400) {
+		return nil, fmt.Errorf("%s must be 0 or between 120 and 86400", playbackSegmentRetentionSettingKey)
+	}
+	cfg.Playback.SegmentRetentionSeconds = segmentRetentionSeconds
 	cfg.Playback.HWAccel = stringOr(m, "playback.hw_accel", "auto")
 	cfg.Playback.HWDevice = stringOr(m, "playback.hw_device", "")
 	chapterThumbnailWorkers, err := intOr(m, "playback.chapter_thumbnail_workers", 1)
@@ -35,6 +35,17 @@ func TestLoadFromDBMetadataPresignExpiryRejectsInvalidDuration(t *testing.T) {
 	}
 }

+func TestLoadFromDBRejectsInvalidSegmentRetention(t *testing.T) {
+	for _, value := range []string{"-1", "119", "86401"} {
+		t.Run(value, func(t *testing.T) {
+			_, err := LoadFromDB(map[string]string{playbackSegmentRetentionSettingKey: value})
+			if err == nil || !strings.Contains(err.Error(), playbackSegmentRetentionSettingKey) {
+				t.Fatalf("LoadFromDB() error = %v, want retention bounds error", err)
+			}
+		})
+	}
+}
+
 func TestLoadFromDBJellyfinWebEnabledDefaultsToTrue(t *testing.T) {
 	cfg, err := LoadFromDB(map[string]string{})
 	if err != nil {
@@ -186,6 +197,16 @@ func TestYAMLToSettingsMapJellyfinCompatEnabledDefaultsToLegacyListener(t *testi
 	}
 }

+func TestYAMLToSettingsMapPreservesExplicitlyDisabledSegmentRetention(t *testing.T) {
+	m := yamlSettingsMapFromString(t, `
+playback:
+  segment_retention_seconds: 0
+`)
+	if got := m[playbackSegmentRetentionSettingKey]; got != "0" {
+		t.Fatalf("segment retention = %q, want explicit disable", got)
+	}
+}
+
 func yamlSettingsMapFromString(t *testing.T, body string) map[string]string {
 	t.Helper()
 	path := t.TempDir() + "/silo.yaml"
@@ -209,6 +209,7 @@ func YAMLToSettingsMap(path string) (map[string]string, error) {
 	// Playback
 	setIfNonEmpty(m, "playback.ffmpeg_path", raw.Playback.FFmpegPath)
 	setIfNonEmpty(m, "playback.transcode_dir", raw.Playback.TranscodeDir)
+	m[playbackSegmentRetentionSettingKey] = strconv.Itoa(raw.Playback.SegmentRetentionSeconds)
 	setIfNonEmpty(m, "playback.hw_accel", raw.Playback.HWAccel)
 	if raw.Playback.ChapterThumbnailWorkers != 0 {
 		m["playback.chapter_thumbnail_workers"] = strconv.Itoa(raw.Playback.ChapterThumbnailWorkers)
@@ -12,6 +12,7 @@ package httpstream
 import (
 	"context"
 	"errors"
+	"fmt"
 	"io"
 	"net"
 	"net/http"
@@ -171,6 +172,26 @@ func (s *RollingDeadlineWriter) BytesWritten() int64 {
 	return s.bytesWritten
 }

+// CompletedFullResponse reports whether the complete representation was
+// accepted by the transport without a write error. The request context is not
+// consulted because net/http may cancel it after the response reaches the
+// client but before the handler performs post-response accounting. ServeContent
+// returns 206 for a whole-file Range request such as "bytes=0-", so byte count
+// alone is not enough to distinguish that transfer from a partial range.
+func (s *RollingDeadlineWriter) CompletedFullResponse(fullSize int64) bool {
+	if fullSize <= 0 || s.bytesWritten != fullSize || s.firstWriteErr != nil {
+		return false
+	}
+	switch s.statusCode {
+	case http.StatusOK:
+		return true
+	case http.StatusPartialContent:
+		return s.Header().Get("Content-Range") == fmt.Sprintf("bytes 0-%d/%d", fullSize-1, fullSize)
+	default:
+		return false
+	}
+}
+
 // Outcome classifies the first write failure, or a canceled request when no
 // write failure was surfaced by the transport.
 func (s *RollingDeadlineWriter) Outcome(ctx context.Context) StreamOutcome {
@@ -271,6 +271,42 @@ func TestServeContentReadFromOutcomeCompletedAndCounted(t *testing.T) {
 	}
 }

+func TestCompletedFullResponse(t *testing.T) {
+	const body = "complete segment"
+	tests := []struct {
+		name             string
+		rangeHeader      string
+		method           string
+		cancelAfterWrite bool
+		want             bool
+	}{
+		{name: "ordinary get", method: http.MethodGet, want: true},
+		{name: "completed get with canceled context", method: http.MethodGet, cancelAfterWrite: true, want: true},
+		{name: "open ended full range", method: http.MethodGet, rangeHeader: "bytes=0-", want: true},
+		{name: "explicit full range", method: http.MethodGet, rangeHeader: "bytes=0-15", want: true},
+		{name: "partial range", method: http.MethodGet, rangeHeader: "bytes=1-", want: false},
+		{name: "head", method: http.MethodHead, want: false},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			rr := httptest.NewRecorder()
+			sw := newRollingDeadlineWriter(rr, time.Second, 0)
+			ctx, cancel := context.WithCancel(context.Background())
+			t.Cleanup(cancel)
+			req := httptest.NewRequest(tt.method, "/segment.ts", nil).WithContext(ctx)
+			req.Header.Set("Range", tt.rangeHeader)
+			http.ServeContent(sw, req, "segment.ts", time.Time{}, strings.NewReader(body))
+			if tt.cancelAfterWrite {
+				cancel()
+			}
+			if got := sw.CompletedFullResponse(int64(len(body))); got != tt.want {
+				t.Fatalf("CompletedFullResponse = %v, want %v (status=%d range=%q bytes=%d)",
+					got, tt.want, sw.StatusCode(), rr.Header().Get("Content-Range"), sw.BytesWritten())
+			}
+		})
+	}
+}
+
 // TestReadFromPreservesCompletion exercises the io.ReaderFrom path used by
 // http.ServeContent (sendfile) under a server WriteTimeout shorter than the
 // transfer, with a source large enough to require multiple bounded slices.
@@ -184,6 +184,7 @@ type PlaybackHandler struct {
 	FFmpegPath              string
 	HWAccel                 string
 	TranscodeDir            string
+	SegmentRetentionSeconds func() int
 	// tm is the shared transcode-session lifecycle (live map, reconstruct) — the
 	// same type the native handler uses, so jellycompat gets the reconstruct cap
 	// and node-affinity rule for free. The reconstruction recipe is carried in the
@@ -268,36 +269,40 @@ func NewPlaybackHandler(
 	transcodeDir := filepath.Join(os.TempDir(), "silo-transcode")
 	ffmpegPath := ""
 	hwAccel := ""
+	segmentRetentionSeconds := 600
 	if cfg != nil {
 		if cfg.Playback.TranscodeDir != "" {
 			transcodeDir = cfg.Playback.TranscodeDir
 		}
 		ffmpegPath = cfg.Playback.FFmpegPath
 		hwAccel = cfg.Playback.HWAccel
+		segmentRetentionSeconds = cfg.Playback.SegmentRetentionSeconds
 	}

 	h := &PlaybackHandler{
-		cfg:            cfg,
-		content:        content,
-		codec:          codec,
-		deviceProfiles: deviceProfiles,
-		playbackStore:  playbackStore,
-		sessionMgr:     sessionMgr,
-		fileResolver:   fileResolver,
-		storeProvider:  storeProvider,
-		FFmpegPath:     ffmpegPath,
-		HWAccel:        hwAccel,
-		TranscodeDir:   transcodeDir,
-		tm:             playback.NewTranscodeManager(),
+		cfg:                     cfg,
+		content:                 content,
+		codec:                   codec,
+		deviceProfiles:          deviceProfiles,
+		playbackStore:           playbackStore,
+		sessionMgr:              sessionMgr,
+		fileResolver:            fileResolver,
+		storeProvider:           storeProvider,
+		FFmpegPath:              ffmpegPath,
+		HWAccel:                 hwAccel,
+		TranscodeDir:            transcodeDir,
+		SegmentRetentionSeconds: func() int { return segmentRetentionSeconds },
+		tm:                      playback.NewTranscodeManager(),
 	}
 	// Wire the shared transcode manager with closures so it reads the handler's
 	// (late-set) JWTSecret lazily, matching the native handler.
 	h.tm.JWTSecretFn = func() string { return h.JWTSecret }
 	h.tm.Config = func() playback.TranscodeRuntimeConfig {
 		return playback.TranscodeRuntimeConfig{
-			TranscodeDir: h.TranscodeDir,
-			FFmpegPath:   h.FFmpegPath,
-			HWAccel:      h.HWAccel,
+			TranscodeDir:            h.TranscodeDir,
+			FFmpegPath:              h.FFmpegPath,
+			HWAccel:                 h.HWAccel,
+			SegmentRetentionSeconds: h.segmentRetentionSeconds(),
 		}
 	}
 	if reg, ok := sessionMgr.(interface {
@@ -337,6 +342,13 @@ func NewPlaybackHandler(
 	return h
 }

+func (h *PlaybackHandler) segmentRetentionSeconds() int {
+	if h.SegmentRetentionSeconds != nil {
+		return h.SegmentRetentionSeconds()
+	}
+	return 600
+}
+
 // CleanupOrphanedTranscodes removes stale per-session transcode dirs, sparing
 // those whose recipe card still exists. Delegates to the shared manager.
 func (h *PlaybackHandler) CleanupOrphanedTranscodes() (int, error) {
@@ -0,0 +1,22 @@
+package jellycompat
+
+import (
+	"testing"
+
+	"github.com/Silo-Server/silo-server/internal/config"
+)
+
+func TestPlaybackHandlerReadsLiveSegmentRetention(t *testing.T) {
+	cfg := &config.Config{Playback: config.PlaybackConfig{SegmentRetentionSeconds: 300}}
+	handler := NewPlaybackHandler(cfg, nil, nil, nil, nil, nil, nil, nil)
+
+	if got := handler.tm.Config().SegmentRetentionSeconds; got != 300 {
+		t.Fatalf("configured segment retention = %d, want 300", got)
+	}
+	liveRetention := 300
+	handler.SegmentRetentionSeconds = func() int { return liveRetention }
+	liveRetention = 120
+	if got := handler.tm.Config().SegmentRetentionSeconds; got != 120 {
+		t.Fatalf("reloaded segment retention = %d, want 120", got)
+	}
+}
@@ -103,6 +103,13 @@ func NewRouter(deps Dependencies) chi.Router {
 	}
 	userDataHandler := NewUserDataHandler(deps.ContentService, deps.UserDataService, deps.IDCodec, deps.Config)
 	playbackHandler := NewPlaybackHandler(deps.Config, deps.ContentService, deps.IDCodec, deps.DeviceProfiles, deps.PlaybackStore, deps.SessionMgr, deps.FileResolver, deps.UserStoreProvider)
+	startupSegmentRetention := playbackHandler.SegmentRetentionSeconds
+	playbackHandler.SegmentRetentionSeconds = func() int {
+		if cfg := deps.CurrentConfig(); cfg != nil {
+			return cfg.Playback.SegmentRetentionSeconds
+		}
+		return startupSegmentRetention()
+	}
 	if deps.DB != nil {
 		playbackHandler.profileStaler = recommendations.NewRepo(deps.DB)
 	}
@@ -21,6 +21,7 @@ import (

 	"github.com/go-chi/chi/v5"

+	"github.com/Silo-Server/silo-server/internal/httpstream"
 	"github.com/Silo-Server/silo-server/internal/models"
 	"github.com/Silo-Server/silo-server/internal/nodepool"
 	"github.com/Silo-Server/silo-server/internal/playback"
@@ -418,8 +419,8 @@ func (h *PlaybackHandler) HandleHLSSegment(w http.ResponseWriter, r *http.Reques
 		}
 	}

-	segmentFile := name + "." + ext
-	segmentPath, err := transcodeSession.GetSegment(segmentFile)
+	segmentName := name + "." + ext
+	segmentLease, err := transcodeSession.OpenSegment(segmentName)
 	if err != nil && errors.Is(err, playback.ErrSegmentNotFound) {
 		segNum, parseErr := playback.ParseSegmentNumber(name)
 		if parseErr == nil {
@@ -430,7 +431,7 @@ func (h *PlaybackHandler) HandleHLSSegment(w http.ResponseWriter, r *http.Reques
 				lastProducedAgeMS = now.Sub(decision.Progress.LastProducedAt).Milliseconds()
 			}
 			slog.InfoContext(r.Context(), "transcode segment missing", "component", "jellycompat",
-				"segment", segmentFile,
+				"segment", segmentName,
 				"requested_segment", segNum,
 				"produced_head", decision.Progress.ProducedHead,
 				"last_requested_segment", decision.Progress.LastRequestedSegment,
@@ -444,7 +445,7 @@ func (h *PlaybackHandler) HandleHLSSegment(w http.ResponseWriter, r *http.Reques
 			)
 			if decision.Wait {
 				slog.InfoContext(r.Context(), "transcode segment wait", "component", "jellycompat",
-					"segment", segmentFile,
+					"segment", segmentName,
 					"requested_segment", segNum,
 					"produced_head", decision.Progress.ProducedHead,
 					"last_requested_segment", decision.Progress.LastRequestedSegment,
@@ -456,10 +457,10 @@ func (h *PlaybackHandler) HandleHLSSegment(w http.ResponseWriter, r *http.Reques
 					"session", playSession.UpstreamSessionID,
 					"playback_session_id", playSession.UpstreamSessionID,
 				)
-				segmentPath, err = transcodeSession.WaitForSegment(segmentFile, decision.WaitTimeout)
+				segmentLease, err = transcodeSession.WaitForOpenSegment(segmentName, decision.WaitTimeout)
 				if err != nil && errors.Is(err, playback.ErrSegmentNotFound) {
 					slog.InfoContext(r.Context(), "transcode segment wait timeout", "component", "jellycompat",
-						"segment", segmentFile,
+						"segment", segmentName,
 						"requested_segment", segNum,
 						"produced_head", decision.Progress.ProducedHead,
 						"last_requested_segment", decision.Progress.LastRequestedSegment,
@@ -479,7 +480,7 @@ func (h *PlaybackHandler) HandleHLSSegment(w http.ResponseWriter, r *http.Reques
 				if seekErr != nil && !errors.Is(seekErr, playback.ErrManifestNotReady) {
 					slog.ErrorContext(r.Context(), "resolve transcode seek target", "component", "jellycompat",
 						"error", seekErr,
-						"segment", segmentFile,
+						"segment", segmentName,
 						"play_session", playSessionID,
 						"session", playSession.UpstreamSessionID,
 						"playback_session_id", playSession.UpstreamSessionID,
@@ -498,7 +499,7 @@ func (h *PlaybackHandler) HandleHLSSegment(w http.ResponseWriter, r *http.Reques

 				if ok {
 					slog.InfoContext(r.Context(), "transcode seek restart", "component", "jellycompat",
-						"segment", segmentFile,
+						"segment", segmentName,
 						"requested_segment", segNum,
 						"produced_head", decision.Progress.ProducedHead,
 						"last_requested_segment", decision.Progress.LastRequestedSegment,
@@ -519,14 +520,14 @@ func (h *PlaybackHandler) HandleHLSSegment(w http.ResponseWriter, r *http.Reques
 						seekSeconds,
 						segNum,
 					); restartErr == nil {
-						segmentPath, err = transcodeSession.WaitForSegment(segmentFile, 30*time.Second)
+						segmentLease, err = transcodeSession.WaitForOpenSegment(segmentName, 30*time.Second)
 					}
 				}
 			}
 		} else if transcodeSession.IsRunning() {
 			// Non-numbered segment (e.g. init.mp4 for fMP4 HLS).
 			// Wait briefly — the init segment is written almost immediately.
-			segmentPath, err = transcodeSession.WaitForSegment(segmentFile, 10*time.Second)
+			segmentLease, err = transcodeSession.WaitForOpenSegment(segmentName, 10*time.Second)
 		}
 	}
 	if err != nil {
@@ -535,11 +536,15 @@ func (h *PlaybackHandler) HandleHLSSegment(w http.ResponseWriter, r *http.Reques
 		return
 	}

-	if segNum, parseErr := playback.ParseSegmentNumber(name); parseErr == nil {
-		transcodeSession.ReportSegmentDownloaded(segNum)
+	defer func() { _ = segmentLease.Close() }()
+	sw := httpstream.NewRollingDeadlineWriter(w)
+	http.ServeContent(sw, r, segmentLease.Info.Name(), segmentLease.Info.ModTime(), segmentLease.File)
+	if r.Method == http.MethodGet &&
+		sw.CompletedFullResponse(segmentLease.Info.Size()) {
+		if segNum, parseErr := playback.ParseSegmentNumber(name); parseErr == nil {
+			transcodeSession.ReportSegmentDownloadedForGeneration(segNum, segmentLease.Generation)
+		}
 	}
-
-	http.ServeFile(w, r, segmentPath)
 }

 // hlsSegmentErrorResponse maps a segment-retrieval error to a Jellyfin-faithful
@@ -1669,6 +1674,7 @@ func (h *PlaybackHandler) ensureTranscodeSession(ctx context.Context, playSessio
 		TotalDuration:       float64(source.Version.Duration),
 		FastStart:           true,
 	}
+	opts.SegmentRetentionSeconds = h.segmentRetentionSeconds()
 	if source.TranscodeAudio {
 		opts.TargetCodecVideo = "copy"
 	}
@@ -0,0 +1,345 @@
+package playback
+
+import (
+	"errors"
+	"fmt"
+	"log/slog"
+	"os"
+	"path/filepath"
+	"strings"
+	"time"
+)
+
+const (
+	segmentPruneHysteresis = 5
+	segmentPruneBatchSize  = 512
+	segmentPruneRetryDelay = 5 * time.Second
+)
+
+type segmentPruneCandidate struct {
+	number int
+	path   string
+}
+
+// scheduleSegmentPruneLocked starts one asynchronous prune pass after the
+// client has advanced far enough to make useful work. The caller must hold
+// s.mu. Segment retention is measured in media time so custom HLS segment
+// durations retain the same back-seek window.
+func (s *TranscodeSession) scheduleSegmentPruneLocked() {
+	retentionSeconds := s.opts.SegmentRetentionSeconds
+	if retentionSeconds <= 0 || s.segmentPruneRunning || s.restarting {
+		return
+	}
+
+	if strings.EqualFold(s.opts.TargetCodecVideo, "copy") {
+		// Copy-mode fragments follow source keyframes, so their real media-time
+		// floor is resolved asynchronously from EXTINF durations below. Use the
+		// download high-water mark only to avoid reparsing the manifest for every
+		// segment response.
+		if s.lastCompletedSegment-s.lastPruneHighWater < segmentPruneHysteresis {
+			return
+		}
+	} else {
+		segmentDuration := s.opts.SegmentDuration
+		if segmentDuration <= 0 {
+			segmentDuration = defaultSegmentDuration
+		}
+		retainedSegments := (retentionSeconds + segmentDuration - 1) / segmentDuration
+		floor := s.lastCompletedSegment - retainedSegments
+		beforeStartReady := s.pruneBeforeStart && floor == s.opts.StartSegmentNumber
+		if floor < s.opts.StartSegmentNumber || (!beforeStartReady && floor == s.opts.StartSegmentNumber) || floor-s.lastPruneFloor < segmentPruneHysteresis {
+			return
+		}
+	}
+
+	s.segmentPruneRunning = true
+	generation := s.segmentGeneration
+	downloadedThrough := s.lastCompletedSegment
+	s.lastPruneHighWater = downloadedThrough
+	go s.pruneDownloadedSegments(generation, downloadedThrough, false)
+}
+
+// pruneDownloadedSegments removes completed media files strictly behind floor.
+// It visits each newly expired segment number at most once in bounded batches,
+// avoiding repeated full-directory scans when FFmpeg has generated far ahead
+// of the client. The current process's startup window remains present so real
+// manifest reloads cannot get stuck behind startupFilesReady after cleanup.
+func (s *TranscodeSession) pruneDownloadedSegments(generation uint64, downloadedThrough int, continuation bool) {
+	started := time.Now()
+	s.mu.Lock()
+	if generation != s.segmentGeneration || s.restarting {
+		s.mu.Unlock()
+		return
+	}
+	opts := s.opts
+	outputDir := s.outputDir
+	fromFloor := s.lastPruneFloor
+	pruneBeforeStart := s.pruneBeforeStart
+	s.mu.Unlock()
+
+	floor, complete, err := segmentRetentionFloor(outputDir, opts, downloadedThrough)
+	if err != nil {
+		slog.Warn("resolve transcode segment retention floor", "component", "playback", "error", err, "session", opts.SessionID, "playback_session_id", opts.SessionID)
+		s.finishSegmentPrune(generation, fromFloor, fromFloor, downloadedThrough, segmentPruneRetryDelay)
+		return
+	}
+	beforeStartReady := pruneBeforeStart && floor == opts.StartSegmentNumber
+	if !complete || floor < opts.StartSegmentNumber || (!beforeStartReady && floor == opts.StartSegmentNumber) || (!continuation && floor-fromFloor < segmentPruneHysteresis) {
+		s.finishSegmentPruneAttempt(generation)
+		return
+	}
+
+	segmentDuration := opts.SegmentDuration
+	if segmentDuration <= 0 {
+		segmentDuration = defaultSegmentDuration
+	}
+	freshGuard := 2*time.Duration(segmentDuration)*time.Second + 30*time.Second
+	startupEnd := opts.StartSegmentNumber + startupSegmentRequirement(opts)
+	targetFloor := floor
+	fromSegment := max(fromFloor, startupEnd)
+	if pruneBeforeStart {
+		// A forward restart intentionally keeps the previous generation's back
+		// buffer until the replacement has produced its own. Retire that older
+		// range first, including its startup files, then resume normal pruning
+		// without touching the replacement process's startup window. The old
+		// range can be very sparse after a long seek, so enumerate actual files
+		// below the new start rather than walking every intervening number.
+		targetFloor = min(floor, opts.StartSegmentNumber)
+	}
+	if !pruneBeforeStart && fromSegment >= targetFloor {
+		s.finishSegmentPrune(generation, targetFloor, targetFloor, downloadedThrough, 0)
+		return
+	}
+
+	if _, err := os.Stat(outputDir); err != nil {
+		if errors.Is(err, os.ErrNotExist) {
+			s.finishSegmentPrune(generation, targetFloor, targetFloor, downloadedThrough, 0)
+			return
+		}
+		slog.Warn("stat transcode directory for pruning", "component", "playback", "error", err, "session", opts.SessionID, "playback_session_id", opts.SessionID)
+		retryFloor := fromSegment
+		if pruneBeforeStart {
+			retryFloor = fromFloor
+		}
+		s.finishSegmentPrune(generation, retryFloor, targetFloor, downloadedThrough, segmentPruneRetryDelay)
+		return
+	}
+
+	var candidates []segmentPruneCandidate
+	moreCandidates := false
+	if pruneBeforeStart {
+		entries, readErr := os.ReadDir(outputDir)
+		if readErr != nil {
+			slog.Warn("read transcode directory for preserved segment pruning", "component", "playback", "error", readErr, "session", opts.SessionID, "playback_session_id", opts.SessionID)
+			s.finishSegmentPrune(generation, fromFloor, targetFloor, downloadedThrough, segmentPruneRetryDelay)
+			return
+		}
+		for _, entry := range entries {
+			segment, parseErr := ParseSegmentNumber(entry.Name())
+			if entry.IsDir() || parseErr != nil || segment >= targetFloor || entry.Name() != segmentFilename(segment, opts) {
+				continue
+			}
+			if len(candidates) == segmentPruneBatchSize {
+				moreCandidates = true
+				break
+			}
+			candidates = append(candidates, segmentPruneCandidate{
+				number: segment,
+				path:   filepath.Join(outputDir, entry.Name()),
+			})
+		}
+	} else {
+		toSegment := min(targetFloor, fromSegment+segmentPruneBatchSize)
+		for segment := fromSegment; segment < toSegment; segment++ {
+			candidates = append(candidates, segmentPruneCandidate{
+				number: segment,
+				path:   filepath.Join(outputDir, segmentFilename(segment, opts)),
+			})
+		}
+		moreCandidates = toSegment < targetFloor
+	}
+	if len(candidates) == 0 && !moreCandidates {
+		s.finishSegmentPrune(generation, targetFloor, targetFloor, downloadedThrough, 0)
+		return
+	}
+
+	processedFloor := targetFloor
+	if !pruneBeforeStart {
+		processedFloor = candidates[len(candidates)-1].number + 1
+	}
+	var retryAfter time.Duration
+	removed := 0
+	var freedBytes int64
+	for _, candidate := range candidates {
+		info, err := os.Stat(candidate.path)
+		if err != nil {
+			if errors.Is(err, os.ErrNotExist) {
+				continue
+			}
+			if pruneBeforeStart {
+				retryAfter = max(retryAfter, segmentPruneRetryDelay)
+			} else {
+				processedFloor, retryAfter = earlierPruneRetry(processedFloor, retryAfter, candidate.number, segmentPruneRetryDelay)
+			}
+			slog.Warn("stat downloaded transcode segment", "component", "playback", "error", err, "segment", candidate.number, "session", opts.SessionID, "playback_session_id", opts.SessionID)
+			continue
+		}
+		freshUntil := info.ModTime().Add(freshGuard)
+		if delay := time.Until(freshUntil); delay > 0 {
+			if pruneBeforeStart {
+				retryAfter = max(retryAfter, delay)
+			} else {
+				processedFloor, retryAfter = earlierPruneRetry(processedFloor, retryAfter, candidate.number, delay)
+			}
+			continue
+		}
+
+		// Serialize the unlink with restart and shutdown generation changes.
+		// Once either advances the generation, this pass cannot delete files
+		// generated for a replacement timeline or session object.
+		s.mu.Lock()
+		if generation != s.segmentGeneration || s.restarting {
+			s.mu.Unlock()
+			return
+		}
+		removeErr := os.Remove(candidate.path)
+		s.mu.Unlock()
+		if removeErr != nil {
+			if !errors.Is(removeErr, os.ErrNotExist) {
+				if pruneBeforeStart {
+					retryAfter = max(retryAfter, segmentPruneRetryDelay)
+				} else {
+					processedFloor, retryAfter = earlierPruneRetry(processedFloor, retryAfter, candidate.number, segmentPruneRetryDelay)
+				}
+				slog.Warn("remove downloaded transcode segment", "component", "playback", "error", removeErr, "segment", candidate.number, "session", opts.SessionID, "playback_session_id", opts.SessionID)
+			}
+			continue
+		}
+		removed++
+		freedBytes += info.Size()
+	}
+	if pruneBeforeStart && (moreCandidates || retryAfter > 0) {
+		processedFloor = fromFloor
+	}
+
+	s.finishSegmentPrune(generation, processedFloor, targetFloor, downloadedThrough, retryAfter)
+	if removed > 0 {
+		slog.Info("pruned downloaded transcode segments",
+			"component", "playback",
+			"count", removed,
+			"freed_bytes", freedBytes,
+			"floor_segment", processedFloor,
+			"duration_ms", time.Since(started).Milliseconds(),
+			"session", opts.SessionID,
+			"playback_session_id", opts.SessionID,
+		)
+	}
+}
+
+// segmentRetentionFloor returns the first segment that must remain to preserve
+// the configured media-time window behind downloadedThrough. Encoded HLS uses
+// fixed-duration fragments. Copy HLS follows source keyframes, so its floor is
+// derived from the current manifest's actual EXTINF durations. complete is
+// false when the manifest does not yet cover the full requested back buffer.
+func segmentRetentionFloor(outputDir string, opts TranscodeOpts, downloadedThrough int) (floor int, complete bool, err error) {
+	retentionSeconds := opts.SegmentRetentionSeconds
+	if retentionSeconds <= 0 {
+		return 0, false, nil
+	}
+
+	if !strings.EqualFold(opts.TargetCodecVideo, "copy") {
+		segmentDuration := opts.SegmentDuration
+		if segmentDuration <= 0 {
+			segmentDuration = defaultSegmentDuration
+		}
+		retainedSegments := (retentionSeconds + segmentDuration - 1) / segmentDuration
+		return downloadedThrough - retainedSegments, true, nil
+	}
+
+	manifest, err := os.ReadFile(filepath.Join(outputDir, "stream.m3u8"))
+	if err != nil {
+		return 0, false, fmt.Errorf("read copy manifest: %w", err)
+	}
+	timeline, err := parseManifestTimeline(manifest)
+	if err != nil {
+		return 0, false, fmt.Errorf("parse copy manifest: %w", err)
+	}
+
+	downloadedIndex := -1
+	for i, entry := range timeline.entries {
+		if entry.duration <= 0 {
+			return 0, false, fmt.Errorf("copy segment %d has non-positive duration %.6f", entry.number, entry.duration)
+		}
+		if entry.number == downloadedThrough {
+			downloadedIndex = i
+		}
+	}
+	if downloadedIndex < 0 {
+		return 0, false, nil
+	}
+
+	retainedSeconds := 0.0
+	for i := downloadedIndex - 1; i >= 0; i-- {
+		floor = timeline.entries[i].number
+		retainedSeconds += timeline.entries[i].duration
+		if retainedSeconds >= float64(retentionSeconds) {
+			return floor, true, nil
+		}
+	}
+	return 0, false, nil
+}
+
+func earlierPruneRetry(currentFloor int, currentDelay time.Duration, segment int, delay time.Duration) (int, time.Duration) {
+	if segment < currentFloor {
+		return segment, max(delay, time.Millisecond)
+	}
+	return currentFloor, currentDelay
+}
+
+func (s *TranscodeSession) finishSegmentPruneAttempt(generation uint64) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if generation != s.segmentGeneration {
+		return
+	}
+	s.segmentPruneRunning = false
+	s.scheduleSegmentPruneLocked()
+}
+
+func (s *TranscodeSession) finishSegmentPrune(
+	generation uint64,
+	processedFloor int,
+	targetFloor int,
+	downloadedThrough int,
+	retryAfter time.Duration,
+) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if generation != s.segmentGeneration {
+		return
+	}
+	if processedFloor > s.lastPruneFloor {
+		s.lastPruneFloor = processedFloor
+	}
+	if s.pruneBeforeStart && s.lastPruneFloor >= s.opts.StartSegmentNumber {
+		s.pruneBeforeStart = false
+	}
+	if retryAfter > 0 {
+		time.AfterFunc(retryAfter, func() {
+			s.mu.Lock()
+			if generation != s.segmentGeneration || s.restarting {
+				s.mu.Unlock()
+				return
+			}
+			s.mu.Unlock()
+			go s.pruneDownloadedSegments(generation, downloadedThrough, true)
+		})
+		return
+	}
+	if processedFloor < targetFloor {
+		go s.pruneDownloadedSegments(generation, downloadedThrough, true)
+		return
+	}
+	s.segmentPruneRunning = false
+	s.scheduleSegmentPruneLocked()
+}
@@ -0,0 +1,388 @@
+package playback
+
+import (
+	"errors"
+	"io"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+)
+
+func TestReportSegmentDownloadedPrunesOnlyExpiredBackBuffer(t *testing.T) {
+	dir := t.TempDir()
+	old := time.Now().Add(-time.Minute)
+	for segment := 0; segment <= 15; segment++ {
+		writePrunerTestFile(t, filepath.Join(dir, segmentFilename(segment, TranscodeOpts{})), []byte("segment"), old)
+	}
+	writePrunerTestFile(t, filepath.Join(dir, "init.mp4"), []byte("init"), old)
+	writePrunerTestFile(t, filepath.Join(dir, "stream.m3u8"), []byte("manifest"), old)
+	writePrunerTestFile(t, filepath.Join(dir, "seg_00004.ts.tmp"), []byte("temporary"), old)
+	writePrunerTestFile(t, filepath.Join(dir, "notes.txt"), []byte("other"), old)
+	// A newly-created file below the floor may belong to a replacement ffmpeg
+	// process and must survive this pass.
+	writePrunerTestFile(t, filepath.Join(dir, segmentFilename(4, TranscodeOpts{})), []byte("fresh"), time.Now())
+
+	session := &TranscodeSession{
+		opts: TranscodeOpts{
+			SessionID:               "prune-test",
+			SegmentDuration:         2,
+			SegmentRetentionSeconds: 10,
+		},
+		outputDir:            dir,
+		lastPruneFloor:       -1,
+		lastRequestedSegment: 0,
+	}
+	session.ReportSegmentDownloaded(15)
+	waitForPrunerFileMissing(t, filepath.Join(dir, segmentFilename(9, TranscodeOpts{})))
+	t.Cleanup(func() {
+		session.mu.Lock()
+		session.segmentGeneration++
+		session.segmentPruneRunning = false
+		session.mu.Unlock()
+	})
+
+	for _, segment := range []int{0, 1, 2, 4, 10, 11, 12, 13, 14, 15} {
+		assertPrunerFileExists(t, filepath.Join(dir, segmentFilename(segment, TranscodeOpts{})))
+	}
+	for _, segment := range []int{3, 5, 6, 7, 8, 9} {
+		assertPrunerFileMissing(t, filepath.Join(dir, segmentFilename(segment, TranscodeOpts{})))
+	}
+	for _, name := range []string{"init.mp4", "stream.m3u8", "seg_00004.ts.tmp", "notes.txt"} {
+		assertPrunerFileExists(t, filepath.Join(dir, name))
+	}
+}
+
+func TestFreshSegmentIsRetriedAfterGuardExpires(t *testing.T) {
+	dir := t.TempDir()
+	opts := TranscodeOpts{
+		SessionID:               "retry-test",
+		SegmentDuration:         2,
+		SegmentRetentionSeconds: 10,
+	}
+	guard := 2*time.Duration(opts.SegmentDuration)*time.Second + 30*time.Second
+	path := filepath.Join(dir, segmentFilename(3, opts))
+	writePrunerTestFile(t, path, []byte("segment"), time.Now().Add(-guard+time.Second))
+	session := &TranscodeSession{
+		opts:                 opts,
+		outputDir:            dir,
+		lastPruneFloor:       -1,
+		lastRequestedSegment: 0,
+	}
+
+	session.ReportSegmentDownloaded(9)
+	waitForPrunerFileMissing(t, path)
+	waitForPrunePass(t, session, 4)
+}
+
+func TestPruningContinuesAcrossBoundedBatches(t *testing.T) {
+	dir := t.TempDir()
+	old := time.Now().Add(-time.Minute)
+	opts := TranscodeOpts{
+		SessionID:               "batch-test",
+		SegmentDuration:         2,
+		SegmentRetentionSeconds: 10,
+	}
+	for segment := 0; segment <= 600; segment++ {
+		writePrunerTestFile(t, filepath.Join(dir, segmentFilename(segment, opts)), []byte("segment"), old)
+	}
+	session := &TranscodeSession{
+		opts:                 opts,
+		outputDir:            dir,
+		lastPruneFloor:       -1,
+		lastRequestedSegment: 0,
+	}
+
+	session.ReportSegmentDownloaded(600)
+	waitForPrunerFileMissing(t, filepath.Join(dir, segmentFilename(594, opts)))
+	waitForPrunePass(t, session, 595)
+
+	for _, segment := range []int{0, 1, 2, 595, 600} {
+		assertPrunerFileExists(t, filepath.Join(dir, segmentFilename(segment, opts)))
+	}
+}
+
+func TestForwardRestartPrunesOnlyFilesPresentAcrossSparseGap(t *testing.T) {
+	dir := t.TempDir()
+	old := time.Now().Add(-time.Minute)
+	opts := TranscodeOpts{
+		SessionID:               "sparse-restart-test",
+		SegmentDuration:         2,
+		SegmentRetentionSeconds: 10,
+		StartSegmentNumber:      100_000,
+	}
+	for _, segment := range []int{3, 70, 99_999, 100_000} {
+		writePrunerTestFile(t, filepath.Join(dir, segmentFilename(segment, opts)), []byte("segment"), old)
+	}
+	writePrunerTestFile(t, filepath.Join(dir, "seg_00004.ts.tmp"), []byte("temporary"), old)
+	session := &TranscodeSession{
+		opts:                 opts,
+		outputDir:            dir,
+		lastPruneFloor:       -1,
+		lastRequestedSegment: opts.StartSegmentNumber,
+		lastCompletedSegment: opts.StartSegmentNumber - 1,
+		pruneBeforeStart:     true,
+	}
+
+	session.ReportSegmentDownloaded(opts.StartSegmentNumber + 5)
+	waitForPrunerFileMissing(t, filepath.Join(dir, segmentFilename(99_999, opts)))
+	waitForPrunePass(t, session, opts.StartSegmentNumber)
+
+	for _, segment := range []int{3, 70, 99_999} {
+		assertPrunerFileMissing(t, filepath.Join(dir, segmentFilename(segment, opts)))
+	}
+	assertPrunerFileExists(t, filepath.Join(dir, segmentFilename(100_000, opts)))
+	assertPrunerFileExists(t, filepath.Join(dir, "seg_00004.ts.tmp"))
+}
+
+func TestCopyRetentionUsesManifestDurations(t *testing.T) {
+	dir := t.TempDir()
+	old := time.Now().Add(-time.Minute)
+	opts := TranscodeOpts{
+		SessionID:               "copy-duration-test",
+		SourceVideoCodec:        "h264",
+		TargetCodecVideo:        "copy",
+		SegmentDuration:         2,
+		SegmentRetentionSeconds: 10,
+	}
+	manifest := strings.Join([]string{
+		"#EXTM3U",
+		"#EXT-X-VERSION:7",
+		"#EXT-X-MEDIA-SEQUENCE:8",
+		"#EXTINF:1.0,",
+		"seg_00008.m4s",
+		"#EXTINF:2.0,",
+		"seg_00009.m4s",
+		"#EXTINF:3.0,",
+		"seg_00010.m4s",
+		"#EXTINF:4.0,",
+		"seg_00011.m4s",
+		"#EXTINF:5.0,",
+		"seg_00012.m4s",
+		"#EXTINF:6.0,",
+		"seg_00013.m4s",
+		"#EXTINF:7.0,",
+		"seg_00014.m4s",
+		"#EXTINF:8.0,",
+		"seg_00015.m4s",
+		"",
+	}, "\n")
+	writePrunerTestFile(t, filepath.Join(dir, "stream.m3u8"), []byte(manifest), old)
+	for segment := 8; segment <= 15; segment++ {
+		writePrunerTestFile(t, filepath.Join(dir, segmentFilename(segment, opts)), []byte("segment"), old)
+	}
+
+	session := &TranscodeSession{
+		opts:                 opts,
+		outputDir:            dir,
+		lastPruneFloor:       -1,
+		lastRequestedSegment: 0,
+	}
+	session.ReportSegmentDownloaded(15)
+	waitForPrunerFileMissing(t, filepath.Join(dir, segmentFilename(12, opts)))
+	t.Cleanup(func() {
+		session.mu.Lock()
+		session.segmentGeneration++
+		session.segmentPruneRunning = false
+		session.mu.Unlock()
+	})
+
+	assertPrunerFileExists(t, filepath.Join(dir, segmentFilename(13, opts)))
+	assertPrunerFileExists(t, filepath.Join(dir, segmentFilename(14, opts)))
+	assertPrunerFileExists(t, filepath.Join(dir, segmentFilename(15, opts)))
+}
+
+func TestReportSegmentDownloadedDisabledKeepsSegments(t *testing.T) {
+	dir := t.TempDir()
+	path := filepath.Join(dir, segmentFilename(3, TranscodeOpts{}))
+	writePrunerTestFile(t, path, []byte("segment"), time.Now().Add(-time.Minute))
+	session := &TranscodeSession{
+		opts:                 TranscodeOpts{SegmentDuration: 2},
+		outputDir:            dir,
+		lastPruneFloor:       -1,
+		lastRequestedSegment: 0,
+	}
+
+	session.ReportSegmentDownloaded(500)
+	time.Sleep(20 * time.Millisecond)
+	assertPrunerFileExists(t, path)
+}
+
+func TestOldGenerationDownloadDoesNotAdvanceRestartedSession(t *testing.T) {
+	session := &TranscodeSession{
+		opts:                 TranscodeOpts{SegmentRetentionSeconds: 600},
+		lastRequestedSegment: 100,
+	}
+	oldGeneration := session.SegmentGeneration()
+
+	session.mu.Lock()
+	session.segmentGeneration++
+	session.lastRequestedSegment = 20
+	session.mu.Unlock()
+
+	session.ReportSegmentDownloadedForGeneration(101, oldGeneration)
+	if got := session.LastRequestedSegment(); got != 20 {
+		t.Fatalf("last requested segment = %d, want restarted position 20", got)
+	}
+}
+
+func TestCloseInvalidatesInFlightSegmentGeneration(t *testing.T) {
+	session := &TranscodeSession{
+		opts:                 TranscodeOpts{SegmentRetentionSeconds: 600},
+		outputDir:            t.TempDir(),
+		lastRequestedSegment: 20,
+	}
+	downloadGeneration := session.SegmentGeneration()
+
+	if err := session.Close(); err != nil {
+		t.Fatalf("Close: %v", err)
+	}
+	session.ReportSegmentDownloadedForGeneration(100, downloadGeneration)
+	if got := session.LastRequestedSegment(); got != 20 {
+		t.Fatalf("last requested segment = %d, want closed-session position 20", got)
+	}
+}
+
+func TestOpenSegmentDescriptorSurvivesUnlink(t *testing.T) {
+	dir := t.TempDir()
+	path := filepath.Join(dir, "seg_00001.ts")
+	want := []byte("complete segment")
+	writePrunerTestFile(t, path, want, time.Now())
+	session := &TranscodeSession{outputDir: dir}
+
+	segment, err := session.OpenSegment("seg_00001.ts")
+	if err != nil {
+		t.Fatalf("OpenSegment: %v", err)
+	}
+	defer func() { _ = segment.Close() }()
+	if err := os.Remove(path); err != nil {
+		t.Fatalf("remove opened segment: %v", err)
+	}
+	got, err := io.ReadAll(segment.File)
+	if err != nil {
+		t.Fatalf("read opened segment: %v", err)
+	}
+	if string(got) != string(want) {
+		t.Fatalf("opened segment = %q, want %q", got, want)
+	}
+}
+
+func TestOpenSegmentLeaseCapturesGeneration(t *testing.T) {
+	dir := t.TempDir()
+	path := filepath.Join(dir, "seg_00001.ts")
+	writePrunerTestFile(t, path, []byte("complete segment"), time.Now())
+	session := &TranscodeSession{outputDir: dir, segmentGeneration: 7}
+
+	segment, err := session.OpenSegment("seg_00001.ts")
+	if err != nil {
+		t.Fatalf("OpenSegment: %v", err)
+	}
+	defer func() { _ = segment.Close() }()
+	if segment.Generation != 7 {
+		t.Fatalf("segment generation = %d, want 7", segment.Generation)
+	}
+}
+
+func TestGenerationTokenRejectsPriorSessionIncarnation(t *testing.T) {
+	oldDir := t.TempDir()
+	newDir := t.TempDir()
+	const name = "seg_00007.ts"
+	writePrunerTestFile(t, filepath.Join(oldDir, name), []byte("old segment"), time.Now())
+	writePrunerTestFile(t, filepath.Join(newDir, name), []byte("new segment"), time.Now())
+
+	oldSession := &TranscodeSession{outputDir: oldDir}
+	oldLease, err := oldSession.OpenSegment(name)
+	if err != nil {
+		t.Fatalf("open old segment: %v", err)
+	}
+	defer func() { _ = oldLease.Close() }()
+
+	newSession := &TranscodeSession{outputDir: newDir}
+	newLease, err := newSession.OpenSegment(name)
+	if err != nil {
+		t.Fatalf("open new segment: %v", err)
+	}
+	defer func() { _ = newLease.Close() }()
+	if oldLease.Generation != newLease.Generation {
+		t.Fatalf("numeric generations differ: old=%d new=%d; test requires restart collision", oldLease.Generation, newLease.Generation)
+	}
+	if oldLease.GenerationToken == newLease.GenerationToken {
+		t.Fatal("separate session objects reused an opaque generation token")
+	}
+
+	newSession.ReportSegmentDownloadedForGenerationToken(7, oldLease.GenerationToken)
+	if got := newSession.LastRequestedSegment(); got != 0 {
+		t.Fatalf("prior-session token advanced replacement to %d", got)
+	}
+	newSession.ReportSegmentDownloadedForGenerationToken(7, newLease.GenerationToken)
+	if got := newSession.LastRequestedSegment(); got != 7 {
+		t.Fatalf("current token advanced replacement to %d, want 7", got)
+	}
+}
+
+func TestOpenSegmentWaitsThroughRestartInsteadOfLeasingStaleFile(t *testing.T) {
+	dir := t.TempDir()
+	path := filepath.Join(dir, "seg_00001.ts")
+	writePrunerTestFile(t, path, []byte("stale segment"), time.Now())
+	session := &TranscodeSession{outputDir: dir, restarting: true}
+
+	if segment, err := session.OpenSegment("seg_00001.ts"); !errors.Is(err, ErrSegmentNotFound) {
+		if segment != nil {
+			_ = segment.Close()
+		}
+		t.Fatalf("OpenSegment error = %v, want ErrSegmentNotFound", err)
+	}
+}
+
+func writePrunerTestFile(t *testing.T, path string, contents []byte, modTime time.Time) {
+	t.Helper()
+	if err := os.WriteFile(path, contents, 0o600); err != nil {
+		t.Fatalf("write %s: %v", filepath.Base(path), err)
+	}
+	if err := os.Chtimes(path, modTime, modTime); err != nil {
+		t.Fatalf("set times for %s: %v", filepath.Base(path), err)
+	}
+}
+
+func waitForPrunePass(t *testing.T, session *TranscodeSession, wantFloor int) {
+	t.Helper()
+	deadline := time.Now().Add(2 * time.Second)
+	for time.Now().Before(deadline) {
+		session.mu.Lock()
+		running := session.segmentPruneRunning
+		floor := session.lastPruneFloor
+		session.mu.Unlock()
+		if !running && floor >= wantFloor {
+			return
+		}
+		time.Sleep(5 * time.Millisecond)
+	}
+	t.Fatal("segment prune pass did not finish")
+}
+
+func waitForPrunerFileMissing(t *testing.T, path string) {
+	t.Helper()
+	deadline := time.Now().Add(2 * time.Second)
+	for time.Now().Before(deadline) {
+		if _, err := os.Stat(path); os.IsNotExist(err) {
+			return
+		}
+		time.Sleep(5 * time.Millisecond)
+	}
+	t.Fatalf("expected %s to be removed", filepath.Base(path))
+}
+
+func assertPrunerFileExists(t *testing.T, path string) {
+	t.Helper()
+	if _, err := os.Stat(path); err != nil {
+		t.Fatalf("expected %s to exist: %v", filepath.Base(path), err)
+	}
+}
+
+func assertPrunerFileMissing(t *testing.T, path string) {
+	t.Helper()
+	if _, err := os.Stat(path); !os.IsNotExist(err) {
+		t.Fatalf("expected %s to be removed, stat error = %v", filepath.Base(path), err)
+	}
+}
@@ -20,6 +20,7 @@ import (
 	"time"

 	"github.com/Silo-Server/silo-server/internal/models"
+	"github.com/google/uuid"
 )

 func init() {
@@ -52,15 +53,16 @@ type TranscodeOpts struct {
 	StreamOriginSeconds float64
 	// CopySeekAnchorResolved distinguishes a valid zero-second origin from
 	// older/shared recipes that never resolved a copy seek anchor.
-	CopySeekAnchorResolved bool
-	TargetResolution       string // e.g., 1080p, 720p
-	TargetCodecVideo       string // e.g., h264 (or hevc if allowed)
-	TargetCodecAudio       string // e.g., aac
-	SegmentDuration        int    // seconds, default 6
-	StartSegmentNumber     int    // -hls_segment_start_number, default 0
-	FFmpegPath             string // optional explicit ffmpeg binary path
-	HWAccel                string // auto, qsv, vaapi, nvenc, none
-	HWDevice               string // e.g., /dev/dri/renderD128 (default if empty)
+	CopySeekAnchorResolved  bool
+	TargetResolution        string // e.g., 1080p, 720p
+	TargetCodecVideo        string // e.g., h264 (or hevc if allowed)
+	TargetCodecAudio        string // e.g., aac
+	SegmentDuration         int    // seconds, default 6
+	SegmentRetentionSeconds int    // downloaded media retained behind the client; 0 disables pruning
+	StartSegmentNumber      int    // -hls_segment_start_number, default 0
+	FFmpegPath              string // optional explicit ffmpeg binary path
+	HWAccel                 string // auto, qsv, vaapi, nvenc, none
+	HWDevice                string // e.g., /dev/dri/renderD128 (default if empty)
 	// AvoidHWDevice asks the initial multi-device allocator to prefer any other
 	// present render device. It is a process-local startup hint used after an
 	// early GPU failure; the selected concrete device remains fully reserved and
@@ -119,6 +121,13 @@ type TranscodeSession struct {
 	done                 chan struct{} // closed when the monitor goroutine finishes
 	stdinPipe            io.WriteCloser
 	lastRequestedSegment int
+	lastCompletedSegment int
+	lastPruneFloor       int
+	lastPruneHighWater   int
+	segmentPruneRunning  bool
+	pruneBeforeStart     bool
+	segmentGeneration    uint64
+	segmentIncarnation   string
 	throttler            *TranscodeThrottler
 	stderrLinesLogged    int
 	stderrBytesLogged    int
@@ -246,6 +255,10 @@ func StartTranscode(ctx context.Context, opts TranscodeOpts) (*TranscodeSession,
 		done:                     make(chan struct{}),
 		stderr:                   newBoundedTailBuffer(stderrTailMaxBytes),
 		lastRequestedSegment:     opts.StartSegmentNumber,
+		lastCompletedSegment:     opts.StartSegmentNumber - 1,
+		lastPruneFloor:           opts.StartSegmentNumber - 1,
+		lastPruneHighWater:       opts.StartSegmentNumber - 1,
+		segmentIncarnation:       uuid.NewString(),
 		reserveHWDeviceOnRestart: reserveHWDeviceOnRestart,
 	}

@@ -1878,6 +1891,59 @@ func (s *TranscodeSession) GetSegment(name string) (string, error) {
 	return segPath, nil
 }

+// SegmentLease binds an opened segment descriptor to the FFmpeg generation it
+// came from. Keeping the descriptor open makes it a filesystem-backed lease: a
+// concurrent prune may unlink the directory entry, but the response can still
+// read the complete file on POSIX filesystems.
+type SegmentLease struct {
+	File            *os.File
+	Info            os.FileInfo
+	Generation      uint64
+	GenerationToken string
+}
+
+// Close releases the opened segment descriptor.
+func (l *SegmentLease) Close() error {
+	return l.File.Close()
+}
+
+// OpenSegment opens a completed segment for serving and captures its FFmpeg
+// generation atomically with the open.
+func (s *TranscodeSession) OpenSegment(name string) (*SegmentLease, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if s.restarting {
+		return nil, ErrSegmentNotFound
+	}
+
+	clean := filepath.Base(name)
+	segment, err := os.Open(filepath.Join(s.outputDir, clean))
+	if err != nil {
+		if os.IsNotExist(err) {
+			return nil, ErrSegmentNotFound
+		}
+		return nil, fmt.Errorf("open segment: %w", err)
+	}
+	info, err := segment.Stat()
+	if err != nil {
+		_ = segment.Close()
+		return nil, fmt.Errorf("stat segment: %w", err)
+	}
+	if info.Size() <= 0 {
+		_ = segment.Close()
+		return nil, ErrSegmentNotFound
+	}
+	if s.segmentIncarnation == "" {
+		s.segmentIncarnation = uuid.NewString()
+	}
+	return &SegmentLease{
+		File:            segment,
+		Info:            info,
+		Generation:      s.segmentGeneration,
+		GenerationToken: s.segmentGenerationTokenLocked(),
+	}, nil
+}
+
 // Close terminates the ffmpeg process and removes the temporary output directory.
 func (s *TranscodeSession) Close() error {
 	return s.shutdown(true)
@@ -1912,6 +1978,8 @@ func (s *TranscodeSession) shutdown(removeOutput bool) error {
 	defer s.mu.Unlock()

 	s.running = false
+	s.segmentGeneration++
+	s.segmentPruneRunning = false

 	// Clean up temporary directory.
 	if removeOutput && s.outputDir != "" {
@@ -2027,6 +2095,22 @@ func (s *TranscodeSession) restart(
 		return nil
 	}
 	s.restarting = true
+	s.segmentGeneration++
+	s.segmentPruneRunning = false
+	previousRequestedSegment := s.lastRequestedSegment
+	forwardRestart := startSegment > previousRequestedSegment
+	s.lastRequestedSegment = startSegment
+	s.lastCompletedSegment = startSegment - 1
+	if forwardRestart && s.opts.SegmentRetentionSeconds > 0 {
+		// Keep the old prune cursor reachable until the replacement process has
+		// produced a complete back buffer. Otherwise every forward restart leaves
+		// one retained window permanently behind startSegment.
+		s.pruneBeforeStart = true
+	} else {
+		s.lastPruneFloor = startSegment - 1
+		s.pruneBeforeStart = false
+	}
+	s.lastPruneHighWater = startSegment - 1
 	cancelCurrent := s.cancel
 	done := s.done
 	s.mu.Unlock()
@@ -2123,6 +2207,7 @@ func (s *TranscodeSession) restart(
 	s.restarting = false
 	s.stdinPipe = stdinPipe
 	s.lastRequestedSegment = startSegment
+	s.lastCompletedSegment = startSegment - 1
 	s.done = make(chan struct{})
 	hook := s.restartHook
 	s.mu.Unlock()
@@ -2186,6 +2271,49 @@ func (s *TranscodeSession) WaitForSegment(name string, timeout time.Duration) (s
 	}
 }

+// WaitForOpenSegment is the opened-file counterpart to WaitForSegment. It
+// closes the stat-to-open race for HTTP serving while preserving the same
+// restart and ffmpeg-exit error contract.
+func (s *TranscodeSession) WaitForOpenSegment(name string, timeout time.Duration) (*SegmentLease, error) {
+	deadline := time.After(timeout)
+	for {
+		segment, err := s.OpenSegment(name)
+		if err == nil {
+			return segment, nil
+		}
+		if !errors.Is(err, ErrSegmentNotFound) {
+			return nil, err
+		}
+
+		s.mu.Lock()
+		running := s.running
+		restarting := s.restarting
+		waitErr := s.waitErr
+		s.mu.Unlock()
+
+		if restarting {
+			select {
+			case <-deadline:
+				return nil, ErrSegmentNotFound
+			case <-time.After(100 * time.Millisecond):
+				continue
+			}
+		}
+		if !running && waitErr != nil {
+			return nil, fmt.Errorf("%w: %w", ErrTranscodeFailed, waitErr)
+		}
+		if !running {
+			return nil, ErrSegmentNotFound
+		}
+
+		select {
+		case <-deadline:
+			return nil, ErrSegmentNotFound
+		case <-time.After(100 * time.Millisecond):
+		}
+	}
+}
+
 // RewriteManifestPaths prefixes relative segment references in an HLS manifest
 // with segPrefix (e.g. "segment/") and optionally appends rawQuery as a query
 // string. This ensures the HLS player's segment requests match server routes
@@ -2433,9 +2561,54 @@ func (s *TranscodeSession) RestartSeekTarget(segNum int) (float64, bool, error)
 func (s *TranscodeSession) ReportSegmentDownloaded(segNum int) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
+	s.reportSegmentDownloadedLocked(segNum)
+}
+
+// SegmentGeneration identifies the current ffmpeg timeline. Callers that hold
+// an open segment across a restart use it to prevent the old response from
+// advancing the replacement process's download position.
+func (s *TranscodeSession) SegmentGeneration() uint64 {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.segmentGeneration
+}
+
+// ReportSegmentDownloadedForGeneration records completion only if the served
+// file belongs to the current ffmpeg timeline.
+func (s *TranscodeSession) ReportSegmentDownloadedForGeneration(segNum int, generation uint64) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if generation != s.segmentGeneration {
+		return
+	}
+	s.reportSegmentDownloadedLocked(segNum)
+}
+
+// ReportSegmentDownloadedForGenerationToken records a completion relayed over
+// HTTP only when it belongs to this exact session object and FFmpeg timeline.
+// The opaque incarnation prevents a delayed proxy acknowledgement from matching
+// a reconstructed session whose numeric generation restarted at zero.
+func (s *TranscodeSession) ReportSegmentDownloadedForGenerationToken(segNum int, generationToken string) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if generationToken == "" || generationToken != s.segmentGenerationTokenLocked() {
+		return
+	}
+	s.reportSegmentDownloadedLocked(segNum)
+}
+
+func (s *TranscodeSession) segmentGenerationTokenLocked() string {
+	return s.segmentIncarnation + ":" + strconv.FormatUint(s.segmentGeneration, 10)
+}
+
+func (s *TranscodeSession) reportSegmentDownloadedLocked(segNum int) {
 	if segNum > s.lastRequestedSegment {
 		s.lastRequestedSegment = segNum
 	}
+	if segNum > s.lastCompletedSegment {
+		s.lastCompletedSegment = segNum
+	}
+	s.scheduleSegmentPruneLocked()
 }

 // LastRequestedSegment returns the highest segment number downloaded by the client.
@@ -20,10 +20,11 @@ import (
 // internal/playback does not import internal/config (avoiding an import cycle);
 // each embedding handler adapts its own config snapshot into this shape.
 type TranscodeRuntimeConfig struct {
-	TranscodeDir string
-	FFmpegPath   string
-	HWAccel      string
-	HWDevice     string
+	TranscodeDir            string
+	FFmpegPath              string
+	HWAccel                 string
+	HWDevice                string
+	SegmentRetentionSeconds int
 }

 // sessionReconstructor is the SessionManager capability used to re-register a
@@ -579,6 +580,7 @@ func (m *TranscodeManager) doReconstructTranscode(ctx context.Context, sessionID
 	// operator config change applies to reconstructed sessions too.
 	opts.HWAccel = cfg.HWAccel
 	opts.HWDevice = cfg.HWDevice
+	opts.SegmentRetentionSeconds = cfg.SegmentRetentionSeconds

 	// Resume near the segment the client is actually requesting. The card records
 	// the original start; if the client has played past it, spawning ffmpeg at the
@@ -2,11 +2,148 @@ package playback

 import (
 	"context"
+	"os"
 	"os/exec"
+	"path/filepath"
 	"testing"
 	"time"
 )

+func TestForwardRestartPreservesConfiguredBackBuffer(t *testing.T) {
+	truePath, err := exec.LookPath("true")
+	if err != nil {
+		t.Skipf("`true` not found in PATH: %v", err)
+	}
+
+	dir := t.TempDir()
+	old := time.Now().Add(-time.Minute)
+	for _, segment := range []int{3, 40, 64, 65, 69, 70, 100} {
+		name := filepath.Join(dir, segmentFilename(segment, TranscodeOpts{}))
+		if err := os.WriteFile(name, []byte("segment"), 0o600); err != nil {
+			t.Fatal(err)
+		}
+		if err := os.Chtimes(name, old, old); err != nil {
+			t.Fatal(err)
+		}
+	}
+	session := &TranscodeSession{
+		outputDir:            dir,
+		lastRequestedSegment: 70,
+		lastCompletedSegment: 70,
+		lastPruneFloor:       -1,
+		opts: TranscodeOpts{
+			OutputDir:               dir,
+			TargetCodecVideo:        "h264",
+			SegmentDuration:         4,
+			SegmentRetentionSeconds: 20,
+			StartSegmentNumber:      0,
+			FFmpegPath:              truePath,
+		},
+	}
+
+	if err := session.Restart(context.Background(), 400, 100); err != nil {
+		t.Fatalf("Restart: %v", err)
+	}
+	// Restart itself has no downstream-completion proof for the replacement
+	// generation, so it must not unlink even old files synchronously.
+	for _, segment := range []int{3, 40, 64, 65, 69, 70, 100} {
+		path := filepath.Join(dir, segmentFilename(segment, TranscodeOpts{}))
+		if _, err := os.Stat(path); err != nil {
+			t.Errorf("segment %d was removed during restart: %v", segment, err)
+		}
+	}
+
+	// Once the replacement generation has produced a complete back buffer,
+	// the preserved range must become eligible again rather than leaking one
+	// full window after every forward seek.
+	session.ReportSegmentDownloaded(105)
+	waitForPrunerFileMissing(t, filepath.Join(dir, segmentFilename(70, TranscodeOpts{})))
+	for _, segment := range []int{3, 40, 64, 65, 69, 70} {
+		path := filepath.Join(dir, segmentFilename(segment, TranscodeOpts{}))
+		if _, err := os.Stat(path); !os.IsNotExist(err) {
+			t.Errorf("expired pre-restart segment %d survived completed-download pruning: %v", segment, err)
+		}
+	}
+	if _, err := os.Stat(filepath.Join(dir, segmentFilename(100, TranscodeOpts{}))); err != nil {
+		t.Fatalf("replacement startup segment was removed: %v", err)
+	}
+}
+
+func TestForwardRestartWithoutCompletedDownloadPreservesFreshFiles(t *testing.T) {
+	truePath, err := exec.LookPath("true")
+	if err != nil {
+		t.Skipf("`true` not found in PATH: %v", err)
+	}
+
+	dir := t.TempDir()
+	for _, segment := range []int{3, 40, 70} {
+		if err := os.WriteFile(filepath.Join(dir, segmentFilename(segment, TranscodeOpts{})), []byte("segment"), 0o600); err != nil {
+			t.Fatal(err)
+		}
+	}
+	session := &TranscodeSession{
+		outputDir:            dir,
+		lastRequestedSegment: 70,
+		lastCompletedSegment: -1,
+		lastPruneFloor:       -1,
+		opts: TranscodeOpts{
+			OutputDir:               dir,
+			TargetCodecVideo:        "h264",
+			SegmentDuration:         4,
+			SegmentRetentionSeconds: 20,
+			StartSegmentNumber:      0,
+			FFmpegPath:              truePath,
+		},
+	}
+
+	if err := session.Restart(context.Background(), 400, 100); err != nil {
+		t.Fatalf("Restart: %v", err)
+	}
+	time.Sleep(20 * time.Millisecond)
+	for _, segment := range []int{3, 40, 70} {
+		if _, err := os.Stat(filepath.Join(dir, segmentFilename(segment, TranscodeOpts{}))); err != nil {
+			t.Errorf("fresh segment %d was removed without a completed download: %v", segment, err)
+		}
+	}
+}
+
+func TestForwardRestartKeepsSegmentsWhenRetentionDisabled(t *testing.T) {
+	truePath, err := exec.LookPath("true")
+	if err != nil {
+		t.Skipf("`true` not found in PATH: %v", err)
+	}
+
+	dir := t.TempDir()
+	for _, segment := range []int{3, 40, 70} {
+		name := filepath.Join(dir, segmentFilename(segment, TranscodeOpts{}))
+		if err := os.WriteFile(name, []byte("segment"), 0o600); err != nil {
+			t.Fatal(err)
+		}
+	}
+	session := &TranscodeSession{
+		outputDir:            dir,
+		lastRequestedSegment: 70,
+		lastCompletedSegment: 70,
+		opts: TranscodeOpts{
+			OutputDir:          dir,
+			TargetCodecVideo:   "h264",
+			SegmentDuration:    4,
+			StartSegmentNumber: 0,
+			FFmpegPath:         truePath,
+		},
+	}
+
+	if err := session.Restart(context.Background(), 400, 100); err != nil {
+		t.Fatalf("Restart: %v", err)
+	}
+	for _, segment := range []int{3, 40, 70} {
+		path := filepath.Join(dir, segmentFilename(segment, TranscodeOpts{}))
+		if _, err := os.Stat(path); err != nil {
+			t.Errorf("disabled retention removed segment %d: %v", segment, err)
+		}
+	}
+}
+
 // TestSegmentRecoveryDecisionWaitsWhileRestarting covers half of issue #243's
 // seek-freeze: while a restart is already in flight, a concurrent segment
 // request must WAIT for the restart's output rather than trigger another
@@ -0,0 +1,133 @@
+package proxy
+
+import (
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/Silo-Server/silo-server/internal/config"
+	"github.com/Silo-Server/silo-server/internal/nodeconfig"
+	"github.com/Silo-Server/silo-server/internal/streamtoken"
+	"github.com/Silo-Server/silo-server/internal/transcodeproxy"
+)
+
+func TestTranscodeProxyAcknowledgesOnlyFullDownstreamResponse(t *testing.T) {
+	const (
+		body       = "complete segment"
+		generation = "incarnation:17"
+	)
+	tests := []struct {
+		name        string
+		rangeHeader string
+		wantStatus  int
+		wantAck     int32
+	}{
+		{name: "ordinary get", wantStatus: http.StatusOK, wantAck: 1},
+		{name: "whole-file range", rangeHeader: "bytes=0-", wantStatus: http.StatusPartialContent, wantAck: 1},
+		{name: "partial range", rangeHeader: "bytes=1-", wantStatus: http.StatusPartialContent},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			var acknowledgements atomic.Int32
+			seenRange := make(chan string, 1)
+			node := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				switch r.Method {
+				case http.MethodGet:
+					if r.Header.Get(transcodeproxy.RequestHeader) != "1" {
+						t.Error("segment request omitted proxy marker")
+					}
+					seenRange <- r.Header.Get("Range")
+					w.Header().Set(transcodeproxy.GenerationHeader, generation)
+					http.ServeContent(w, r, "seg_00007.ts", time.Time{}, strings.NewReader(body))
+				case http.MethodPost:
+					if r.Header.Get(transcodeproxy.GenerationHeader) != generation {
+						t.Errorf("ack generation = %q, want %q", r.Header.Get(transcodeproxy.GenerationHeader), generation)
+					}
+					acknowledgements.Add(1)
+					w.WriteHeader(http.StatusNoContent)
+				default:
+					http.Error(w, "unexpected method", http.StatusMethodNotAllowed)
+				}
+			}))
+			defer node.Close()
+
+			server := newCompletionTestProxy(node.Client())
+			claims := &streamtoken.Claims{SessionID: "public", TranscodeNode: node.URL}
+			req := httptest.NewRequest(http.MethodGet, "/stream/transcode/token/segment/seg_00007.ts", nil)
+			req.Header.Set("Range", tt.rangeHeader)
+			rr := httptest.NewRecorder()
+			server.proxyToTranscodeNode(rr, req, claims, "/transcode/remote/segment/seg_00007.ts")
+
+			if rr.Code != tt.wantStatus {
+				t.Fatalf("status = %d, want %d; body = %q", rr.Code, tt.wantStatus, rr.Body.String())
+			}
+			if got := <-seenRange; got != tt.rangeHeader {
+				t.Fatalf("forwarded Range = %q, want %q", got, tt.rangeHeader)
+			}
+			if got := rr.Header().Get(transcodeproxy.GenerationHeader); got != "" {
+				t.Fatalf("internal generation leaked downstream: %q", got)
+			}
+			if got := acknowledgements.Load(); got != tt.wantAck {
+				t.Fatalf("acknowledgements = %d, want %d", got, tt.wantAck)
+			}
+		})
+	}
+}
+
+func TestTranscodeProxyDoesNotAcknowledgeFailedDownstreamWrite(t *testing.T) {
+	var acknowledgements atomic.Int32
+	node := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.Method == http.MethodPost {
+			acknowledgements.Add(1)
+			w.WriteHeader(http.StatusNoContent)
+			return
+		}
+		w.Header().Set(transcodeproxy.GenerationHeader, "incarnation:21")
+		http.ServeContent(w, r, "seg_00007.ts", time.Time{}, strings.NewReader("complete segment"))
+	}))
+	defer node.Close()
+
+	server := newCompletionTestProxy(node.Client())
+	claims := &streamtoken.Claims{SessionID: "public", TranscodeNode: node.URL}
+	req := httptest.NewRequest(http.MethodGet, "/stream/transcode/token/segment/seg_00007.ts", nil)
+	w := &failingCompletionResponseWriter{header: make(http.Header), remaining: 5}
+	server.proxyToTranscodeNode(w, req, claims, "/transcode/remote/segment/seg_00007.ts")
+
+	if got := acknowledgements.Load(); got != 0 {
+		t.Fatalf("failed downstream transfer produced %d acknowledgement(s)", got)
+	}
+}
+
+func newCompletionTestProxy(client *http.Client) *Server {
+	watcher := nodeconfig.NewWatcher(nil, nil, nil, nodeconfig.BootstrapOverrides{})
+	cfg := &config.Config{}
+	cfg.Auth.JWTSecret = "proxy-test-secret"
+	watcher.SetConfigForTest(cfg)
+	server := NewServer(watcher, nil)
+	server.httpClient = client
+	return server
+}
+
+type failingCompletionResponseWriter struct {
+	header    http.Header
+	status    int
+	remaining int
+}
+
+func (w *failingCompletionResponseWriter) Header() http.Header { return w.header }
+
+func (w *failingCompletionResponseWriter) WriteHeader(status int) { w.status = status }
+
+func (w *failingCompletionResponseWriter) Write(p []byte) (int, error) {
+	if w.remaining <= 0 {
+		return 0, io.ErrClosedPipe
+	}
+	n := min(len(p), w.remaining)
+	w.remaining -= n
+	return n, io.ErrClosedPipe
+}
@@ -6,6 +6,7 @@ import (
 	"io"
 	"log/slog"
 	"net/http"
+	"path/filepath"
 	"strconv"
 	"strings"
 	"time"
@@ -13,10 +14,12 @@ import (
 	"github.com/go-chi/chi/v5"
 	"github.com/go-chi/cors"

+	"github.com/Silo-Server/silo-server/internal/httpstream"
 	"github.com/Silo-Server/silo-server/internal/nodeconfig"
 	"github.com/Silo-Server/silo-server/internal/nodesessions"
 	"github.com/Silo-Server/silo-server/internal/playback"
 	"github.com/Silo-Server/silo-server/internal/streamtoken"
+	"github.com/Silo-Server/silo-server/internal/transcodeproxy"
 )

 // Server is the HTTP handler for proxy mode.
@@ -343,6 +346,9 @@ func (s *Server) proxyToTranscodeNode(w http.ResponseWriter, r *http.Request, cl
 	}

 	targetURL := claims.TranscodeNode + path
+	isSegmentRoute := strings.Contains(path, "/segment/")
+	_, segmentParseErr := playback.ParseSegmentNumber(filepath.Base(path))
+	isMediaSegment := segmentParseErr == nil
 	if rawQuery := r.URL.RawQuery; rawQuery != "" {
 		targetURL = fmt.Sprintf("%s?%s", targetURL, rawQuery)
 	}
@@ -353,6 +359,9 @@ func (s *Server) proxyToTranscodeNode(w http.ResponseWriter, r *http.Request, cl
 		return
 	}
 	req.Header.Set("Authorization", "Bearer "+cfg.Auth.JWTSecret)
+	if isSegmentRoute {
+		transcodeproxy.PrepareRequest(req, r)
+	}
 	// Forward the verified stream token so the transcode node can self-reconstruct
 	// a lost session after its OWN restart: the token carries the full byte-affecting
 	// recipe, so the node can re-spawn ffmpeg seeked to the requested segment instead
@@ -368,16 +377,21 @@ func (s *Server) proxyToTranscodeNode(w http.ResponseWriter, r *http.Request, cl
 		http.Error(w, "transcode node unavailable", http.StatusBadGateway)
 		return
 	}
-	defer resp.Body.Close()
+	defer func() { _ = resp.Body.Close() }()

-	// Copy response headers
-	for k, vv := range resp.Header {
-		for _, v := range vv {
-			w.Header().Add(k, v)
+	generation := resp.Header.Get(transcodeproxy.GenerationHeader)
+	transcodeproxy.CopyResponseHeaders(w.Header(), resp.Header)
+	sw := httpstream.NewRollingDeadlineWriter(w)
+	sw.WriteHeader(resp.StatusCode)
+	if _, copyErr := io.Copy(sw, resp.Body); copyErr != nil {
+		return
+	}
+	if isMediaSegment && generation != "" && r.Method == http.MethodGet &&
+		sw.CompletedFullResponse(transcodeproxy.FullRepresentationSize(resp)) {
+		if ackErr := transcodeproxy.Acknowledge(r.Context(), s.httpClient, claims.TranscodeNode+path, cfg.Auth.JWTSecret, generation); ackErr != nil {
+			slog.WarnContext(r.Context(), "acknowledge transcode segment completion", "component", "proxy", "error", ackErr, "playback_session_id", claims.SessionID)
 		}
 	}
-	w.WriteHeader(resp.StatusCode)
-	io.Copy(w, resp.Body)
 }

 func (s *Server) handleForceReload(w http.ResponseWriter, r *http.Request) {
@@ -19,10 +19,12 @@ import (
 	"golang.org/x/sync/singleflight"

 	"github.com/Silo-Server/silo-server/internal/chapterthumbs"
+	"github.com/Silo-Server/silo-server/internal/httpstream"
 	"github.com/Silo-Server/silo-server/internal/nodeconfig"
 	"github.com/Silo-Server/silo-server/internal/nodesessions"
 	"github.com/Silo-Server/silo-server/internal/playback"
 	"github.com/Silo-Server/silo-server/internal/streamtoken"
+	"github.com/Silo-Server/silo-server/internal/transcodeproxy"
 )

 // TranscodeStartRequest is the JSON body for POST /transcode/start.
@@ -387,6 +389,7 @@ func (s *Server) Handler() http.Handler {
 		r.Delete("/transcode/{session_id}", s.handleStop)
 		r.Get("/transcode/{session_id}/master.m3u8", s.handleManifest)
 		r.Get("/transcode/{session_id}/segment/{name}", s.handleSegment)
+		r.Post("/transcode/{session_id}/segment/{name}/downloaded", s.handleSegmentDownloaded)
 		r.Post("/admin/force-reload", s.handleForceReload)
 		r.Get("/status", s.handleStatus)
 	})
@@ -481,27 +484,28 @@ func (s *Server) handleStart(w http.ResponseWriter, r *http.Request) {
 	outputDir := filepath.Join(cfg.Playback.TranscodeDir, req.SessionID)

 	opts := playback.TranscodeOpts{
-		InputPath:              req.InputPath,
-		OutputDir:              outputDir,
-		SessionID:              req.SessionID,
-		SourceVideoCodec:       req.SourceVideoCodec,
-		SourceVideoProfile:     req.SourceVideoProfile,
-		SourceVideoBitDepth:    req.SourceVideoBitDepth,
-		SoftwareVideoDecode:    req.SoftwareVideoDecode,
-		VideoBitstreamFilter:   req.VideoBitstreamFilter,
-		SeekSeconds:            req.SeekSeconds,
-		StreamOriginSeconds:    req.StreamOriginSeconds,
-		CopySeekAnchorResolved: req.CopySeekAnchorResolved,
-		StartSegmentNumber:     req.StartSegmentNumber,
-		TargetResolution:       req.TargetResolution,
-		TargetCodecVideo:       req.TargetCodecVideo,
-		TargetCodecAudio:       req.TargetCodecAudio,
-		TargetAudioChannels:    req.TargetAudioChannels,
-		TargetAudioBitrateKbps: req.TargetAudioBitrateKbps,
-		TargetBitrateKbps:      req.TargetBitrateKbps,
-		SegmentDuration:        req.SegmentDuration,
-		FFmpegPath:             cfg.Playback.FFmpegPath,
-		HWAccel:                req.HWAccel,
+		InputPath:               req.InputPath,
+		OutputDir:               outputDir,
+		SessionID:               req.SessionID,
+		SourceVideoCodec:        req.SourceVideoCodec,
+		SourceVideoProfile:      req.SourceVideoProfile,
+		SourceVideoBitDepth:     req.SourceVideoBitDepth,
+		SoftwareVideoDecode:     req.SoftwareVideoDecode,
+		VideoBitstreamFilter:    req.VideoBitstreamFilter,
+		SeekSeconds:             req.SeekSeconds,
+		StreamOriginSeconds:     req.StreamOriginSeconds,
+		CopySeekAnchorResolved:  req.CopySeekAnchorResolved,
+		StartSegmentNumber:      req.StartSegmentNumber,
+		TargetResolution:        req.TargetResolution,
+		TargetCodecVideo:        req.TargetCodecVideo,
+		TargetCodecAudio:        req.TargetCodecAudio,
+		TargetAudioChannels:     req.TargetAudioChannels,
+		TargetAudioBitrateKbps:  req.TargetAudioBitrateKbps,
+		TargetBitrateKbps:       req.TargetBitrateKbps,
+		SegmentDuration:         req.SegmentDuration,
+		SegmentRetentionSeconds: cfg.Playback.SegmentRetentionSeconds,
+		FFmpegPath:              cfg.Playback.FFmpegPath,
+		HWAccel:                 req.HWAccel,
 		// This node's configured device (or device list — StartTranscode
 		// resolves it to one GPU), matching what reconstruction uses so fresh
 		// and reconstructed sessions balance identically.
@@ -706,6 +710,7 @@ func (s *Server) spawnReconstruct(r *http.Request, sessionID string, requestedSe
 	// rebuild. Run as a transcode node, not integrated (card.TranscodeOpts defaults).
 	opts.HWAccel = cfg.Playback.HWAccel
 	opts.HWDevice = cfg.Playback.HWDevice
+	opts.SegmentRetentionSeconds = cfg.Playback.SegmentRetentionSeconds
 	opts.NodeType = "transcode"
 	opts.ExecutionMode = "transcode_node"

@@ -900,7 +905,7 @@ func (s *Server) handleSegment(w http.ResponseWriter, r *http.Request) {
 		s.touchSession(sessionID)
 	}

-	segPath, err := session.GetSegment(name)
+	segmentLease, err := session.OpenSegment(name)
 	if err != nil && err == playback.ErrSegmentNotFound {
 		segNum, parseErr := playback.ParseSegmentNumber(name)
 		if parseErr == nil {
@@ -935,7 +940,7 @@ func (s *Server) handleSegment(w http.ResponseWriter, r *http.Request) {
 					"session", sessionID,
 					"playback_session_id", sessionID,
 				)
-				segPath, err = session.WaitForSegment(name, decision.WaitTimeout)
+				segmentLease, err = session.WaitForOpenSegment(name, decision.WaitTimeout)
 				if err != nil && err == playback.ErrSegmentNotFound {
 					slog.InfoContext(r.Context(), "transcode segment wait timeout", "component", "transcodenode",
 						"segment", name,
@@ -980,7 +985,7 @@ func (s *Server) handleSegment(w http.ResponseWriter, r *http.Request) {
 						seekSeconds,
 						segNum,
 					); restartErr == nil {
-						segPath, err = session.WaitForSegment(name, 30*time.Second)
+						segmentLease, err = session.WaitForOpenSegment(name, 30*time.Second)
 					}
 				}
 				if !ok && session.IsCopyVideo() {
@@ -990,7 +995,7 @@ func (s *Server) handleSegment(w http.ResponseWriter, r *http.Request) {
 		} else if session.IsRunning() {
 			// Non-numbered segment (e.g., init.mp4 for fMP4 HLS).
 			// Wait briefly — the init segment is written almost immediately.
-			segPath, err = session.WaitForSegment(name, 10*time.Second)
+			segmentLease, err = session.WaitForOpenSegment(name, 10*time.Second)
 		}
 	}
 	if err != nil {
@@ -1000,7 +1005,46 @@ func (s *Server) handleSegment(w http.ResponseWriter, r *http.Request) {

 	w.Header().Set("Cache-Control", "no-store, max-age=0")
 	w.Header().Set("Pragma", "no-cache")
-	http.ServeFile(w, r, segPath)
+	proxied := r.Header.Get(transcodeproxy.RequestHeader) == "1"
+	if proxied {
+		w.Header().Set(transcodeproxy.GenerationHeader, segmentLease.GenerationToken)
+	}
+	defer func() { _ = segmentLease.Close() }()
+	sw := httpstream.NewRollingDeadlineWriter(w)
+	http.ServeContent(sw, r, segmentLease.Info.Name(), segmentLease.Info.ModTime(), segmentLease.File)
+	if !proxied && r.Method == http.MethodGet &&
+		sw.CompletedFullResponse(segmentLease.Info.Size()) {
+		if segmentNumber, parseErr := playback.ParseSegmentNumber(name); parseErr == nil {
+			session.ReportSegmentDownloadedForGeneration(segmentNumber, segmentLease.Generation)
+		}
+	}
+}
+
+// handleSegmentDownloaded records completion only after the central API has
+// delivered the full segment to its downstream playback client. The generation
+// returned with the original segment response prevents a delayed acknowledgement
+// from advancing a replacement FFmpeg timeline.
+func (s *Server) handleSegmentDownloaded(w http.ResponseWriter, r *http.Request) {
+	sessionID := chi.URLParam(r, "session_id")
+	name := chi.URLParam(r, "name")
+	segmentNumber, err := playback.ParseSegmentNumber(name)
+	if err != nil {
+		http.Error(w, "invalid segment", http.StatusBadRequest)
+		return
+	}
+	generationToken := r.Header.Get(transcodeproxy.GenerationHeader)
+	if generationToken == "" {
+		http.Error(w, "invalid segment generation", http.StatusBadRequest)
+		return
+	}
+
+	session, ok := s.acquireSessionTouched(sessionID)
+	if !ok {
+		http.Error(w, "session not found", http.StatusNotFound)
+		return
+	}
+	session.ReportSegmentDownloadedForGenerationToken(segmentNumber, generationToken)
+	w.WriteHeader(http.StatusNoContent)
 }

 func (s *Server) handleForceReload(w http.ResponseWriter, r *http.Request) {
@@ -7,6 +7,7 @@ import (
 	"net/http"
 	"net/http/httptest"
 	"os"
+	"os/exec"
 	"path/filepath"
 	"strings"
 	"testing"
@@ -19,6 +20,7 @@ import (
 	"github.com/Silo-Server/silo-server/internal/nodesessions"
 	"github.com/Silo-Server/silo-server/internal/playback"
 	"github.com/Silo-Server/silo-server/internal/streamtoken"
+	"github.com/Silo-Server/silo-server/internal/transcodeproxy"
 )

 const testSecret = "node-reconstruct-test-secret"
@@ -34,11 +36,122 @@ func newTestServer(t *testing.T) *Server {
 	cfg.Playback.TranscodeDir = t.TempDir()
 	w.SetConfigForTest(cfg)
 	return &Server{
-		watcher:  w,
-		sessions: make(map[string]*playback.TranscodeSession),
+		watcher:    w,
+		sessions:   make(map[string]*playback.TranscodeSession),
+		lastAccess: make(map[string]time.Time),
 	}
 }

+func TestProxiedSegmentCompletionRequiresDownstreamAcknowledgement(t *testing.T) {
+	truePath, err := exec.LookPath("true")
+	if err != nil {
+		t.Skipf("`true` not found in PATH: %v", err)
+	}
+
+	const sessionID = "proxy-completion-session"
+	server := newTestServer(t)
+	outputDir := t.TempDir()
+	session, err := playback.StartTranscode(context.Background(), playback.TranscodeOpts{
+		SessionID:               sessionID,
+		OutputDir:               outputDir,
+		FFmpegPath:              truePath,
+		TargetCodecVideo:        "h264",
+		TargetCodecAudio:        "aac",
+		SegmentDuration:         2,
+		SegmentRetentionSeconds: 600,
+	})
+	if err != nil {
+		t.Fatalf("StartTranscode: %v", err)
+	}
+	t.Cleanup(func() { _ = session.Close() })
+	server.sessions[sessionID] = session
+	server.lastAccess[sessionID] = time.Now()
+
+	const segmentName = "seg_00007.ts"
+	if err := os.WriteFile(filepath.Join(outputDir, segmentName), []byte("complete segment"), 0o600); err != nil {
+		t.Fatal(err)
+	}
+
+	segmentReq := withNodeRouteParams(
+		httptest.NewRequest(http.MethodGet, "/transcode/"+sessionID+"/segment/"+segmentName, nil),
+		map[string]string{"session_id": sessionID, "name": segmentName},
+	)
+	segmentReq.Header.Set(transcodeproxy.RequestHeader, "1")
+	segmentRR := httptest.NewRecorder()
+	server.handleSegment(segmentRR, segmentReq)
+	if segmentRR.Code != http.StatusOK {
+		t.Fatalf("segment status = %d, body = %q", segmentRR.Code, segmentRR.Body.String())
+	}
+	generation := segmentRR.Header().Get(transcodeproxy.GenerationHeader)
+	if generation == "" {
+		t.Fatal("proxied segment omitted generation")
+	}
+	if got := session.LastRequestedSegment(); got != 0 {
+		t.Fatalf("node counted proxy-hop completion as client completion: got %d, want 0", got)
+	}
+
+	ackReq := withNodeRouteParams(
+		httptest.NewRequest(http.MethodPost, "/transcode/"+sessionID+"/segment/"+segmentName+"/downloaded", nil),
+		map[string]string{"session_id": sessionID, "name": segmentName},
+	)
+	ackReq.Header.Set(transcodeproxy.GenerationHeader, generation)
+	ackRR := httptest.NewRecorder()
+	server.handleSegmentDownloaded(ackRR, ackReq)
+	if ackRR.Code != http.StatusNoContent {
+		t.Fatalf("ack status = %d, body = %q", ackRR.Code, ackRR.Body.String())
+	}
+	if got := session.LastRequestedSegment(); got != 7 {
+		t.Fatalf("acknowledged segment = %d, want 7", got)
+	}
+}
+
+func TestProxiedSegmentCompletionRejectsStaleGeneration(t *testing.T) {
+	truePath, err := exec.LookPath("true")
+	if err != nil {
+		t.Skipf("`true` not found in PATH: %v", err)
+	}
+
+	const sessionID = "stale-proxy-completion-session"
+	server := newTestServer(t)
+	session, err := playback.StartTranscode(context.Background(), playback.TranscodeOpts{
+		SessionID:        sessionID,
+		OutputDir:        t.TempDir(),
+		FFmpegPath:       truePath,
+		TargetCodecVideo: "h264",
+		TargetCodecAudio: "aac",
+		SegmentDuration:  2,
+	})
+	if err != nil {
+		t.Fatalf("StartTranscode: %v", err)
+	}
+	t.Cleanup(func() { _ = session.Close() })
+	server.sessions[sessionID] = session
+	server.lastAccess[sessionID] = time.Now()
+
+	const segmentName = "seg_00007.ts"
+	ackReq := withNodeRouteParams(
+		httptest.NewRequest(http.MethodPost, "/transcode/"+sessionID+"/segment/"+segmentName+"/downloaded", nil),
+		map[string]string{"session_id": sessionID, "name": segmentName},
+	)
+	ackReq.Header.Set(transcodeproxy.GenerationHeader, "999")
+	ackRR := httptest.NewRecorder()
+	server.handleSegmentDownloaded(ackRR, ackReq)
+	if ackRR.Code != http.StatusNoContent {
+		t.Fatalf("ack status = %d, body = %q", ackRR.Code, ackRR.Body.String())
+	}
+	if got := session.LastRequestedSegment(); got != 0 {
+		t.Fatalf("stale acknowledgement advanced segment to %d", got)
+	}
+}
+
+func withNodeRouteParams(req *http.Request, values map[string]string) *http.Request {
+	routeCtx := chi.NewRouteContext()
+	for key, value := range values {
+		routeCtx.URLParams.Add(key, value)
+	}
+	return req.WithContext(context.WithValue(req.Context(), chi.RouteCtxKey, routeCtx))
+}
+
 func TestHandleStartRequireReadyRejectsExitedFFmpeg(t *testing.T) {
 	server := newTestServer(t)
 	ffmpegPath := filepath.Join(t.TempDir(), "failing-ffmpeg.sh")
@@ -0,0 +1,110 @@
+// Package transcodeproxy owns the private HTTP completion contract shared by
+// Silo's integrated API proxy and its dedicated proxy node.
+package transcodeproxy
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"net/http"
+	"strconv"
+	"strings"
+	"time"
+)
+
+const (
+	// RequestHeader marks a segment hop whose immediate consumer is another
+	// Silo server, not the playback client. The transcode node defers download
+	// accounting until that server confirms downstream completion.
+	RequestHeader = "X-Silo-Transcode-Proxy"
+	// GenerationHeader carries an opaque session-incarnation and FFmpeg-timeline
+	// token. It is private to Silo hops and must not be forwarded to the client.
+	GenerationHeader = "X-Silo-Transcode-Segment-Generation"
+)
+
+var representationRequestHeaders = []string{
+	"Range",
+	"If-Range",
+	"If-Match",
+	"If-None-Match",
+	"If-Modified-Since",
+	"If-Unmodified-Since",
+}
+
+// PrepareRequest makes the upstream node serve the same byte representation
+// requested by the downstream client and suppresses completion at this hop.
+func PrepareRequest(upstream, downstream *http.Request) {
+	upstream.Header.Set(RequestHeader, "1")
+	for _, name := range representationRequestHeaders {
+		for _, value := range downstream.Header.Values(name) {
+			upstream.Header.Add(name, value)
+		}
+	}
+}
+
+// CopyResponseHeaders forwards public response metadata while retaining the
+// generation token inside the trusted Silo hop.
+func CopyResponseHeaders(dst, src http.Header) {
+	for name, values := range src {
+		if http.CanonicalHeaderKey(name) == GenerationHeader {
+			continue
+		}
+		for _, value := range values {
+			dst.Add(name, value)
+		}
+	}
+}
+
+// FullRepresentationSize resolves the total entity size needed to distinguish
+// a whole-file 206 response from an ordinary partial range.
+func FullRepresentationSize(resp *http.Response) int64 {
+	if resp == nil {
+		return -1
+	}
+	if resp.StatusCode == http.StatusOK {
+		return resp.ContentLength
+	}
+	if resp.StatusCode != http.StatusPartialContent {
+		return -1
+	}
+	_, total, ok := strings.Cut(resp.Header.Get("Content-Range"), "/")
+	if !ok || total == "*" {
+		return -1
+	}
+	size, err := strconv.ParseInt(total, 10, 64)
+	if err != nil || size <= 0 {
+		return -1
+	}
+	return size
+}
+
+// Acknowledge reports a completed downstream response to the transcode node.
+// The opaque generation token makes delayed acknowledgements harmless after a
+// restart or reconstruction.
+func Acknowledge(ctx context.Context, client *http.Client, targetURL, jwtSecret, generation string) error {
+	if client == nil {
+		client = http.DefaultClient
+	}
+	// The downstream request is commonly canceled as soon as its response body
+	// reaches the client, while the handler is still issuing this acknowledgement.
+	// Preserve request-scoped values, but give the post-response hop its own
+	// bounded lifetime so a successful delivery is not mistaken for a disconnect.
+	ackCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second)
+	defer cancel()
+	req, err := http.NewRequestWithContext(ackCtx, http.MethodPost, targetURL+"/downloaded", nil)
+	if err != nil {
+		return fmt.Errorf("build acknowledgement: %w", err)
+	}
+	req.Header.Set("Authorization", "Bearer "+jwtSecret)
+	req.Header.Set(GenerationHeader, generation)
+	resp, err := client.Do(req)
+	if err != nil {
+		return fmt.Errorf("send acknowledgement: %w", err)
+	}
+	defer func() { _ = resp.Body.Close() }()
+	_, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, 4<<10))
+	if resp.StatusCode < http.StatusOK || resp.StatusCode >= http.StatusMultipleChoices {
+		return fmt.Errorf("acknowledgement status %d", resp.StatusCode)
+	}
+	return nil
+}
@@ -0,0 +1,42 @@
+package transcodeproxy
+
+import (
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+)
+
+func TestAcknowledgeSurvivesCompletedDownstreamRequest(t *testing.T) {
+	const (
+		jwtSecret  = "test-secret"
+		generation = "incarnation:generation"
+	)
+
+	acknowledged := make(chan struct{}, 1)
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.Method != http.MethodPost {
+			t.Errorf("method = %q, want POST", r.Method)
+		}
+		if got := r.Header.Get("Authorization"); got != "Bearer "+jwtSecret {
+			t.Errorf("Authorization = %q", got)
+		}
+		if got := r.Header.Get(GenerationHeader); got != generation {
+			t.Errorf("generation = %q, want %q", got, generation)
+		}
+		acknowledged <- struct{}{}
+		w.WriteHeader(http.StatusNoContent)
+	}))
+	t.Cleanup(server.Close)
+
+	requestCtx, cancelRequest := context.WithCancel(context.Background())
+	cancelRequest()
+	if err := Acknowledge(requestCtx, server.Client(), server.URL+"/segment", jwtSecret, generation); err != nil {
+		t.Fatalf("Acknowledge() error = %v", err)
+	}
+	select {
+	case <-acknowledged:
+	default:
+		t.Fatal("acknowledgement request was not delivered")
+	}
+}
@@ -16,6 +16,7 @@ import {
 const KEYS = [
  "playback.ffmpeg_path",
  "playback.transcode_dir",
+  "playback.segment_retention_seconds",
  "playback.hw_accel",
  "playback.hw_device",
  "playback.transcode_enabled",
@@ -197,6 +198,13 @@ export default function PlaybackSettings() {
        </FieldGroup>

        <FieldGroup label="Segments">
+          <SettingField
+            label="Transcode Back Buffer (seconds)"
+            type="number"
+            hint="Keeps this much already-downloaded media for instant backward seeking, then reclaims older transcode segments. Use 0 to disable cleanup; enabled values must be at least 120 seconds. Pair with transcode throttling to bound both behind- and ahead-of-client disk usage."
+            value={form.getValue("playback.segment_retention_seconds")}
+            onChange={(v) => form.setValue("playback.segment_retention_seconds", v)}
+          />
          <SettingField
            label="Chapter Thumbnail Workers"
            type="number"
Author	SHA1	Message	Date
Quick104	fd3d737a58	fix(playback): trust completed segment writes	2026-08-11 15:02:38 -04:00
Quick104	86a02a6f61	fix(playback): bound sparse seek cleanup	2026-08-11 14:55:27 -04:00
Quick104	1cbb59d0fa	fix(playback): finish segment acknowledgements	2026-08-11 14:45:33 -04:00
Quick104	874a26de0f	fix(playback): confirm proxy downstream delivery	2026-08-11 14:30:55 -04:00
Quick104	0544213730	fix(playback): verify segment completion before pruning	2026-08-11 14:21:33 -04:00
Quick104	e05c30c637	fix(playback): keep restart buffers prunable	2026-08-11 13:42:43 -04:00
Quick104	7f6225bb11	fix(playback): remove redundant retention floor assignment	2026-08-11 13:21:53 -04:00
Quick104	977104eac5	Merge remote-tracking branch 'origin/main' into t3code/pr565-review	2026-08-11 13:14:44 -04:00
Quick104	dffa6ae34e	Merge origin/main into codex/bound-transcode-segments	2026-08-11 13:05:12 -04:00
Quick104	e2899ff619	fix(playback): preserve retained segments across restarts	2026-08-11 12:59:25 -04:00
Quick104	f524c48d19	fix(playback): close retention gaps across ranges and seeks	2026-08-08 18:49:27 -04:00
Quick104	c26ab22809	test(playback): strengthen retention regression coverage	2026-08-08 16:08:31 -04:00
Quick104	0a3adf6552	fix(playback): harden segment retention lifecycle	2026-08-08 15:59:56 -04:00
Quick104	35de68b589	fix(playback): synchronize prune option snapshot	2026-08-08 15:29:02 -04:00
Quick104	6a3fcef165	fix(playback): prune downloaded transcode segments	2026-08-08 15:23:16 -04:00