diff --git a/pkg/upload/asyncconf.go b/pkg/upload/asyncconf.go
new file mode 100644
index 00000000000..bf561ced14f
--- /dev/null
+++ b/pkg/upload/asyncconf.go
@@ -0,0 +1,45 @@
+package upload
+
+import (
+	"github.com/mitchellh/mapstructure"
+)
+
+// AsyncConf is how a service asks for async uploads: whether they are enabled,
+// and the consumer subscription to use if they are.
+type AsyncConf struct {
+	Enabled       bool
+	ConsumerGroup string
+	NumConsumers  int
+	// MountID is the storage id this provider answers for, used to drop
+	// postprocessing events belonging to other storages.
+	MountID string
+}
+
+// AsyncConfFromDriverConf reads the postprocessing settings off the driver's own
+// config keys, so the coordinator and the driver cannot disagree about them.
+func AsyncConfFromDriverConf(driverConf map[string]interface{}) AsyncConf {
+	if driverConf == nil {
+		return AsyncConf{}
+	}
+	var ac struct {
+		AsyncFileUploads bool   `mapstructure:"asyncfileuploads"`
+		MountID          string `mapstructure:"mount_id"`
+		Events           struct {
+			NumConsumers  int    `mapstructure:"numconsumers"`
+			ConsumerGroup string `mapstructure:"consumer_group"`
+		} `mapstructure:"events"`
+	}
+	_ = mapstructure.Decode(driverConf, &ac)
+	group := ac.Events.ConsumerGroup
+	if group == "" {
+		// decomposedfs's default (options.go:177). The coordinator takes over the
+		// driver's subscription, so it must land in the same group.
+		group = "dcfs"
+	}
+	return AsyncConf{
+		Enabled:       ac.AsyncFileUploads,
+		ConsumerGroup: group,
+		NumConsumers:  ac.Events.NumConsumers,
+		MountID:       ac.MountID,
+	}
+}
diff --git a/pkg/upload/asyncconf_test.go b/pkg/upload/asyncconf_test.go
new file mode 100644
index 00000000000..b7e357ffc60
--- /dev/null
+++ b/pkg/upload/asyncconf_test.go
@@ -0,0 +1,39 @@
+package upload
+
+import (
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+var _ = Describe("AsyncConfFromDriverConf", func() {
+	It("is disabled unless the driver asks for async uploads", func() {
+		Expect(AsyncConfFromDriverConf(nil).Enabled).To(BeFalse())
+		Expect(AsyncConfFromDriverConf(map[string]interface{}{"root": "/x"}).Enabled).To(BeFalse())
+		Expect(AsyncConfFromDriverConf(map[string]interface{}{"asyncfileuploads": false}).Enabled).To(BeFalse())
+		Expect(AsyncConfFromDriverConf(map[string]interface{}{"asyncfileuploads": true}).Enabled).To(BeTrue())
+	})
+
+	// The coordinator takes over the driver's subscription, so it has to resolve the
+	// group to the same value the driver does, default included.
+	It("defaults the consumer group to the driver's default", func() {
+		Expect(AsyncConfFromDriverConf(map[string]interface{}{}).ConsumerGroup).To(Equal("dcfs"))
+		Expect(AsyncConfFromDriverConf(map[string]interface{}{
+			"events": map[string]interface{}{"numconsumers": 3},
+		}).ConsumerGroup).To(Equal("dcfs"))
+	})
+
+	It("reads an explicit subscription", func() {
+		ac := AsyncConfFromDriverConf(map[string]interface{}{
+			"asyncfileuploads": true,
+			"mount_id":         "storage-users-1",
+			"events": map[string]interface{}{
+				"consumer_group": "custom",
+				"numconsumers":   4,
+			},
+		})
+
+		Expect(ac.ConsumerGroup).To(Equal("custom"))
+		Expect(ac.NumConsumers).To(Equal(4))
+		Expect(ac.MountID).To(Equal("storage-users-1"))
+	})
+})
diff --git a/pkg/upload/filestore.go b/pkg/upload/filestore.go
new file mode 100644
index 00000000000..1f7e493c4b5
--- /dev/null
+++ b/pkg/upload/filestore.go
@@ -0,0 +1,205 @@
+package upload
+
+import (
+	"context"
+	"encoding/json"
+	iofs "io/fs"
+	"os"
+	"path/filepath"
+	"strings"
+	"syscall"
+
+	"github.com/google/uuid"
+	"github.com/mitchellh/mapstructure"
+	"github.com/pkg/errors"
+	"github.com/rs/zerolog"
+	tusd "github.com/tus/tusd/v2/pkg/handler"
+
+	"github.com/owncloud/reva/v2/pkg/appctx"
+)
+
+// TokenOptions carries the JWT-signing configuration needed to produce transfer
+// URLs for the postprocessing service.
+type TokenOptions struct {
+	DownloadEndpoint     string
+	DataGatewayEndpoint  string
+	TransferSharedSecret string
+	TransferExpires      int64
+}
+
+// SessionStore abstracts upload-session persistence for the Coordinator.
+type SessionStore interface {
+	New(ctx context.Context) Session
+	Get(ctx context.Context, id string) (Session, error)
+	List(ctx context.Context) ([]Session, error)
+}
+
+// FileStore is a filesystem-backed SessionStore. Sessions are stored as a pair
+// of files in the upload directory:
+//
+//   - <id>.info  — JSON-encoded tusd.FileInfo
+//   - <id>       — staged binary bytes
+//
+// This is the same on-disk format used by OcisStore so existing sessions
+// survive a rolling deploy that switches to FileStore.
+type FileStore struct {
+	uploadDir string
+	opts      TokenOptions
+	log       *zerolog.Logger
+}
+
+// FileStoreFromDriverConf builds a FileStore from a reva driver config map.
+// Returns nil if the config carries no upload path (driver does not support
+// coordinated uploads). Each service that mounts the same driver calls this
+// independently.
+func FileStoreFromDriverConf(driverConf map[string]interface{}, log *zerolog.Logger) *FileStore {
+	if driverConf == nil {
+		return nil
+	}
+
+	// storage_root is the ocm driver's spelling of root; it stages uploads under
+	// the same <root>/uploads/ layout the decomposedfs family uses.
+	type driverRootConf struct {
+		Root            string `mapstructure:"root"`
+		UploadDirectory string `mapstructure:"upload_directory"`
+		StorageRoot     string `mapstructure:"storage_root"`
+	}
+	var rc driverRootConf
+	_ = mapstructure.Decode(driverConf, &rc)
+
+	// upload_directory already names the upload directory itself, the way
+	// decomposedfs reads it (options.go:172, posix tree.go:168). The other two are
+	// storage roots that stage uploads in a subdirectory, so only they get joined.
+	if rc.UploadDirectory != "" {
+		return newFileStoreWithTokens(rc.UploadDirectory, driverConf, log)
+	}
+
+	root := rc.Root
+	if root == "" {
+		root = rc.StorageRoot
+	}
+	if root == "" {
+		return nil
+	}
+
+	return newFileStoreWithTokens(filepath.Join(root, "uploads"), driverConf, log)
+}
+
+// NewFileStoreFromConfig builds a FileStore staging uploads in uploadDir when
+// set, falling back to the active driver config. This allows drivers that have
+// no local root (e.g. KW) to still get a coordinator by setting
+// upload_directory at the service level rather than inside the driver.
+// Returns nil only when neither source resolves to a non-empty path.
+func NewFileStoreFromConfig(uploadDir string, driverConf map[string]interface{}, log *zerolog.Logger) *FileStore {
+	if uploadDir != "" {
+		// Still take the tokens from the driver config: they sign the transfer URL
+		// that postprocessing downloads the staged bytes from, and a service-level
+		// upload directory says nothing about them.
+		return newFileStoreWithTokens(uploadDir, driverConf, log)
+	}
+	return FileStoreFromDriverConf(driverConf, log)
+}
+
+func newFileStoreWithTokens(uploadDir string, driverConf map[string]interface{}, log *zerolog.Logger) *FileStore {
+	type tokenConf struct {
+		DownloadEndpoint     string `mapstructure:"download_endpoint"`
+		DataGatewayEndpoint  string `mapstructure:"datagateway_endpoint"`
+		TransferSharedSecret string `mapstructure:"transfer_shared_secret"`
+		TransferExpires      int64  `mapstructure:"transfer_expires"`
+	}
+	var tc tokenConf
+	if tokens, ok := driverConf["tokens"]; ok {
+		_ = mapstructure.Decode(tokens, &tc)
+	}
+	return NewFileStore(uploadDir, TokenOptions{
+		DownloadEndpoint:     tc.DownloadEndpoint,
+		DataGatewayEndpoint:  tc.DataGatewayEndpoint,
+		TransferSharedSecret: tc.TransferSharedSecret,
+		TransferExpires:      tc.TransferExpires,
+	}, log)
+}
+
+// NewFileStore creates a FileStore staging uploads in uploadDir.
+// uploadDir must be on a shared filesystem when multiple pods handle the same space.
+func NewFileStore(uploadDir string, opts TokenOptions, log *zerolog.Logger) *FileStore {
+	return &FileStore{uploadDir: uploadDir, opts: opts, log: log}
+}
+
+// UploadDir returns the directory this FileStore stages uploads in.
+func (fs *FileStore) UploadDir() string {
+	return fs.uploadDir
+}
+
+// Setup creates the upload directory eagerly so permission problems are caught
+// at startup rather than on the first upload.
+func (fs *FileStore) Setup() error {
+	return os.MkdirAll(fs.uploadDir, 0700)
+}
+
+// New allocates a fresh session with a new UUID.
+func (fs *FileStore) New(_ context.Context) Session {
+	return &FileSession{
+		store: fs,
+		info: tusd.FileInfo{
+			ID: uuid.New().String(),
+			Storage: map[string]string{
+				"Type": "OCISStore",
+			},
+			MetaData: tusd.MetaData{},
+		},
+	}
+}
+
+// Get loads the session with the given id from disk.
+func (fs *FileStore) Get(ctx context.Context, id string) (Session, error) {
+	infoPath := fileSessionPath(fs.uploadDir, id)
+
+	data, err := os.ReadFile(infoPath)
+	if err != nil {
+		if pathErr, ok := err.(*os.PathError); ok && pathErr.Err == syscall.ESTALE {
+			return nil, tusd.ErrNotFound
+		}
+		if errors.Is(err, iofs.ErrNotExist) {
+			return nil, tusd.ErrNotFound
+		}
+		return nil, err
+	}
+
+	var info tusd.FileInfo
+	if err := json.Unmarshal(data, &info); err != nil {
+		return nil, err
+	}
+
+	session := &FileSession{store: fs, info: info}
+
+	stat, err := os.Stat(session.binPath())
+	if err != nil {
+		if os.IsNotExist(err) {
+			return nil, tusd.ErrNotFound
+		}
+		return nil, err
+	}
+	session.info.Offset = stat.Size()
+
+	return session, nil
+}
+
+// List returns all sessions found in the upload directory.
+func (fs *FileStore) List(ctx context.Context) ([]Session, error) {
+	infoFiles, err := filepath.Glob(filepath.Join(fs.uploadDir, "*.info"))
+	if err != nil {
+		return nil, err
+	}
+
+	sessions := make([]Session, 0, len(infoFiles))
+	for _, path := range infoFiles {
+		id := strings.TrimSuffix(filepath.Base(path), ".info")
+		session, err := fs.Get(ctx, id)
+		if err != nil {
+			appctx.GetLogger(ctx).Error().Str("path", path).Err(err).Msg("filestore: could not load session")
+			continue
+		}
+		sessions = append(sessions, session)
+	}
+	return sessions, nil
+}
diff --git a/pkg/upload/filestore_test.go b/pkg/upload/filestore_test.go
new file mode 100644
index 00000000000..43e23a5c2d7
--- /dev/null
+++ b/pkg/upload/filestore_test.go
@@ -0,0 +1,275 @@
+package upload
+
+import (
+	"context"
+	"os"
+	"path/filepath"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+	"github.com/rs/zerolog"
+	tusd "github.com/tus/tusd/v2/pkg/handler"
+)
+
+func nopLog() *zerolog.Logger {
+	l := zerolog.Nop()
+	return &l
+}
+
+var _ = Describe("FileStore", func() {
+	var (
+		ctx       context.Context
+		uploadDir string
+		fs        *FileStore
+	)
+
+	BeforeEach(func() {
+		ctx = context.Background()
+		uploadDir = filepath.Join(GinkgoT().TempDir(), "uploads")
+		fs = NewFileStore(uploadDir, TokenOptions{}, nopLog())
+	})
+
+	Describe("Setup", func() {
+		It("creates the upload directory", func() {
+			Expect(fs.Setup()).To(Succeed())
+
+			info, err := os.Stat(uploadDir)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(info.IsDir()).To(BeTrue())
+		})
+
+		It("is idempotent", func() {
+			Expect(fs.Setup()).To(Succeed())
+			Expect(fs.Setup()).To(Succeed())
+		})
+	})
+
+	Describe("New", func() {
+		BeforeEach(func() {
+			Expect(fs.Setup()).To(Succeed())
+		})
+
+		It("allocates a non-empty id", func() {
+			Expect(fs.New(ctx).ID()).ToNot(BeEmpty())
+		})
+
+		It("allocates unique ids", func() {
+			Expect(fs.New(ctx).ID()).ToNot(Equal(fs.New(ctx).ID()))
+		})
+
+		It("stamps the storage type", func() {
+			info, err := fs.New(ctx).GetInfo(ctx)
+			Expect(err).ToNot(HaveOccurred())
+			// Deliberately OcisStore's value: sessions written by either store must stay
+			// readable across a rolling deploy.
+			Expect(info.Storage["Type"]).To(Equal("OCISStore"))
+		})
+
+		It("initialises the metadata map", func() {
+			info, err := fs.New(ctx).GetInfo(ctx)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(info.MetaData).ToNot(BeNil())
+		})
+	})
+
+	Describe("Get", func() {
+		BeforeEach(func() {
+			Expect(fs.Setup()).To(Succeed())
+		})
+
+		It("loads a persisted session", func() {
+			s := fs.New(ctx).(*FileSession)
+			Expect(s.TouchBin()).To(Succeed())
+			Expect(s.Persist(ctx)).To(Succeed())
+
+			got, err := fs.Get(ctx, s.ID())
+			Expect(err).ToNot(HaveOccurred())
+			Expect(got.ID()).To(Equal(s.ID()))
+
+			info, err := got.GetInfo(ctx)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(info.Storage["Type"]).To(Equal("OCISStore"))
+		})
+
+		It("takes the offset from the staged binary size", func() {
+			s := fs.New(ctx).(*FileSession)
+			Expect(s.TouchBin()).To(Succeed())
+			Expect(s.Persist(ctx)).To(Succeed())
+
+			payload := []byte("hello world")
+			Expect(os.WriteFile(s.binPath(), payload, 0600)).To(Succeed())
+
+			got, err := fs.Get(ctx, s.ID())
+			Expect(err).ToNot(HaveOccurred())
+			Expect(got.Offset()).To(Equal(int64(len(payload))))
+		})
+
+		It("reports a missing info file as not found", func() {
+			_, err := fs.Get(ctx, "no-such-id")
+			Expect(err).To(MatchError(tusd.ErrNotFound))
+		})
+
+		It("reports corrupt info as an error, not as not found", func() {
+			s := fs.New(ctx).(*FileSession)
+			Expect(s.TouchBin()).To(Succeed())
+			Expect(s.Persist(ctx)).To(Succeed())
+
+			// Overwrite the .info with garbage JSON.
+			Expect(os.WriteFile(s.infoPath(), []byte("{not valid json"), 0600)).To(Succeed())
+
+			_, err := fs.Get(ctx, s.ID())
+			Expect(err).To(HaveOccurred())
+			Expect(err).ToNot(MatchError(tusd.ErrNotFound))
+		})
+
+		It("reports a missing staged binary as not found", func() {
+			s := fs.New(ctx).(*FileSession)
+			// Write .info but do NOT create the .bin file.
+			Expect(s.Persist(ctx)).To(Succeed())
+
+			_, err := fs.Get(ctx, s.ID())
+			Expect(err).To(MatchError(tusd.ErrNotFound))
+		})
+	})
+
+	Describe("List", func() {
+		BeforeEach(func() {
+			Expect(fs.Setup()).To(Succeed())
+		})
+
+		It("is empty on a fresh store", func() {
+			sessions, err := fs.List(ctx)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(sessions).To(BeEmpty())
+		})
+
+		It("returns every persisted session", func() {
+			s1 := fs.New(ctx).(*FileSession)
+			Expect(s1.TouchBin()).To(Succeed())
+			Expect(s1.Persist(ctx)).To(Succeed())
+
+			s2 := fs.New(ctx).(*FileSession)
+			Expect(s2.TouchBin()).To(Succeed())
+			Expect(s2.Persist(ctx)).To(Succeed())
+
+			sessions, err := fs.List(ctx)
+			Expect(err).ToNot(HaveOccurred())
+
+			ids := make([]string, 0, len(sessions))
+			for _, s := range sessions {
+				ids = append(ids, s.ID())
+			}
+			Expect(ids).To(ConsistOf(s1.ID(), s2.ID()))
+		})
+
+		It("skips a session whose staged binary is gone", func() {
+			good := fs.New(ctx).(*FileSession)
+			Expect(good.TouchBin()).To(Succeed())
+			Expect(good.Persist(ctx)).To(Succeed())
+
+			bad := fs.New(ctx).(*FileSession)
+			Expect(bad.TouchBin()).To(Succeed())
+			Expect(bad.Persist(ctx)).To(Succeed())
+			Expect(os.Remove(bad.binPath())).To(Succeed())
+
+			sessions, err := fs.List(ctx)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(sessions).To(HaveLen(1))
+			Expect(sessions[0].ID()).To(Equal(good.ID()))
+		})
+	})
+})
+
+var _ = Describe("FileStoreFromDriverConf", func() {
+	It("returns nil for a nil config", func() {
+		Expect(FileStoreFromDriverConf(nil, nopLog())).To(BeNil())
+	})
+
+	// root is a storage root, so uploads are staged in a subdirectory of it. This
+	// is the layout OcisStore uses (decomposedfs.go:258 joins o.Root itself).
+	It("stages uploads below the root key", func() {
+		root := GinkgoT().TempDir()
+		fs := FileStoreFromDriverConf(map[string]interface{}{"root": root}, nopLog())
+
+		Expect(fs).ToNot(BeNil())
+		Expect(fs.uploadDir).To(Equal(filepath.Join(root, "uploads")))
+	})
+
+	// storage_root is the ocm driver's spelling of root, same layout
+	// (ocm/storage/received/upload.go:212).
+	It("stages uploads below the storage_root key", func() {
+		root := GinkgoT().TempDir()
+		fs := FileStoreFromDriverConf(map[string]interface{}{"storage_root": root}, nopLog())
+
+		Expect(fs).ToNot(BeNil())
+		Expect(fs.uploadDir).To(Equal(filepath.Join(root, "uploads")))
+	})
+
+	// upload_directory already names the upload directory, the way decomposedfs
+	// reads it (options.go:172, posix tree.go:168), so it must not be joined again.
+	It("takes upload_directory as the upload directory itself", func() {
+		dir := GinkgoT().TempDir()
+		fs := FileStoreFromDriverConf(map[string]interface{}{"upload_directory": dir}, nopLog())
+
+		Expect(fs).ToNot(BeNil())
+		Expect(fs.uploadDir).To(Equal(dir))
+	})
+
+	It("prefers upload_directory over root", func() {
+		root := GinkgoT().TempDir()
+		uploadDir := GinkgoT().TempDir()
+		fs := FileStoreFromDriverConf(map[string]interface{}{
+			"root":             root,
+			"upload_directory": uploadDir,
+		}, nopLog())
+
+		Expect(fs).ToNot(BeNil())
+		Expect(fs.uploadDir).To(Equal(uploadDir))
+	})
+
+	It("returns nil when no root key is present", func() {
+		Expect(FileStoreFromDriverConf(map[string]interface{}{"some_other_key": "value"}, nopLog())).To(BeNil())
+	})
+})
+
+var _ = Describe("NewFileStoreFromConfig", func() {
+	// The service-level value already names the upload directory, so it is used
+	// verbatim rather than joined.
+	It("uses the service-level upload dir when set", func() {
+		uploadDir := GinkgoT().TempDir()
+		fs := NewFileStoreFromConfig(uploadDir, map[string]interface{}{"root": "/ignored"}, nopLog())
+
+		Expect(fs).ToNot(BeNil())
+		Expect(fs.uploadDir).To(Equal(uploadDir))
+	})
+
+	It("falls back to the driver config", func() {
+		root := GinkgoT().TempDir()
+		fs := NewFileStoreFromConfig("", map[string]interface{}{"root": root}, nopLog())
+
+		Expect(fs).ToNot(BeNil())
+		Expect(fs.uploadDir).To(Equal(filepath.Join(root, "uploads")))
+	})
+
+	It("returns nil when neither source resolves", func() {
+		Expect(NewFileStoreFromConfig("", nil, nopLog())).To(BeNil())
+	})
+
+	// A service-level upload directory must not drop the driver's tokens: they sign
+	// the transfer URL postprocessing downloads the staged bytes from.
+	It("keeps the driver tokens when the upload dir comes from the service", func() {
+		uploadDir := GinkgoT().TempDir()
+		fs := NewFileStoreFromConfig(uploadDir, map[string]interface{}{
+			"root": "/ignored",
+			"tokens": map[string]interface{}{
+				"transfer_shared_secret": "s3cret",
+				"download_endpoint":      "https://dl.example.com/data/",
+			},
+		}, nopLog())
+
+		Expect(fs).ToNot(BeNil())
+		Expect(fs.uploadDir).To(Equal(uploadDir))
+		Expect(fs.opts.TransferSharedSecret).To(Equal("s3cret"))
+		Expect(fs.opts.DownloadEndpoint).To(Equal("https://dl.example.com/data/"))
+	})
+})
diff --git a/pkg/upload/session.go b/pkg/upload/session.go
new file mode 100644
index 00000000000..565b2d72bfa
--- /dev/null
+++ b/pkg/upload/session.go
@@ -0,0 +1,491 @@
+package upload
+
+import (
+	"context"
+	"crypto/md5"  //nolint:gosec
+	"crypto/sha1" //nolint:gosec
+	"encoding/hex"
+	"encoding/json"
+	"fmt"
+	"hash"
+	"hash/adler32"
+	"io"
+	"os"
+	"path/filepath"
+	"strconv"
+	"strings"
+	"time"
+
+	userpb "github.com/cs3org/go-cs3apis/cs3/identity/user/v1beta1"
+	provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
+	typespb "github.com/cs3org/go-cs3apis/cs3/types/v1beta1"
+	"github.com/golang-jwt/jwt/v5"
+	"github.com/google/renameio/v2"
+	"github.com/pkg/errors"
+	tusd "github.com/tus/tusd/v2/pkg/handler"
+
+	"github.com/owncloud/reva/v2/pkg/appctx"
+	ctxpkg "github.com/owncloud/reva/v2/pkg/ctx"
+	"github.com/owncloud/reva/v2/pkg/errtypes"
+	"github.com/owncloud/reva/v2/pkg/storage"
+	"github.com/owncloud/reva/v2/pkg/utils"
+)
+
+const defaultFilePerm = os.FileMode(0664)
+
+// FileSession is the Session implementation for disk-backed uploads. While an
+// upload is in progress, incoming bytes are staged in a .bin file and upload
+// metadata (size, owner, checksums, etc.) is persisted in a .info file. Both
+// survive process restarts, allowing TUS resumption.
+//
+// In scope: read/write the staged .bin file, persist/load upload metadata in the .info file.
+// Out of scope: TUS protocol, checksums, event publishing, postprocessing — those live in coordinatedUpload and coordinator.
+type FileSession struct {
+	store *FileStore
+	info  tusd.FileInfo
+}
+
+// Session is the driver-agnostic view of an upload session the Coordinator
+// needs. Implementations must be pure state (CRUD): protocol orchestration
+// belongs to coordinatedUpload or the coordinator itself.
+type Session interface {
+	storage.UploadSession
+
+	// Data access — delegated to by coordinatedUpload for TUS reads/writes.
+	GetInfo(ctx context.Context) (tusd.FileInfo, error)
+	GetReader(ctx context.Context) (io.ReadCloser, error)
+	WriteChunk(ctx context.Context, offset int64, src io.Reader) (int64, error)
+
+	// Internal coordinator plumbing.
+	Chunk() string
+	BinPath() string
+	ExecutantUser() *userpb.User
+	ProviderID() string
+	SpaceID() string
+	NodeID() string
+	NodeParentID() string
+	NodeExists() bool
+	Dir() string
+	URL(ctx context.Context) (string, error)
+	SetScanData(result string, date time.Time)
+	Checksums() storage.UploadChecksums
+	SetChecksums(sha1, md5, adler32 []byte)
+	SizeDiff() int64
+	SetSizeDiff(d int64)
+	VersionCreated() bool
+	SetVersionCreated(v bool)
+	Metadata() map[string]string
+	Persist(ctx context.Context) error
+	Cleanup(ctx context.Context, cleanBin, cleanInfo bool)
+	Context(ctx context.Context) context.Context
+
+	// Typed setters used by Coordinator.InitiateUpload to populate a new session
+	// without knowing internal storage key names.
+	SetStorageValue(key, value string)
+	SetMetadata(key, value string)
+	SetSize(size int64)
+	SetSizeIsDeferred(value bool)
+	SetExecutant(u *userpb.User)
+	TouchBin() error
+}
+
+func (s *FileSession) GetInfo(_ context.Context) (tusd.FileInfo, error) {
+	return s.info, nil
+}
+
+func (s *FileSession) GetReader(_ context.Context) (io.ReadCloser, error) {
+	return os.Open(s.binPath())
+}
+
+func (s *FileSession) WriteChunk(ctx context.Context, offset int64, src io.Reader) (int64, error) {
+	file, err := os.OpenFile(s.binPath(), os.O_WRONLY|os.O_APPEND, defaultFilePerm)
+	if err != nil {
+		return 0, err
+	}
+	defer file.Close()
+
+	n, err := io.Copy(file, src)
+	if err != nil && err != io.ErrUnexpectedEOF {
+		return n, err
+	}
+	s.info.Offset += n
+	return n, nil
+}
+
+// Purge removes all on-disk state for this session.
+func (s *FileSession) Purge(ctx context.Context) {
+	s.Cleanup(ctx, true, true)
+}
+
+// ScanData returns the AV scan result and scan date stored on the session.
+func (s *FileSession) ScanData() (string, time.Time) {
+	date := s.info.MetaData["scanDate"]
+	if date == "" {
+		return "", time.Time{}
+	}
+	d, _ := time.Parse(time.RFC3339, date)
+	return s.info.MetaData["scanResult"], d
+}
+
+// ID returns the upload session ID.
+func (s *FileSession) ID() string {
+	return s.info.ID
+}
+
+// Filename returns the filename stored in the session.
+func (s *FileSession) Filename() string {
+	return s.info.Storage["NodeName"]
+}
+
+// Size returns the declared upload size.
+func (s *FileSession) Size() int64 {
+	return s.info.Size
+}
+
+// Offset returns the current upload offset.
+func (s *FileSession) Offset() int64 {
+	return s.info.Offset
+}
+
+// Chunk returns the chunk basename stored in the session, or "" for non-chunked uploads.
+func (s *FileSession) Chunk() string {
+	return s.info.Storage["Chunk"]
+}
+
+// BinPath returns the path to the staged binary file.
+func (s *FileSession) BinPath() string {
+	return s.binPath()
+}
+
+// SpaceGid returns the numeric GID of the space owner, or "" if not set.
+func (s *FileSession) SpaceGid() string {
+	return s.info.Storage["SpaceGid"]
+}
+
+// ProviderID returns the storage provider ID stored in the session.
+func (s *FileSession) ProviderID() string {
+	return s.info.MetaData["providerID"]
+}
+
+// SpaceID returns the space (root) ID.
+func (s *FileSession) SpaceID() string {
+	return s.info.Storage["SpaceRoot"]
+}
+
+// NodeID returns the node ID for this upload.
+func (s *FileSession) NodeID() string {
+	return s.info.Storage["NodeId"]
+}
+
+// NodeParentID returns the parent node ID for this upload.
+func (s *FileSession) NodeParentID() string {
+	return s.info.Storage["NodeParentId"]
+}
+
+// NodeExists returns whether the target node existed when the upload was initiated.
+func (s *FileSession) NodeExists() bool {
+	return s.info.Storage["NodeExists"] == "true"
+}
+
+// Dir returns the directory portion of the upload path.
+func (s *FileSession) Dir() string {
+	return s.info.Storage["Dir"]
+}
+
+// IsProcessing returns true if all bytes are received but postprocessing has not finished.
+func (s *FileSession) IsProcessing() bool {
+	return s.info.Size == s.info.Offset && s.info.MetaData["scanResult"] == ""
+}
+
+// SpaceOwner returns the space owner user ID.
+func (s *FileSession) SpaceOwner() *userpb.UserId {
+	return &userpb.UserId{
+		OpaqueId: s.info.Storage["SpaceOwnerOrManager"],
+		Idp:      s.info.Storage["SpaceOwnerIdp"],
+		Type:     userpb.UserType(userpb.UserType_value[s.info.Storage["SpaceOwnerType"]]),
+	}
+}
+
+// Executant returns the user ID of the user who initiated this upload.
+func (s *FileSession) Executant() userpb.UserId {
+	return userpb.UserId{
+		Type:     utils.UserTypeMap(s.info.Storage["UserType"]),
+		Idp:      s.info.Storage["Idp"],
+		OpaqueId: s.info.Storage["UserId"],
+	}
+}
+
+// Expires returns the upload expiry time.
+func (s *FileSession) Expires() time.Time {
+	var t time.Time
+	if value, ok := s.info.MetaData["expires"]; ok {
+		t, _ = utils.MTimeToTime(value)
+	}
+	return t
+}
+
+// Reference returns a CS3 reference for the resource being uploaded.
+func (s *FileSession) Reference() provider.Reference {
+	return provider.Reference{
+		ResourceId: &provider.ResourceId{
+			StorageId: s.info.MetaData["providerID"],
+			SpaceId:   s.info.Storage["SpaceRoot"],
+			OpaqueId:  s.info.Storage["NodeId"],
+		},
+	}
+}
+
+// Checksums returns the pre-computed checksums stored on the session.
+func (s *FileSession) Checksums() storage.UploadChecksums {
+	decode := func(key string) []byte {
+		b, _ := hex.DecodeString(s.info.MetaData[key])
+		return b
+	}
+	return storage.UploadChecksums{
+		SHA1:    decode("checksumSHA1"),
+		MD5:     decode("checksumMD5"),
+		Adler32: decode("checksumAdler32"),
+	}
+}
+
+// Metadata returns the upload metadata the coordinator passes to the driver.
+func (s *FileSession) Metadata() map[string]string {
+	return map[string]string{
+		"providerID":          s.info.MetaData["providerID"],
+		"mtime":               s.info.MetaData["mtime"],
+		"nodeExists":          s.info.Storage["NodeExists"],
+		"sessionID":           s.info.ID,
+		"if-match":            s.info.MetaData["if-match"],
+		"if-none-match":       s.info.MetaData["if-none-match"],
+		"if-unmodified-since": s.info.MetaData["if-unmodified-since"],
+	}
+}
+
+// SizeDiff returns the tree size change PrepareUpload propagated optimistically.
+// Rolling an upload back has to undo exactly that amount.
+func (s *FileSession) SizeDiff() int64 {
+	d, _ := strconv.ParseInt(s.info.MetaData["sizeDiff"], 10, 64)
+	return d
+}
+
+// SetSizeDiff records the size change PrepareUpload reported. It is persisted
+// because the async path prepares and commits in different processes, so the
+// value cannot be held in memory between the two.
+func (s *FileSession) SetSizeDiff(d int64) {
+	s.info.MetaData["sizeDiff"] = strconv.FormatInt(d, 10)
+}
+
+// VersionCreated reports whether this upload superseded existing content, which
+// UploadReady consumers use to tell an overwrite from a new file.
+func (s *FileSession) VersionCreated() bool {
+	return s.info.MetaData["versionCreated"] == "true"
+}
+
+// SetVersionCreated records what PrepareUpload reported. Persisted for the same
+// reason as the size diff: on the async path the commit runs in another process.
+func (s *FileSession) SetVersionCreated(v bool) {
+	s.info.MetaData["versionCreated"] = strconv.FormatBool(v)
+}
+
+// SetScanData stores AV scan results on the session.
+func (s *FileSession) SetScanData(result string, date time.Time) {
+	s.info.MetaData["scanResult"] = result
+	s.info.MetaData["scanDate"] = date.Format(time.RFC3339)
+}
+
+// SetChecksums stores pre-computed checksums so CommitUpload can use them without
+// re-reading the binary file.
+func (s *FileSession) SetChecksums(sha1Sum, md5Sum, adler32Sum []byte) {
+	s.info.MetaData["checksumSHA1"] = hex.EncodeToString(sha1Sum)
+	s.info.MetaData["checksumMD5"] = hex.EncodeToString(md5Sum)
+	s.info.MetaData["checksumAdler32"] = hex.EncodeToString(adler32Sum)
+}
+
+// SetMetadata sets a user-visible upload metadata field.
+func (s *FileSession) SetMetadata(key, value string) {
+	s.info.MetaData[key] = value
+}
+
+// SetStorageValue sets an internal storage field on the session.
+func (s *FileSession) SetStorageValue(key, value string) {
+	s.info.Storage[key] = value
+}
+
+// SetSize updates the declared upload size.
+func (s *FileSession) SetSize(size int64) {
+	s.info.Size = size
+}
+
+// SetSizeIsDeferred marks the upload size as not yet known.
+func (s *FileSession) SetSizeIsDeferred(value bool) {
+	s.info.SizeIsDeferred = value
+}
+
+// SetExecutant stores the identity of the user who initiated the upload.
+func (s *FileSession) SetExecutant(u *userpb.User) {
+	s.info.Storage["Idp"] = u.GetId().GetIdp()
+	s.info.Storage["UserId"] = u.GetId().GetOpaqueId()
+	s.info.Storage["UserType"] = utils.UserTypeToString(u.GetId().Type)
+	s.info.Storage["UserName"] = u.GetUsername()
+	s.info.Storage["UserDisplayName"] = u.GetDisplayName()
+	b, _ := json.Marshal(u.GetOpaque())
+	s.info.Storage["UserOpaque"] = string(b)
+	g, _ := json.Marshal(u.GetGroups())
+	s.info.Storage["UserGroups"] = string(g)
+}
+
+// TouchBin creates the empty staging file.
+func (s *FileSession) TouchBin() error {
+	f, err := os.OpenFile(s.binPath(), os.O_CREATE|os.O_WRONLY, defaultFilePerm)
+	if err != nil {
+		return err
+	}
+	return f.Close()
+}
+
+// Persist writes the session metadata atomically to disk.
+func (s *FileSession) Persist(ctx context.Context) error {
+	infoPath := s.infoPath()
+	if err := os.MkdirAll(filepath.Dir(infoPath), 0700); err != nil {
+		return err
+	}
+	d, err := json.Marshal(s.info)
+	if err != nil {
+		return err
+	}
+	return renameio.WriteFile(infoPath, d, 0600)
+}
+
+// Cleanup removes the staged binary and/or info file.
+// Node deletion and processing flag changes are the coordinator's responsibility.
+func (s *FileSession) Cleanup(ctx context.Context, cleanBin, cleanInfo bool) {
+	if cleanBin {
+		if err := os.Remove(s.binPath()); err != nil && !errors.Is(err, os.ErrNotExist) {
+			appctx.GetLogger(ctx).Error().Str("path", s.binPath()).Err(err).Msg("filestore: removing staged binary failed")
+		}
+	}
+	if cleanInfo {
+		if err := os.Remove(s.infoPath()); err != nil && !errors.Is(err, os.ErrNotExist) {
+			appctx.GetLogger(ctx).Error().Str("path", s.infoPath()).Err(err).Msg("filestore: removing session info failed")
+		}
+	}
+}
+
+// Context reconstructs a context carrying the user, logger, lock ID, and
+// initiator ID that were recorded when the upload was initiated.
+func (s *FileSession) Context(ctx context.Context) context.Context {
+	sub := s.store.log.With().Int("pid", os.Getpid()).Logger()
+	ctx = appctx.WithLogger(ctx, &sub)
+	ctx = ctxpkg.ContextSetLockID(ctx, s.info.MetaData["lockid"])
+	ctx = ctxpkg.ContextSetUser(ctx, s.ExecutantUser())
+	return ctxpkg.ContextSetInitiator(ctx, s.info.MetaData["initiatorid"])
+}
+
+// URL returns a signed JWT URL that the postprocessing service can use to
+// download the staged binary via the data gateway.
+func (s *FileSession) URL(_ context.Context) (string, error) {
+	type transferClaims struct {
+		jwt.RegisteredClaims
+		Target string `json:"target"`
+	}
+
+	target := joinURLParts(s.store.opts.DownloadEndpoint, "tus/", s.info.ID)
+	ttl := time.Duration(s.store.opts.TransferExpires) * time.Second
+	claims := transferClaims{
+		RegisteredClaims: jwt.RegisteredClaims{
+			ExpiresAt: jwt.NewNumericDate(time.Now().Add(ttl)),
+			Audience:  jwt.ClaimStrings{"reva"},
+			IssuedAt:  jwt.NewNumericDate(time.Now()),
+		},
+		Target: target,
+	}
+	t := jwt.NewWithClaims(jwt.GetSigningMethod("HS256"), claims)
+	tkn, err := t.SignedString([]byte(s.store.opts.TransferSharedSecret))
+	if err != nil {
+		return "", errors.Wrapf(err, "filestore: error signing transfer token with claims %+v", claims)
+	}
+	return joinURLParts(s.store.opts.DataGatewayEndpoint, tkn), nil
+}
+
+func (s *FileSession) ToFileInfo() tusd.FileInfo {
+	return s.info
+}
+
+func (s *FileSession) InitiatorID() string {
+	return s.info.MetaData["initiatorid"]
+}
+
+func (s *FileSession) binPath() string {
+	return filepath.Join(s.store.uploadDir, s.info.ID)
+}
+
+func (s *FileSession) infoPath() string {
+	return fileSessionPath(s.store.uploadDir, s.info.ID)
+}
+
+// ExecutantUser returns the full identity of the user who initiated the upload.
+// Upload events must carry it rather than the bare id from Executant(): consumers
+// read the display name straight off the event and do not look it up, so an
+// id-only user reaches the activity feed with a blank name.
+func (s *FileSession) ExecutantUser() *userpb.User {
+	var o *typespb.Opaque
+	_ = json.Unmarshal([]byte(s.info.Storage["UserOpaque"]), &o)
+	var groups []string
+	_ = json.Unmarshal([]byte(s.info.Storage["UserGroups"]), &groups)
+	return &userpb.User{
+		Id: &userpb.UserId{
+			Type:     utils.UserTypeMap(s.info.Storage["UserType"]),
+			Idp:      s.info.Storage["Idp"],
+			OpaqueId: s.info.Storage["UserId"],
+		},
+		Username:    s.info.Storage["UserName"],
+		DisplayName: s.info.Storage["UserDisplayName"],
+		Opaque:      o,
+		Groups:      groups,
+	}
+}
+
+// fileSessionPath returns the path to the .info file for the given session ID.
+func fileSessionPath(uploadDir, id string) string {
+	return filepath.Join(uploadDir, id+".info")
+}
+
+// calculateChecksums computes sha1, md5, and adler32 in a single pass over path.
+func calculateChecksums(_ context.Context, path string) (hash.Hash, hash.Hash, hash.Hash32, error) {
+	sha1h := sha1.New() //nolint:gosec
+	md5h := md5.New()   //nolint:gosec
+	adler32h := adler32.New()
+
+	f, err := os.Open(path)
+	if err != nil {
+		return nil, nil, nil, err
+	}
+	defer f.Close()
+
+	r1 := io.TeeReader(f, sha1h)
+	r2 := io.TeeReader(r1, md5h)
+	if _, err = io.Copy(adler32h, r2); err != nil {
+		return nil, nil, nil, err
+	}
+	return sha1h, md5h, adler32h, nil
+}
+
+func checkHash(expected string, h hash.Hash) error {
+	got := hex.EncodeToString(h.Sum(nil))
+	if expected != got {
+		return errtypes.ChecksumMismatch(fmt.Sprintf("invalid checksum: expected %s got %s", expected, got))
+	}
+	return nil
+}
+
+// joinURLParts concatenates URL path segments, inserting "/" between them if needed.
+func joinURLParts(parts ...string) string {
+	var b strings.Builder
+	for i, p := range parts {
+		b.WriteString(p)
+		if i < len(parts)-1 && !strings.HasSuffix(p, "/") {
+			b.WriteByte('/')
+		}
+	}
+	return b.String()
+}
diff --git a/pkg/upload/session_test.go b/pkg/upload/session_test.go
new file mode 100644
index 00000000000..6d04702d8d3
--- /dev/null
+++ b/pkg/upload/session_test.go
@@ -0,0 +1,417 @@
+package upload
+
+import (
+	"context"
+	"crypto/sha1" //nolint:gosec
+	"encoding/hex"
+	"errors"
+	"os"
+	"path/filepath"
+	"strings"
+	"time"
+
+	userpb "github.com/cs3org/go-cs3apis/cs3/identity/user/v1beta1"
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+	"github.com/rs/zerolog"
+
+	ctxpkg "github.com/owncloud/reva/v2/pkg/ctx"
+	"github.com/owncloud/reva/v2/pkg/errtypes"
+	"github.com/owncloud/reva/v2/pkg/utils"
+)
+
+var _ = Describe("FileSession", func() {
+	var (
+		ctx  context.Context
+		fs   *FileStore
+		sess *FileSession
+	)
+
+	// newTestStore returns a store rooted in a fresh temp dir. Setup() is not
+	// called: Persist creates the uploads dir on demand, and the specs that need
+	// the store to be readable call it themselves.
+	newTestStore := func(opts TokenOptions) *FileStore {
+		log := zerolog.Nop()
+		return NewFileStore(GinkgoT().TempDir(), opts, &log)
+	}
+
+	BeforeEach(func() {
+		ctx = context.Background()
+		fs = newTestStore(TokenOptions{})
+		sess = fs.New(ctx).(*FileSession)
+	})
+
+	Describe("typed setters", func() {
+		It("stores metadata readable through the typed getter", func() {
+			sess.SetMetadata("providerID", "storage-1")
+			Expect(sess.ProviderID()).To(Equal("storage-1"))
+		})
+
+		It("stores storage values readable through the typed getter", func() {
+			sess.SetStorageValue("SpaceRoot", "space-abc")
+			Expect(sess.SpaceID()).To(Equal("space-abc"))
+		})
+
+		It("stores the declared size", func() {
+			sess.SetSize(1234)
+			Expect(sess.Size()).To(Equal(int64(1234)))
+		})
+
+		It("toggles the deferred size flag", func() {
+			sess.SetSizeIsDeferred(true)
+			Expect(sess.info.SizeIsDeferred).To(BeTrue())
+			sess.SetSizeIsDeferred(false)
+			Expect(sess.info.SizeIsDeferred).To(BeFalse())
+		})
+	})
+
+	Describe("SetExecutant", func() {
+		It("round-trips the executant id through Executant", func() {
+			sess.SetExecutant(&userpb.User{
+				Id: &userpb.UserId{
+					Idp:      "idp.example.com",
+					OpaqueId: "user-42",
+					Type:     userpb.UserType_USER_TYPE_PRIMARY,
+				},
+				Username:    "alice",
+				DisplayName: "Alice",
+			})
+
+			got := sess.Executant()
+			Expect(got.Idp).To(Equal("idp.example.com"))
+			Expect(got.OpaqueId).To(Equal("user-42"))
+			Expect(got.Type).To(Equal(userpb.UserType_USER_TYPE_PRIMARY))
+		})
+	})
+
+	Describe("NodeExists", func() {
+		It("is false when the key is absent", func() {
+			Expect(sess.NodeExists()).To(BeFalse(), "absent key should return false")
+		})
+
+		It("is true only for the literal \"true\"", func() {
+			sess.SetStorageValue("NodeExists", "true")
+			Expect(sess.NodeExists()).To(BeTrue())
+
+			sess.SetStorageValue("NodeExists", "false")
+			Expect(sess.NodeExists()).To(BeFalse())
+
+			sess.SetStorageValue("NodeExists", "yes")
+			Expect(sess.NodeExists()).To(BeFalse(), "non-'true' value should return false")
+		})
+	})
+
+	Describe("Checksums", func() {
+		It("round-trips the raw bytes SetChecksums stored", func() {
+			sha1bytes := []byte{0x01, 0x02, 0x03, 0x04}
+			md5bytes := []byte{0x05, 0x06, 0x07, 0x08}
+			adlerBytes := []byte{0x09, 0x0a, 0x0b, 0x0c}
+
+			sess.SetChecksums(sha1bytes, md5bytes, adlerBytes)
+
+			got := sess.Checksums()
+			Expect(got.SHA1).To(Equal(sha1bytes))
+			Expect(got.MD5).To(Equal(md5bytes))
+			Expect(got.Adler32).To(Equal(adlerBytes))
+		})
+	})
+
+	Describe("ScanData", func() {
+		It("is empty on a fresh session", func() {
+			result, date := sess.ScanData()
+			Expect(result).To(BeEmpty(), "fresh session should have no scan result")
+			Expect(date.IsZero()).To(BeTrue(), "fresh session should have zero scan date")
+		})
+
+		It("round-trips what SetScanData stored", func() {
+			now := time.Now().Truncate(time.Second)
+			sess.SetScanData("clean", now)
+
+			result, date := sess.ScanData()
+			Expect(result).To(Equal("clean"))
+			Expect(date).To(BeTemporally("~", now, time.Second))
+		})
+	})
+
+	Describe("Expires", func() {
+		It("is the zero time when unset", func() {
+			Expect(sess.Expires().IsZero()).To(BeTrue(), "absent expires should be zero time")
+		})
+
+		It("parses the stored OCM mtime", func() {
+			want := time.Now().Add(time.Hour).Truncate(time.Second)
+			sess.SetMetadata("expires", utils.TimeToOCMtime(want))
+			Expect(sess.Expires()).To(BeTemporally("~", want, time.Second))
+		})
+	})
+
+	Describe("IsProcessing", func() {
+		It("is true only once all bytes arrived and no scan result is in", func() {
+			sess.SetSize(100)
+			Expect(sess.IsProcessing()).To(BeFalse(), "size != offset should not be processing")
+
+			sess.info.Offset = 100
+			Expect(sess.IsProcessing()).To(BeTrue(), "size == offset with no scan result should be processing")
+
+			sess.SetScanData("clean", time.Now())
+			Expect(sess.IsProcessing()).To(BeFalse(), "scan result set means processing finished")
+		})
+	})
+
+	Describe("Reference", func() {
+		It("assembles the resource id from the stored parts", func() {
+			sess.SetMetadata("providerID", "prov-1")
+			sess.SetStorageValue("SpaceRoot", "space-2")
+			sess.SetStorageValue("NodeId", "node-3")
+
+			ref := sess.Reference()
+			Expect(ref.ResourceId).ToNot(BeNil())
+			Expect(ref.ResourceId.StorageId).To(Equal("prov-1"))
+			Expect(ref.ResourceId.SpaceId).To(Equal("space-2"))
+			Expect(ref.ResourceId.OpaqueId).To(Equal("node-3"))
+		})
+	})
+
+	Describe("Metadata", func() {
+		It("carries everything the driver needs at commit time", func() {
+			sess.SetMetadata("providerID", "p1")
+			sess.SetMetadata("mtime", "12345.0")
+			sess.SetStorageValue("NodeExists", "true")
+			sess.SetMetadata("if-match", "etag-1")
+			sess.SetMetadata("if-none-match", "*")
+			sess.SetMetadata("if-unmodified-since", "2026-07-30T10:00:00Z")
+
+			m := sess.Metadata()
+			Expect(m["providerID"]).To(Equal("p1"))
+			Expect(m["mtime"]).To(Equal("12345.0"))
+			Expect(m["nodeExists"]).To(Equal("true"))
+			Expect(m["sessionID"]).To(Equal(sess.ID()))
+			// The driver re-checks these at PrepareUpload, so they must survive the session.
+			Expect(m["if-match"]).To(Equal("etag-1"))
+			Expect(m["if-none-match"]).To(Equal("*"))
+			Expect(m["if-unmodified-since"]).To(Equal("2026-07-30T10:00:00Z"))
+		})
+	})
+
+	Describe("Persist", func() {
+		It("survives a round trip through the store", func() {
+			Expect(fs.Setup()).To(Succeed())
+
+			sess.SetSize(512)
+			sess.SetMetadata("providerID", "prov-rt")
+			sess.SetStorageValue("NodeId", "nd-rt")
+			sess.SetStorageValue("NodeExists", "true")
+
+			Expect(sess.TouchBin()).To(Succeed())
+			Expect(sess.Persist(ctx)).To(Succeed())
+
+			loaded, err := fs.Get(ctx, sess.ID())
+			Expect(err).ToNot(HaveOccurred())
+			ls := loaded.(*FileSession)
+
+			Expect(ls.Size()).To(Equal(int64(512)))
+			Expect(ls.ProviderID()).To(Equal("prov-rt"))
+			Expect(ls.NodeID()).To(Equal("nd-rt"))
+			Expect(ls.NodeExists()).To(BeTrue())
+		})
+
+		It("creates intermediate directories", func() {
+			log := zerolog.Nop()
+			nested := NewFileStore(filepath.Join(GinkgoT().TempDir(), "sub", "nested"), TokenOptions{}, &log)
+			s := nested.New(ctx).(*FileSession)
+			s.SetSize(1)
+
+			Expect(s.Persist(ctx)).To(Succeed())
+			_, err := os.Stat(s.infoPath())
+			Expect(err).ToNot(HaveOccurred())
+		})
+	})
+
+	Describe("WriteChunk", func() {
+		It("appends and advances the offset", func() {
+			Expect(os.MkdirAll(filepath.Dir(sess.binPath()), 0700)).To(Succeed())
+			Expect(sess.TouchBin()).To(Succeed())
+
+			n, err := sess.WriteChunk(ctx, 0, strings.NewReader("hello"))
+			Expect(err).ToNot(HaveOccurred())
+			Expect(n).To(Equal(int64(5)))
+			Expect(sess.Offset()).To(Equal(int64(5)))
+
+			n2, err := sess.WriteChunk(ctx, 5, strings.NewReader(" world"))
+			Expect(err).ToNot(HaveOccurred())
+			Expect(n2).To(Equal(int64(6)))
+			Expect(sess.Offset()).To(Equal(int64(11)))
+
+			data, err := os.ReadFile(sess.binPath())
+			Expect(err).ToNot(HaveOccurred())
+			Expect(string(data)).To(Equal("hello world"))
+		})
+
+		It("fails when the staging file does not exist", func() {
+			Expect(os.MkdirAll(filepath.Dir(sess.binPath()), 0700)).To(Succeed())
+
+			_, err := sess.WriteChunk(ctx, 0, strings.NewReader("data"))
+			Expect(err).To(HaveOccurred())
+		})
+	})
+
+	Describe("Cleanup", func() {
+		// A persisted session with both files on disk.
+		var staged *FileSession
+
+		BeforeEach(func() {
+			staged = sess
+			Expect(os.MkdirAll(filepath.Dir(staged.binPath()), 0700)).To(Succeed())
+			Expect(staged.TouchBin()).To(Succeed())
+			Expect(staged.Persist(ctx)).To(Succeed())
+		})
+
+		It("removes the binary only", func() {
+			staged.Cleanup(ctx, true, false)
+			_, err := os.Stat(staged.binPath())
+			Expect(os.IsNotExist(err)).To(BeTrue(), "bin should be removed")
+			_, err = os.Stat(staged.infoPath())
+			Expect(err).ToNot(HaveOccurred(), "info should survive")
+		})
+
+		It("removes the info only", func() {
+			staged.Cleanup(ctx, false, true)
+			_, err := os.Stat(staged.binPath())
+			Expect(err).ToNot(HaveOccurred(), "bin should survive")
+			_, err = os.Stat(staged.infoPath())
+			Expect(os.IsNotExist(err)).To(BeTrue(), "info should be removed")
+		})
+
+		It("removes both", func() {
+			staged.Cleanup(ctx, true, true)
+			_, err := os.Stat(staged.binPath())
+			Expect(os.IsNotExist(err)).To(BeTrue(), "bin should be removed")
+			_, err = os.Stat(staged.infoPath())
+			Expect(os.IsNotExist(err)).To(BeTrue(), "info should be removed")
+		})
+
+		It("removes nothing when both flags are false", func() {
+			staged.Cleanup(ctx, false, false)
+			_, err := os.Stat(staged.binPath())
+			Expect(err).ToNot(HaveOccurred(), "bin should survive")
+			_, err = os.Stat(staged.infoPath())
+			Expect(err).ToNot(HaveOccurred(), "info should survive")
+		})
+
+		It("tolerates files that are already gone", func() {
+			fresh := fs.New(ctx).(*FileSession)
+			Expect(func() { fresh.Cleanup(ctx, true, true) }).ToNot(Panic())
+		})
+	})
+
+	Describe("Context", func() {
+		It("restores the user, lock id, and initiator id", func() {
+			sess.SetExecutant(&userpb.User{
+				Id: &userpb.UserId{
+					Idp:      "idp.test",
+					OpaqueId: "ctx-user",
+					Type:     userpb.UserType_USER_TYPE_PRIMARY,
+				},
+				Username: "ctxuser",
+			})
+			sess.SetMetadata("lockid", "lock-xyz")
+			sess.SetMetadata("initiatorid", "initiator-abc")
+
+			sessCtx := sess.Context(ctx)
+
+			gotUser, ok := ctxpkg.ContextGetUser(sessCtx)
+			Expect(ok).To(BeTrue())
+			Expect(gotUser.GetId().GetOpaqueId()).To(Equal("ctx-user"))
+			Expect(gotUser.GetId().GetIdp()).To(Equal("idp.test"))
+
+			lockID, ok := ctxpkg.ContextGetLockID(sessCtx)
+			Expect(ok).To(BeTrue())
+			Expect(lockID).To(Equal("lock-xyz"))
+
+			initiator, ok := ctxpkg.ContextGetInitiator(sessCtx)
+			Expect(ok).To(BeTrue())
+			Expect(initiator).To(Equal("initiator-abc"))
+		})
+	})
+
+	Describe("URL", func() {
+		It("signs a transfer token behind the data gateway endpoint", func() {
+			store := newTestStore(TokenOptions{
+				DownloadEndpoint:     "http://download.example.com",
+				DataGatewayEndpoint:  "http://gateway.example.com",
+				TransferSharedSecret: "s3cr3t",
+				TransferExpires:      3600,
+			})
+			s := store.New(ctx).(*FileSession)
+
+			url, err := s.URL(ctx)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(url).ToNot(BeEmpty())
+			Expect(url).To(HavePrefix("http://gateway.example.com"), "URL should start with DataGatewayEndpoint")
+		})
+
+		It("can be called repeatedly", func() {
+			store := newTestStore(TokenOptions{
+				DataGatewayEndpoint:  "http://gw.example.com",
+				TransferSharedSecret: "secret",
+				TransferExpires:      60,
+			})
+			s := store.New(ctx).(*FileSession)
+
+			url1, err := s.URL(ctx)
+			Expect(err).ToNot(HaveOccurred())
+			url2, err := s.URL(ctx)
+			Expect(err).ToNot(HaveOccurred())
+
+			Expect(url1).ToNot(BeEmpty())
+			Expect(url2).ToNot(BeEmpty())
+		})
+	})
+})
+
+var _ = Describe("calculateChecksums", func() {
+	It("computes sha1, md5, and adler32 in one pass", func() {
+		path := filepath.Join(GinkgoT().TempDir(), "testfile")
+		Expect(os.WriteFile(path, []byte("hello"), 0600)).To(Succeed())
+
+		sha1h, md5h, adler32h, err := calculateChecksums(context.Background(), path)
+		Expect(err).ToNot(HaveOccurred())
+
+		Expect(hex.EncodeToString(sha1h.Sum(nil))).To(Equal("aaf4c61ddcc5e8a2dabede0f3b482cd9aea9434d"))
+		Expect(hex.EncodeToString(md5h.Sum(nil))).To(Equal("5d41402abc4b2a76b9719d911017c592"))
+		Expect(hex.EncodeToString(adler32h.Sum(nil))).To(Equal("062c0215"))
+	})
+
+	It("errors on a missing file", func() {
+		_, _, _, err := calculateChecksums(context.Background(), "/nonexistent/path/file.bin")
+		Expect(err).To(HaveOccurred())
+	})
+})
+
+var _ = Describe("checkHash", func() {
+	It("accepts a matching digest", func() {
+		hash := sha1.New() //nolint:gosec
+		hash.Write([]byte("hello"))
+		Expect(checkHash("aaf4c61ddcc5e8a2dabede0f3b482cd9aea9434d", hash)).To(Succeed())
+	})
+
+	It("reports a mismatch as errtypes.ChecksumMismatch", func() {
+		hash := sha1.New() //nolint:gosec
+		hash.Write([]byte("hello"))
+
+		err := checkHash("0000000000000000000000000000000000000000", hash)
+		Expect(err).To(HaveOccurred())
+		Expect(errors.As(err, new(errtypes.ChecksumMismatch))).To(BeTrue())
+	})
+})
+
+var _ = DescribeTable("joinURLParts",
+	func(parts []string, want string) {
+		Expect(joinURLParts(parts...)).To(Equal(want))
+	},
+	Entry("trailing slash on the base", []string{"http://host/", "path"}, "http://host/path"),
+	Entry("no trailing slash on the base", []string{"http://host", "path"}, "http://host/path"),
+	Entry("single part", []string{"http://host"}, "http://host"),
+	Entry("three parts", []string{"http://host", "a", "b"}, "http://host/a/b"),
+	Entry("slashes already present", []string{"http://host/", "a/", "b"}, "http://host/a/b"),
+)
diff --git a/pkg/upload/upload_suite_test.go b/pkg/upload/upload_suite_test.go
new file mode 100644
index 00000000000..c436cb301ed
--- /dev/null
+++ b/pkg/upload/upload_suite_test.go
@@ -0,0 +1,13 @@
+package upload_test
+
+import (
+	"testing"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+func TestUpload(t *testing.T) {
+	RegisterFailHandler(Fail)
+	RunSpecs(t, "Upload Coordinator Suite")
+}