fix(distributed): enforce staging admission

Propagate multimodal staging failures before inference and claim matching ephemeral HTTP cache entries. Fall back to PUT when an older worker does not support claims.

Assisted-by: Codex:gpt-6

core/services/nodes/file_stager_http.go

@@ -120,7 +120,9 @@ func (h *HTTPFileStager) EnsureRemote(ctx context.Context, nodeID, localPath, ke
 	}
 
 	// Probe: check if the remote already has the file with matching content hash.
-	if remotePath, ok := h.probeExisting(ctx, addr, localPath, key); ok {
+	if remotePath, ok, probeErr := h.probeExisting(ctx, addr, localPath, key); probeErr != nil {
+		return "", fmt.Errorf("claiming existing file on node %s: %w", nodeID, probeErr)
+	} else if ok {
 		xlog.Info("Upload skipped (file already exists with matching hash)", "node", nodeID, "key", key, "remotePath", remotePath)
 		return remotePath, nil
 	}
@@ -469,14 +471,15 @@ func isTransientError(err error) bool {
 
 // probeExisting sends a HEAD request to check if the remote already has the
 // file with a matching SHA-256 hash. Returns the remote path and true if the
-// upload can be skipped. Any errors (including 405 from older servers) silently
-// fall through so the caller proceeds with a normal PUT.
-func (h *HTTPFileStager) probeExisting(ctx context.Context, addr, localPath, key string) (string, bool) {
+// upload can be skipped. HEAD and hash errors fall through to a normal PUT.
+// Matching ephemeral files are claimed first; a 404 or 405 claim response
+// identifies an older worker and also falls through to PUT.
+func (h *HTTPFileStager) probeExisting(ctx context.Context, addr, localPath, key string) (string, bool, error) {
 	url := fmt.Sprintf("http://%s/v1/files/%s", addr, key)
 
 	req, err := http.NewRequestWithContext(ctx, http.MethodHead, url, nil)
 	if err != nil {
-		return "", false
+		return "", false, nil
 	}
 	if h.token != "" {
 		req.Header.Set("Authorization", "Bearer "+h.token)
@@ -484,18 +487,18 @@ func (h *HTTPFileStager) probeExisting(ctx context.Context, addr, localPath, key
 
 	resp, err := h.client.Do(req)
 	if err != nil {
-		return "", false
+		return "", false, nil
 	}
 	defer resp.Body.Close()
 
 	if resp.StatusCode != http.StatusOK {
-		return "", false
+		return "", false, nil
 	}
 
 	remotePath := resp.Header.Get(HeaderLocalPath)
 	remoteHash := resp.Header.Get(HeaderContentSHA256)
 	if remotePath == "" || remoteHash == "" {
-		return "", false
+		return "", false, nil
 	}
 
 	// A 200 with a content hash is proof the worker is alive and serving right
@@ -505,14 +508,53 @@ func (h *HTTPFileStager) probeExisting(ctx context.Context, addr, localPath, key
 
 	localHash, err := hashLocalCached(ctx, localPath)
 	if err != nil {
-		return "", false
+		return "", false, nil
 	}
 
 	if localHash != remoteHash {
-		return "", false
+		return "", false, nil
 	}
 
-	return remotePath, true
+	if strings.HasPrefix(key, "ephemeral/") {
+		claimed, err := h.claimExisting(ctx, addr, key)
+		if err != nil {
+			return "", false, err
+		}
+		if !claimed {
+			return "", false, nil
+		}
+	}
+
+	return remotePath, true, nil
+}
+
+func (h *HTTPFileStager) claimExisting(ctx context.Context, addr, key string) (bool, error) {
+	claimURL := (&url.URL{
+		Scheme:   "http",
+		Host:     addr,
+		Path:     "/v1/files/" + key,
+		RawQuery: "claim=1",
+	}).String()
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, claimURL, nil)
+	if err != nil {
+		return false, fmt.Errorf("creating claim request for %q: %w", key, err)
+	}
+	if h.token != "" {
+		req.Header.Set("Authorization", "Bearer "+h.token)
+	}
+	resp, err := h.client.Do(req)
+	if err != nil {
+		return false, fmt.Errorf("claiming %q: %w", key, err)
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode == http.StatusNotFound || resp.StatusCode == http.StatusMethodNotAllowed {
+		return false, nil
+	}
+	if resp.StatusCode < http.StatusOK || resp.StatusCode >= http.StatusMultipleChoices {
+		body, _ := io.ReadAll(io.LimitReader(resp.Body, 4096))
+		return false, fmt.Errorf("claiming %q: status %d: %s", key, resp.StatusCode, strings.TrimSpace(string(body)))
+	}
+	return true, nil
 }
 
 // hashChunkSize is how much of a file is hashed between activity ticks and

core/services/nodes/file_staging_client.go

@@ -149,7 +149,11 @@ func (f *FileStagingClient) Predict(ctx context.Context, in *pb.PredictOptions,
 	lifecycle := f.newStagedInputLifecycle()
 	defer lifecycle.release()
 	in = proto.Clone(in).(*pb.PredictOptions)
-	in = f.stageMultimodalInputs(ctx, lifecycle, in)
+	var err error
+	in, err = f.stageMultimodalInputs(ctx, lifecycle, in)
+	if err != nil {
+		return nil, err
+	}
 	return f.Backend.Predict(ctx, in, opts...)
 }
 
@@ -157,7 +161,11 @@ func (f *FileStagingClient) PredictStream(ctx context.Context, in *pb.PredictOpt
 	lifecycle := f.newStagedInputLifecycle()
 	defer lifecycle.release()
 	in = proto.Clone(in).(*pb.PredictOptions)
-	in = f.stageMultimodalInputs(ctx, lifecycle, in)
+	var err error
+	in, err = f.stageMultimodalInputs(ctx, lifecycle, in)
+	if err != nil {
+		return err
+	}
 	return f.Backend.PredictStream(ctx, in, fn, opts...)
 }
 
@@ -528,11 +536,21 @@ func (f *FileStagingClient) stageMultimodalInputs(
 	ctx context.Context,
 	lifecycle *stagedInputLifecycle,
 	in *pb.PredictOptions,
-) *pb.PredictOptions {
-	in.Images = f.stagePathSlice(ctx, lifecycle, in.Images, "inputs")
-	in.Videos = f.stagePathSlice(ctx, lifecycle, in.Videos, "inputs")
-	in.Audios = f.stagePathSlice(ctx, lifecycle, in.Audios, "inputs")
-	return in
+) (*pb.PredictOptions, error) {
+	var err error
+	in.Images, err = f.stagePathSlice(ctx, lifecycle, in.Images, "inputs")
+	if err != nil {
+		return nil, fmt.Errorf("staging predict images: %w", err)
+	}
+	in.Videos, err = f.stagePathSlice(ctx, lifecycle, in.Videos, "inputs")
+	if err != nil {
+		return nil, fmt.Errorf("staging predict videos: %w", err)
+	}
+	in.Audios, err = f.stagePathSlice(ctx, lifecycle, in.Audios, "inputs")
+	if err != nil {
+		return nil, fmt.Errorf("staging predict audios: %w", err)
+	}
+	return in, nil
 }
 
 func (f *FileStagingClient) stagePathSlice(
@@ -540,22 +558,20 @@ func (f *FileStagingClient) stagePathSlice(
 	lifecycle *stagedInputLifecycle,
 	paths []string,
 	category string,
-) []string {
+) ([]string, error) {
 	result := make([]string, len(paths))
 	for i, p := range paths {
 		if isFilePath(p) {
 			backendPath, err := f.stageInputFile(ctx, lifecycle, p, category)
 			if err != nil {
-				xlog.Warn("Failed to stage multimodal file, passing through", "path", p, "error", err)
-				result[i] = p
-				continue
+				return nil, fmt.Errorf("staging %q: %w", p, err)
 			}
 			result[i] = backendPath
 		} else {
 			result[i] = p
 		}
 	}
-	return result
+	return result, nil
 }
 
 // isFilePath checks if a string looks like a local file path (not base64 or URL).

core/services/nodes/file_staging_lifecycle_test.go

@@ -19,6 +19,7 @@ const fullUUIDPattern = `[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-
 type lifecycleStager struct {
 	fakeFileStager
 	ensureErr          error
+	ensureErrAt        int
 	releaseErr         error
 	releasedKeys       []string
 	releaseCtxErr      []error
@@ -28,7 +29,7 @@ type lifecycleStager struct {
 
 func (s *lifecycleStager) EnsureRemote(ctx context.Context, nodeID, localPath, key string) (string, error) {
 	s.fakeFileStager.EnsureRemote(ctx, nodeID, localPath, key)
-	if s.ensureErr != nil {
+	if s.ensureErr != nil && (s.ensureErrAt == 0 || len(s.ensureCalls) == s.ensureErrAt) {
 		return "", s.ensureErr
 	}
 	return "/remote/" + key, nil
@@ -47,11 +48,16 @@ type lifecycleBackend struct {
 	grpc.Backend
 	predictResult *pb.Reply
 	predictErr    error
+	predictCalls  int
+	predictInput  *pb.PredictOptions
+	streamCalls   int
 	streamBlock   <-chan struct{}
 	streamStarted chan<- struct{}
 }
 
-func (b *lifecycleBackend) Predict(_ context.Context, _ *pb.PredictOptions, _ ...ggrpc.CallOption) (*pb.Reply, error) {
+func (b *lifecycleBackend) Predict(_ context.Context, in *pb.PredictOptions, _ ...ggrpc.CallOption) (*pb.Reply, error) {
+	b.predictCalls++
+	b.predictInput = proto.Clone(in).(*pb.PredictOptions)
 	if b.predictResult == nil {
 		b.predictResult = &pb.Reply{}
 	}
@@ -59,6 +65,7 @@ func (b *lifecycleBackend) Predict(_ context.Context, _ *pb.PredictOptions, _ ..
 }
 
 func (b *lifecycleBackend) PredictStream(_ context.Context, _ *pb.PredictOptions, _ func(*pb.Reply), _ ...ggrpc.CallOption) error {
+	b.streamCalls++
 	if b.streamStarted != nil {
 		b.streamStarted <- struct{}{}
 	}
@@ -234,6 +241,40 @@ var _ = Describe("FileStagingClient request lifecycle", func() {
 		Expect(stager.releasedKeys).To(Equal(keysFromEnsureCalls(stager.ensureCalls)))
 	})
 
+	It("does not invoke predict when multimodal staging fails", func(ctx SpecContext) {
+		uploadErr := errors.New("ephemeral capacity exceeded")
+		stager := &lifecycleStager{ensureErr: uploadErr, ensureErrAt: 2}
+		backend := &lifecycleBackend{}
+		client := NewFileStagingClient(backend, stager, "worker-1")
+		request := &pb.PredictOptions{Images: []string{"/tmp/first.png", "/tmp/second.png"}}
+		original := proto.Clone(request)
+
+		result, err := client.Predict(ctx, request)
+
+		Expect(result).To(BeNil())
+		Expect(err).To(MatchError(ContainSubstring("ephemeral capacity exceeded")))
+		Expect(backend.predictCalls).To(BeZero())
+		Expect(proto.Equal(request, original)).To(BeTrue())
+		Expect(stager.ensureCalls).To(HaveLen(2))
+		Expect(stager.releasedKeys).To(Equal(keysFromEnsureCalls(stager.ensureCalls)))
+	})
+
+	It("does not invoke streaming predict when multimodal staging fails", func(ctx SpecContext) {
+		uploadErr := errors.New("ephemeral capacity exceeded")
+		stager := &lifecycleStager{ensureErr: uploadErr}
+		backend := &lifecycleBackend{}
+		client := NewFileStagingClient(backend, stager, "worker-1")
+		request := &pb.PredictOptions{Audios: []string{"/tmp/audio.wav"}}
+		original := proto.Clone(request)
+
+		err := client.PredictStream(ctx, request, func(*pb.Reply) {})
+
+		Expect(err).To(MatchError(ContainSubstring("ephemeral capacity exceeded")))
+		Expect(backend.streamCalls).To(BeZero())
+		Expect(proto.Equal(request, original)).To(BeTrue())
+		Expect(stager.releasedKeys).To(Equal(keysFromEnsureCalls(stager.ensureCalls)))
+	})
+
 	It("uses an active bounded cleanup context after caller cancellation", func() {
 		ctx, cancel := context.WithCancel(context.Background())
 		cancel()

core/services/nodes/file_transfer_server.go

@@ -136,6 +136,8 @@ func startFileTransferServer(lis net.Listener, stagingDir, modelsDir, dataDir, t
 		case http.MethodPost:
 			if key == "temp" {
 				handleAllocTemp(w, r, stagingDir)
+			} else if r.URL.Query().Get("claim") == "1" {
+				handleClaimWithCapacity(w, r, stagingDir, key, capacity)
 			} else {
 				http.Error(w, "not found", http.StatusNotFound)
 			}
@@ -196,6 +198,40 @@ func startFileTransferServer(lis net.Listener, stagingDir, modelsDir, dataDir, t
 	return server, nil
 }
 
+// handleClaimWithCapacity marks an existing ephemeral file as owned by the
+// request that just verified its content.
+func handleClaimWithCapacity(w http.ResponseWriter, _ *http.Request, stagingDir, key string, capacity EphemeralCapacity) {
+	if err := validateEphemeralReleaseKey(key); err != nil {
+		http.Error(w, err.Error(), http.StatusBadRequest)
+		return
+	}
+	filePath := filepath.Join(stagingDir, filepath.FromSlash(key))
+	if err := validatePathInDir(filePath, stagingDir); err != nil {
+		http.Error(w, err.Error(), http.StatusBadRequest)
+		return
+	}
+	info, err := os.Lstat(filePath)
+	if err != nil {
+		if os.IsNotExist(err) {
+			http.Error(w, "not found", http.StatusNotFound)
+		} else {
+			http.Error(w, err.Error(), http.StatusInternalServerError)
+		}
+		return
+	}
+	if !info.Mode().IsRegular() {
+		http.Error(w, "ephemeral path is not a regular file", http.StatusBadRequest)
+		return
+	}
+	if capacity != nil {
+		if err := capacity.Claim(filePath); err != nil {
+			http.Error(w, err.Error(), http.StatusInsufficientStorage)
+			return
+		}
+	}
+	w.WriteHeader(http.StatusNoContent)
+}
+
 func handleRelease(w http.ResponseWriter, _ *http.Request, stagingDir, key string) {
 	handleReleaseWithCapacity(w, nil, stagingDir, key, nil)
 }
@@ -300,6 +336,7 @@ type contentRange struct {
 type EphemeralCapacity interface {
 	Reserve(path string, size int64) error
 	Commit(path string) error
+	Claim(path string) error
 	Release(path string) error
 	CapacityWriter(path string, destination io.Writer) (io.WriteCloser, error)
 }

core/services/nodes/file_transfer_server_test.go

@@ -17,6 +17,7 @@ import (
 	"sync"
 	"time"
 
+	pb "github.com/mudler/LocalAI/pkg/grpc/proto"
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
 )
@@ -25,6 +26,8 @@ type recordingEphemeralCapacity struct {
 	reserved   int64
 	reserveErr error
 	writerErr  error
+	claimCalls []string
+	claimErr   error
 }
 
 type nopWriteCloser struct{ io.Writer }
@@ -43,6 +46,10 @@ func (g *recordingEphemeralCapacity) Reserve(_ string, size int64) error {
 
 func (*recordingEphemeralCapacity) Commit(string) error  { return nil }
 func (*recordingEphemeralCapacity) Release(string) error { return nil }
+func (g *recordingEphemeralCapacity) Claim(path string) error {
+	g.claimCalls = append(g.claimCalls, path)
+	return g.claimErr
+}
 func (g *recordingEphemeralCapacity) CapacityWriter(_ string, destination io.Writer) (io.WriteCloser, error) {
 	if g.writerErr != nil {
 		return failingWriteCloser{err: g.writerErr}, nil
@@ -494,6 +501,148 @@ var _ = Describe("FileTransferServer", func() {
 	// --- EnsureRemote skip tests ---
 
 	Describe("EnsureRemote skip-if-exists", func() {
+		It("claims a matching ephemeral file before returning the worker path", func() {
+			stagingDir := GinkgoT().TempDir()
+			modelsDir := GinkgoT().TempDir()
+			dataDir := GinkgoT().TempDir()
+			guard := &recordingEphemeralCapacity{}
+			key := "ephemeral/audio/request/input.wav"
+			remotePath := filepath.Join(stagingDir, filepath.FromSlash(key))
+			content := []byte("already on worker")
+			Expect(os.MkdirAll(filepath.Dir(remotePath), 0o750)).To(Succeed())
+			Expect(os.WriteFile(remotePath, content, 0o600)).To(Succeed())
+			Expect(os.WriteFile(remotePath+hashSidecarSuffix, []byte(sha256Hex(content)), 0o600)).To(Succeed())
+			localPath := filepath.Join(GinkgoT().TempDir(), "input.wav")
+			Expect(os.WriteFile(localPath, content, 0o600)).To(Succeed())
+
+			mux := http.NewServeMux()
+			mux.HandleFunc("/v1/files/", func(w http.ResponseWriter, r *http.Request) {
+				requestKey := strings.TrimPrefix(r.URL.Path, "/v1/files/")
+				switch r.Method {
+				case http.MethodHead:
+					handleHead(w, r, stagingDir, modelsDir, dataDir, requestKey)
+				case http.MethodPost:
+					handleClaimWithCapacity(w, r, stagingDir, requestKey, guard)
+				default:
+					http.Error(w, "unexpected upload", http.StatusInternalServerError)
+				}
+			})
+			ts := httptest.NewServer(mux)
+			DeferCleanup(ts.Close)
+			stager := NewHTTPFileStager(func(string) (string, error) {
+				return strings.TrimPrefix(ts.URL, "http://"), nil
+			}, "")
+
+			for range 2 {
+				path, err := stager.EnsureRemote(context.Background(), "node-1", localPath, key)
+				Expect(err).NotTo(HaveOccurred())
+				Expect(path).To(Equal(remotePath))
+			}
+			Expect(guard.claimCalls).To(Equal([]string{remotePath, remotePath}))
+		})
+
+		It("propagates an ephemeral cache-hit claim failure", func() {
+			content := []byte("already on worker")
+			localPath := filepath.Join(GinkgoT().TempDir(), "input.wav")
+			Expect(os.WriteFile(localPath, content, 0o600)).To(Succeed())
+			mux := http.NewServeMux()
+			mux.HandleFunc("/v1/files/", func(w http.ResponseWriter, r *http.Request) {
+				if r.Method == http.MethodHead {
+					w.Header().Set(HeaderLocalPath, "/remote/ephemeral/audio/request/input.wav")
+					w.Header().Set(HeaderContentSHA256, sha256Hex(content))
+					w.WriteHeader(http.StatusOK)
+					return
+				}
+				http.Error(w, "ephemeral capacity exceeded", http.StatusInsufficientStorage)
+			})
+			ts := httptest.NewServer(mux)
+			DeferCleanup(ts.Close)
+			stager := NewHTTPFileStager(func(string) (string, error) {
+				return strings.TrimPrefix(ts.URL, "http://"), nil
+			}, "")
+
+			path, err := stager.EnsureRemote(context.Background(), "node-1", localPath, "ephemeral/audio/request/input.wav")
+
+			Expect(path).To(BeEmpty())
+			Expect(err).To(MatchError(ContainSubstring("ephemeral capacity exceeded")))
+		})
+
+		DescribeTable("uploads a matching ephemeral file when an old worker cannot claim it",
+			func(claimStatus int) {
+				stagingDir := GinkgoT().TempDir()
+				content := []byte("compatible upload")
+				localPath := filepath.Join(GinkgoT().TempDir(), "input.wav")
+				Expect(os.WriteFile(localPath, content, 0o600)).To(Succeed())
+				cacheHitPath := filepath.Join(stagingDir, "ephemeral", "audio", "stale", "input.wav")
+				putCalls := 0
+				putRemotePath := ""
+				mux := http.NewServeMux()
+				mux.HandleFunc("/v1/files/", func(w http.ResponseWriter, r *http.Request) {
+					switch r.Method {
+					case http.MethodHead:
+						w.Header().Set(HeaderLocalPath, cacheHitPath)
+						w.Header().Set(HeaderContentSHA256, sha256Hex(content))
+						w.WriteHeader(http.StatusOK)
+					case http.MethodPost:
+						http.Error(w, "claim unsupported", claimStatus)
+					case http.MethodPut:
+						putCalls++
+						key := strings.TrimPrefix(r.URL.Path, "/v1/files/")
+						putRemotePath = filepath.Join(stagingDir, filepath.FromSlash(key))
+						handleUpload(w, r, stagingDir, "", "", key, 0)
+					case http.MethodDelete:
+						w.WriteHeader(http.StatusNoContent)
+					}
+				})
+				ts := httptest.NewServer(mux)
+				DeferCleanup(ts.Close)
+				stager := NewHTTPFileStager(func(string) (string, error) {
+					return strings.TrimPrefix(ts.URL, "http://"), nil
+				}, "")
+
+				backend := &lifecycleBackend{}
+				client := NewFileStagingClient(backend, stager, "node-1")
+				request := &pb.PredictOptions{Audios: []string{localPath}}
+				_, err := client.Predict(context.Background(), request)
+
+				Expect(err).NotTo(HaveOccurred())
+				Expect(putCalls).To(Equal(1))
+				Expect(backend.predictInput).NotTo(BeNil())
+				Expect(backend.predictInput.Audios).To(Equal([]string{putRemotePath}))
+			},
+			Entry("404", http.StatusNotFound),
+			Entry("405", http.StatusMethodNotAllowed),
+		)
+
+		It("keeps matching model probes read-only", func() {
+			content := []byte("model")
+			localPath := filepath.Join(GinkgoT().TempDir(), "model.bin")
+			Expect(os.WriteFile(localPath, content, 0o600)).To(Succeed())
+			unexpectedWrites := 0
+			mux := http.NewServeMux()
+			mux.HandleFunc("/v1/files/", func(w http.ResponseWriter, r *http.Request) {
+				if r.Method != http.MethodHead {
+					unexpectedWrites++
+					http.Error(w, "unexpected write", http.StatusInternalServerError)
+					return
+				}
+				w.Header().Set(HeaderLocalPath, "/models/tracking/model.bin")
+				w.Header().Set(HeaderContentSHA256, sha256Hex(content))
+				w.WriteHeader(http.StatusOK)
+			})
+			ts := httptest.NewServer(mux)
+			DeferCleanup(ts.Close)
+			stager := NewHTTPFileStager(func(string) (string, error) {
+				return strings.TrimPrefix(ts.URL, "http://"), nil
+			}, "")
+
+			path, err := stager.EnsureRemote(context.Background(), "node-1", localPath, "models/tracking/model.bin")
+
+			Expect(err).NotTo(HaveOccurred())
+			Expect(path).To(Equal("/models/tracking/model.bin"))
+			Expect(unexpectedWrites).To(BeZero())
+		})
+
 		It("skips upload when file exists with matching hash", func() {
 			ts, stagingDir, _, _ := setupTestServer("tok", 0)
 

core/services/worker/file_staging_release_test.go

@@ -2,15 +2,19 @@ package worker
 
 import (
 	"context"
+	"crypto/sha256"
 	"encoding/json"
 	"errors"
+	"fmt"
 	"io"
+	"net"
 	"os"
 	"path/filepath"
 	"strings"
 	"time"
 
 	"github.com/mudler/LocalAI/core/services/messaging"
+	"github.com/mudler/LocalAI/core/services/nodes"
 	"github.com/mudler/LocalAI/core/services/storage"
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
@@ -70,6 +74,48 @@ func (m *releaseMessagingClient) IsConnected() bool { return true }
 func (m *releaseMessagingClient) Close()            {}
 
 var _ = Describe("Worker exact-key staging release", func() {
+	It("protects a startup-accounted HTTP cache hit through authenticated repeated probes", func() {
+		stagingDir := GinkgoT().TempDir()
+		root := filepath.Join(stagingDir, "ephemeral")
+		key := "ephemeral/audio/request-id/input.wav"
+		remotePath := filepath.Join(stagingDir, filepath.FromSlash(key))
+		content := []byte("data")
+		Expect(os.MkdirAll(filepath.Dir(remotePath), 0o750)).To(Succeed())
+		Expect(os.WriteFile(remotePath, content, 0o600)).To(Succeed())
+		hash := sha256.Sum256(content)
+		Expect(os.WriteFile(remotePath+".sha256", []byte(fmt.Sprintf("%x", hash)), 0o600)).To(Succeed())
+		old := time.Now().Add(-2 * time.Hour)
+		for _, path := range []string{remotePath, remotePath + ".sha256", filepath.Dir(remotePath)} {
+			Expect(os.Chtimes(path, old, old)).To(Succeed())
+		}
+		guard, err := NewEphemeralCapacityGuard([]string{root}, int64(len(content)+sha256.Size*2), 0)
+		Expect(err).NotTo(HaveOccurred())
+
+		listener, err := net.Listen("tcp", "127.0.0.1:0")
+		Expect(err).NotTo(HaveOccurred())
+		addr := listener.Addr().String()
+		Expect(listener.Close()).To(Succeed())
+		server, err := nodes.StartFileTransferServerWithCapacity(addr, stagingDir, GinkgoT().TempDir(), GinkgoT().TempDir(), "secret", 0, nil, guard)
+		Expect(err).NotTo(HaveOccurred())
+		DeferCleanup(nodes.ShutdownFileTransferServer, server)
+
+		localPath := filepath.Join(GinkgoT().TempDir(), "input.wav")
+		Expect(os.WriteFile(localPath, content, 0o600)).To(Succeed())
+		stager := nodes.NewHTTPFileStager(func(string) (string, error) { return addr, nil }, "secret")
+		for range 2 {
+			path, ensureErr := stager.EnsureRemote(context.Background(), "worker", localPath, key)
+			Expect(ensureErr).NotTo(HaveOccurred())
+			Expect(path).To(Equal(remotePath))
+		}
+
+		CleanEphemeralRoots([]string{root}, time.Hour, guard)
+		Expect(remotePath).To(BeAnExistingFile())
+
+		Expect(stager.ReleaseRemote(context.Background(), "worker", key)).To(Succeed())
+		CleanEphemeralRoots([]string{root}, time.Hour, guard)
+		Expect(remotePath).NotTo(BeAnExistingFile())
+	})
+
 	It("claims a startup-scanned cache hit against stale recovery until release", func() {
 		cacheDir := GinkgoT().TempDir()
 		root := filepath.Join(cacheDir, "ephemeral")