diff --git a/app/artifact-cas/cmd/main.go b/app/artifact-cas/cmd/main.go
index b3986301f..2a647a3d6 100644
--- a/app/artifact-cas/cmd/main.go
+++ b/app/artifact-cas/cmd/main.go
@@ -25,6 +25,7 @@ import (
 
 	"github.com/chainloop-dev/chainloop/app/artifact-cas/internal/conf"
 	"github.com/chainloop-dev/chainloop/app/artifact-cas/internal/server"
+	"github.com/chainloop-dev/chainloop/app/artifact-cas/internal/service"
 	backend "github.com/chainloop-dev/chainloop/pkg/blobmanager"
 	"github.com/chainloop-dev/chainloop/pkg/credentials"
 	"github.com/chainloop-dev/chainloop/pkg/credentials/manager"
@@ -121,6 +122,12 @@ func main() {
 
 	_ = logger.Log(log.LevelInfo, "msg", "starting artifact-cas service", "version", Version)
 
+	// Ensure the upload staging directory exists and sweep any files a previous
+	// crash left behind, so verification always starts from a clean volume.
+	if err := prepareStagingDir(&bc, logger); err != nil {
+		panic(err)
+	}
+
 	flush, err := initSentry(&bc, logger)
 	defer flush()
 	if err != nil {
@@ -152,6 +159,31 @@ func newProtoValidator() (protovalidate.Validator, error) {
 	return protovalidate.New()
 }
 
+// prepareStagingDir resolves the upload staging directory (falling back to the
+// OS temp dir when unconfigured — dev only; production mounts a dedicated
+// volume), creates it, and removes any leftover staging files from a previous
+// run. It must use the same directory the service is configured with (see
+// serviceOpts / conf.staging_dir).
+func prepareStagingDir(bc *conf.Bootstrap, logger log.Logger) error {
+	dir := bc.GetStagingDir()
+	if dir == "" {
+		dir = os.TempDir()
+		_ = logger.Log(log.LevelWarn, "msg", "staging_dir not configured, falling back to OS temp dir (dev only)", "dir", dir)
+	}
+
+	if err := os.MkdirAll(dir, 0o700); err != nil {
+		return err
+	}
+
+	if _, err := service.SweepStagingDir(dir, servicelogger.ScopedHelper(logger, "staging")); err != nil {
+		// A sweep failure is not fatal: the deferred per-upload cleanup still
+		// applies, so log and continue rather than blocking startup.
+		_ = logger.Log(log.LevelWarn, "msg", "failed to sweep staging dir", "dir", dir, "error", err.Error())
+	}
+
+	return nil
+}
+
 func initSentry(c *conf.Bootstrap, logger log.Logger) (cleanupFunc func(), err error) {
 	cleanupFunc = func() {
 		sentry.Flush(2 * time.Second)
diff --git a/app/artifact-cas/cmd/wire.go b/app/artifact-cas/cmd/wire.go
index da2a3735e..a3866888e 100644
--- a/app/artifact-cas/cmd/wire.go
+++ b/app/artifact-cas/cmd/wire.go
@@ -51,10 +51,11 @@ func wireApp(*conf.Bootstrap, *conf.Server, *conf.Auth, credentials.Reader, log.
 	)
 }
 
-func serviceOpts(l log.Logger, audit *service.AuditDispatcher) []service.NewOpt {
+func serviceOpts(l log.Logger, audit *service.AuditDispatcher, bc *conf.Bootstrap) []service.NewOpt {
 	return []service.NewOpt{
 		service.WithLogger(l),
 		service.WithAuditDispatcher(audit),
+		service.WithStagingDir(bc.GetStagingDir()),
 	}
 }
 
diff --git a/app/artifact-cas/cmd/wire_gen.go b/app/artifact-cas/cmd/wire_gen.go
index 6ae068607..4b48eef04 100644
--- a/app/artifact-cas/cmd/wire_gen.go
+++ b/app/artifact-cas/cmd/wire_gen.go
@@ -37,7 +37,7 @@ func wireApp(bootstrap *conf.Bootstrap, confServer *conf.Server, auth *conf.Auth
 		return nil, nil, err
 	}
 	auditDispatcher := service.NewAuditDispatcher(auditLogPublisher, logger)
-	v := serviceOpts(logger, auditDispatcher)
+	v := serviceOpts(logger, auditDispatcher, bootstrap)
 	byteStreamService := service.NewByteStreamService(providers, v...)
 	resourceService := service.NewResourceService(providers, v...)
 	validator, err := newProtoValidator()
@@ -75,8 +75,8 @@ func wireApp(bootstrap *conf.Bootstrap, confServer *conf.Server, auth *conf.Auth
 
 // wire.go:
 
-func serviceOpts(l log.Logger, audit *service.AuditDispatcher) []service.NewOpt {
-	return []service.NewOpt{service.WithLogger(l), service.WithAuditDispatcher(audit)}
+func serviceOpts(l log.Logger, audit *service.AuditDispatcher, bc *conf.Bootstrap) []service.NewOpt {
+	return []service.NewOpt{service.WithLogger(l), service.WithAuditDispatcher(audit), service.WithStagingDir(bc.GetStagingDir())}
 }
 
 // newNatsConfig converts the proto config to a plain natsconn.Config, nil when unset
diff --git a/app/artifact-cas/internal/conf/conf.pb.go b/app/artifact-cas/internal/conf/conf.pb.go
index e724c6098..c364830d7 100644
--- a/app/artifact-cas/internal/conf/conf.pb.go
+++ b/app/artifact-cas/internal/conf/conf.pb.go
@@ -46,7 +46,13 @@ type Bootstrap struct {
 	CredentialsService *v1.Credentials          `protobuf:"bytes,4,opt,name=credentials_service,json=credentialsService,proto3" json:"credentials_service,omitempty"`
 	// Optional NATS server configuration to publish audit events to the
 	// control-plane-owned stream. When unset, event publishing is disabled.
-	NatsServer    *Bootstrap_NatsServer `protobuf:"bytes,5,opt,name=nats_server,json=natsServer,proto3" json:"nats_server,omitempty"`
+	NatsServer *Bootstrap_NatsServer `protobuf:"bytes,5,opt,name=nats_server,json=natsServer,proto3" json:"nats_server,omitempty"`
+	// Local directory where uploads (and, later, downloads) are staged on disk
+	// and verified against the declared digest before reaching the backend. It
+	// must be a writable volume; in production a dedicated emptyDir is mounted
+	// here (NOT tmpfs/RAM, and NOT the /tmp secret mount). When unset the service
+	// falls back to the OS temp dir, which is only appropriate for local dev.
+	StagingDir    string `protobuf:"bytes,6,opt,name=staging_dir,json=stagingDir,proto3" json:"staging_dir,omitempty"`
 	unknownFields protoimpl.UnknownFields
 	sizeCache     protoimpl.SizeCache
 }
@@ -116,6 +122,13 @@ func (x *Bootstrap) GetNatsServer() *Bootstrap_NatsServer {
 	return nil
 }
 
+func (x *Bootstrap) GetStagingDir() string {
+	if x != nil {
+		return x.StagingDir
+	}
+	return ""
+}
+
 type Server struct {
 	state protoimpl.MessageState `protogen:"open.v1"`
 	// Regular HTTP endpoint
@@ -729,14 +742,16 @@ var File_conf_proto protoreflect.FileDescriptor
 const file_conf_proto_rawDesc = "" +
 	"\n" +
 	"\n" +
-	"conf.proto\x1a\x1bcredentials/v1/config.proto\x1a\x1egoogle/protobuf/duration.proto\"\xb9\x05\n" +
+	"conf.proto\x1a\x1bcredentials/v1/config.proto\x1a\x1egoogle/protobuf/duration.proto\"\xda\x05\n" +
 	"\tBootstrap\x12\x1f\n" +
 	"\x06server\x18\x01 \x01(\v2\a.ServerR\x06server\x12\x19\n" +
 	"\x04auth\x18\x02 \x01(\v2\x05.AuthR\x04auth\x12>\n" +
 	"\robservability\x18\x03 \x01(\v2\x18.Bootstrap.ObservabilityR\robservability\x12L\n" +
 	"\x13credentials_service\x18\x04 \x01(\v2\x1b.credentials.v1.CredentialsR\x12credentialsService\x126\n" +
 	"\vnats_server\x18\x05 \x01(\v2\x15.Bootstrap.NatsServerR\n" +
-	"natsServer\x1aH\n" +
+	"natsServer\x12\x1f\n" +
+	"\vstaging_dir\x18\x06 \x01(\tR\n" +
+	"stagingDir\x1aH\n" +
 	"\n" +
 	"NatsServer\x12\x10\n" +
 	"\x03uri\x18\x01 \x01(\tR\x03uri\x12\x16\n" +
diff --git a/app/artifact-cas/internal/conf/conf.proto b/app/artifact-cas/internal/conf/conf.proto
index 9905e1c03..5c14b4227 100644
--- a/app/artifact-cas/internal/conf/conf.proto
+++ b/app/artifact-cas/internal/conf/conf.proto
@@ -28,6 +28,12 @@ message Bootstrap {
   // Optional NATS server configuration to publish audit events to the
   // control-plane-owned stream. When unset, event publishing is disabled.
   NatsServer nats_server = 5;
+  // Local directory where uploads (and, later, downloads) are staged on disk
+  // and verified against the declared digest before reaching the backend. It
+  // must be a writable volume; in production a dedicated emptyDir is mounted
+  // here (NOT tmpfs/RAM, and NOT the /tmp secret mount). When unset the service
+  // falls back to the OS temp dir, which is only appropriate for local dev.
+  string staging_dir = 6;
 
   message NatsServer {
     // NATS server URI, e.g. "nats://localhost:4222"
diff --git a/app/artifact-cas/internal/service/bytestream.go b/app/artifact-cas/internal/service/bytestream.go
index 86c588141..8bc5c1d8d 100644
--- a/app/artifact-cas/internal/service/bytestream.go
+++ b/app/artifact-cas/internal/service/bytestream.go
@@ -25,6 +25,7 @@ import (
 	"fmt"
 	"hash"
 	"io"
+	"os"
 
 	"errors"
 
@@ -116,28 +117,30 @@ func (s *ByteStreamService) Write(stream bytestream.ByteStream_WriteServer) erro
 
 	s.log.Infow("msg", "artifact does not exist, uploading", "digest", req.resource.Digest, "name", req.resource.FileName)
 
-	// Streaming-capable backends (object stores such as S3/Azure) are fed
-	// directly from the client stream through an io.Pipe, so CAS memory stays
-	// bounded by the chunk/pipe size regardless of artifact size (PFM-6923).
-	// The OCI backend, whose push path needs the whole layer content up front,
-	// does not advertise streaming and keeps the fully-buffered path.
-	var committedSize int64
-	if su, ok := storageBackend.(backend.StreamingUploader); ok && su.SupportsStreaming() {
-		committedSize, err = s.streamUpload(ctx, stream, storageBackend, req, info.MaxBytes)
-	} else {
-		committedSize, err = s.bufferedUpload(ctx, stream, storageBackend, req, info.MaxBytes)
-	}
-
-	// Classify the outcome. The error may come from two distinct stages, which
-	// must be treated differently:
+	// Spill the upload to local disk, verify its SHA256 against the declared
+	// digest, and only then hand the verified file to the backend. The canonical
+	// key can therefore never hold content that does not hash to its digest, and
+	// CAS memory stays bounded because the artifact lives on disk.
+	committedSize, err := s.spillVerifyUpload(ctx, stream, storageBackend, req, info.MaxBytes)
+
+	// Classify the outcome. The error may come from several distinct stages,
+	// which must be treated differently:
+	//   - A digest mismatch (digestMismatchError) is the client's fault: the
+	//     bytes do not hash to the key they declared, so the request is invalid
+	//     and no bytes were ever sent to the backend.
 	//   - A backend Upload failure (backendUploadError) is always masked as an
 	//     internal error. It must NOT be interpreted as a client disconnect even
 	//     when it wraps a network reset/cancellation originating backend-side —
 	//     doing so would falsely report success and silently drop the artifact.
-	//   - A stream-read (feed) error is classified: a client disconnect is not a
+	//   - A stream-read (spill) error is classified: a client disconnect is not a
 	//     failure, an exceeded size cap maps to ResourceExhausted, anything else
-	//     is masked.
+	//     (e.g. a staging-disk write failure) is masked.
 	if err != nil {
+		var mismatch *digestMismatchError
+		if errors.As(err, &mismatch) {
+			s.log.Infow("msg", "upload rejected: digest mismatch", "digest", req.resource.Digest, "name", req.resource.FileName, "got", mismatch.got)
+			return status.Error(codes.InvalidArgument, err.Error())
+		}
 		var backendErr *backendUploadError
 		if errors.As(err, &backendErr) {
 			return sl.LogAndMaskErr(backendErr.err, s.log)
@@ -165,91 +168,69 @@ func (s *ByteStreamService) Write(stream bytestream.ByteStream_WriteServer) erro
 	return stream.SendAndClose(&bytestream.WriteResponse{CommittedSize: committedSize})
 }
 
-// bufferedUpload accumulates the whole artifact in memory before handing it to
-// the backend. This is required by the OCI backend: its push implementation
-// does not support streaming/chunked uploads for uncompressed layers (we can not
-// use stream.Layer since it only supports compressed layers, and we want to
-// store raw data with custom mimetypes), so it needs the full content up front.
-// https://github.com/google/go-containerregistry/blob/main/pkg/v1/stream/README.md
-// It returns the total number of bytes committed to the backend. Feed errors are
-// returned unwrapped (classified by the caller); backend Upload failures are
-// wrapped in backendUploadError so the caller always masks them.
-func (s *ByteStreamService) bufferedUpload(ctx context.Context, stream bytestream.ByteStream_WriteServer, storageBackend backend.Uploader, req *writeRequest, maxBytes int64) (int64, error) {
-	// Create a buffer that will be filled in the background before sending its content to the backend
-	buffer := newStreamReader(maxBytes)
-	// Add data from the first request
-	if err := buffer.Write(req.GetData()); err != nil {
-		return 0, err
+// spillVerifyUpload streams the client's upload to a temporary file on the local
+// staging disk while computing its SHA256, verifies the computed digest against
+// the client-declared one, and only then hands the verified file to the backend
+// for storage under the canonical key. Unverified bytes never reach the backend,
+// so the canonical key can never hold content that does not hash to its
+// digest. CAS memory stays bounded because the artifact lives on disk, and
+// the *os.File handed to Upload lets object-store SDKs stream it in bounded parts
+// via io.ReaderAt/io.Seeker rather than buffering it in memory.
+//
+// It returns the number of bytes committed. A digest mismatch is returned as a
+// *digestMismatchError; a backend Upload failure as a *backendUploadError; spill
+// errors (client disconnect, exceeded size cap, staging-disk write failure) are
+// returned unwrapped for the caller to classify.
+func (s *ByteStreamService) spillVerifyUpload(ctx context.Context, stream bytestream.ByteStream_WriteServer, storageBackend backend.Uploader, req *writeRequest, maxBytes int64) (int64, error) {
+	f, err := os.CreateTemp(s.stagingDir, stagingFilePrefix+"*")
+	if err != nil {
+		return 0, fmt.Errorf("creating staging file: %w", err)
 	}
+	// Clean up the staging file on every exit path: nothing is left on disk
+	// whether the upload is rejected, fails, or succeeds. Close before remove so
+	// the handle is released; on Linux either order unlinks the file regardless.
+	defer func() {
+		_ = f.Close()
+		if err := os.Remove(f.Name()); err != nil && !errors.Is(err, os.ErrNotExist) {
+			s.log.Warnw("msg", "failed to remove staging file", "path", f.Name(), "error", err.Error())
+		}
+	}()
 
-	// Start a goroutine that will fill the buffer in the background
-	go bufferStream(ctx, stream, buffer, s.log)
-
-	// Block until the buffer has been filled or the upload process has been canceled
-	if err := <-buffer.errorChan; err != nil {
+	// Tee the stream into the file and a SHA256 hasher in one pass.
+	hasher := sha256.New()
+	size, err := spillStream(ctx, stream, io.MultiWriter(f, hasher), req.GetData(), maxBytes, s.log, req.resource.Digest)
+	if err != nil {
 		return 0, err
 	}
 
-	s.log.Infow("msg", "artifact received, uploading now to backend", "name", req.resource.FileName, "digest", req.resource.Digest, "size", buffer.size)
-	if err := storageBackend.Upload(ctx, buffer, req.resource); err != nil {
-		return 0, &backendUploadError{err}
+	// Fail closed: if the streamed bytes do not hash to the declared digest,
+	// reject the upload and send nothing to the backend.
+	if got := hex.EncodeToString(hasher.Sum(nil)); got != req.resource.Digest {
+		return 0, &digestMismatchError{got: got, want: req.resource.Digest}
 	}
 
-	return buffer.size, nil
-}
-
-// streamUpload pipes the client stream straight into the backend's Upload
-// without buffering the whole artifact in memory. A background goroutine feeds
-// received chunks into an io.Pipe while Upload consumes the other end, so the
-// two run concurrently and peak memory stays bounded (PFM-6923). It returns the
-// total number of bytes committed to the backend.
-func (s *ByteStreamService) streamUpload(ctx context.Context, stream bytestream.ByteStream_WriteServer, storageBackend backend.Uploader, req *writeRequest, maxBytes int64) (int64, error) {
-	pr, pw := io.Pipe()
-
-	var (
-		uploadedSize int64
-		feedErr      error
-	)
-	done := make(chan struct{})
-	go func() {
-		defer close(done)
-		uploadedSize, feedErr = feedPipe(ctx, stream, pw, req.GetData(), maxBytes, s.log, req.resource.Digest)
-		// Closing with feedErr signals EOF to the reader when nil, or propagates
-		// the failure so Upload stops reading.
-		_ = pw.CloseWithError(feedErr)
-	}()
-
-	uploadErr := storageBackend.Upload(ctx, streamingReader{pr}, req.resource)
-	// If Upload returned without draining the pipe (a backend failure, or a
-	// backend that reports success without reading to EOF), the feeding goroutine
-	// may still be blocked on Write; closing the read end unblocks it. Then wait
-	// for it so uploadedSize/feedErr are safe to read.
-	_ = pr.CloseWithError(uploadErr)
-	<-done
-
-	// errPipeConsumerGone means the feed only failed because the reader (Upload)
-	// stopped consuming — a consequence of the upload outcome, not a genuine
-	// stream-read failure, so the backend's own result is authoritative.
-	if errors.Is(feedErr, errPipeConsumerGone) {
-		feedErr = nil
+	// Rewind so the backend reads from the start. A seekable body also lets the
+	// AWS SDK learn the exact length and take its zero-copy SectionReader fast
+	// path instead of buffering parts in memory.
+	if _, err := f.Seek(0, io.SeekStart); err != nil {
+		return 0, fmt.Errorf("rewinding staging file: %w", err)
 	}
 
-	// A genuine feed-side error (client disconnect, exceeded size cap, stream
-	// read failure) is the precise signal and takes precedence: when it occurs it
-	// is what induced the backend error through the pipe. Returned unwrapped so
-	// the caller classifies it (disconnect / ResourceExhausted / mask).
-	if feedErr != nil {
-		return 0, feedErr
-	}
-	// A backend failure is wrapped so the caller always masks it, never mistaking
-	// a backend-side reset/cancellation for a client disconnect.
-	if uploadErr != nil {
-		return 0, &backendUploadError{uploadErr}
+	s.log.Infow("msg", "artifact verified, uploading now to backend", "name", req.resource.FileName, "digest", req.resource.Digest, "size", size)
+	// IMPORTANT: hand the *os.File to Upload unwrapped. Wrapping it (io.TeeReader,
+	// io.LimitReader, a progress reader) hides io.ReaderAt/io.Seeker and silently
+	// forces the object-store SDK back onto in-memory multipart buffering.
+	if err := storageBackend.Upload(ctx, f, req.resource); err != nil {
+		return 0, &backendUploadError{err}
 	}
 
-	return uploadedSize, nil
+	return size, nil
 }
 
+// stagingFilePrefix names the temporary upload files so the boot-time sweep can
+// distinguish CAS's own leftovers from anything else that might share the dir.
+const stagingFilePrefix = "cas-upload-"
+
 // backendUploadError marks a failure returned by the storage backend's Upload,
 // as opposed to an error reading the client stream. Backend failures are always
 // masked as internal errors and are never interpreted as a client disconnect or
@@ -259,28 +240,21 @@ type backendUploadError struct{ err error }
 func (e *backendUploadError) Error() string { return e.err.Error() }
 func (e *backendUploadError) Unwrap() error { return e.err }
 
-// errPipeConsumerGone is returned by feedPipe when a write to the pipe fails,
-// which only happens once the reader (the backend Upload) has stopped consuming
-// — because Upload returned and streamUpload closed the read end, or because it
-// failed. It is not a genuine stream-read failure; streamUpload defers to the
-// backend's own error in that case.
-var errPipeConsumerGone = errors.New("pipe consumer stopped reading")
-
-// streamingReader wraps the upload pipe reader with a stable string form. The
-// pipe is written to concurrently while the backend reads it; exposing the bare
-// *io.PipeReader lets a reflective consumer (a structured logger, a test's mock
-// matcher, etc.) walk the pipe's internal state and race with the writer. The
-// wrapper keeps io.Reader behaviour while presenting an opaque identity to fmt.
-type streamingReader struct {
-	io.Reader
-}
+// digestMismatchError marks an upload whose streamed bytes do not hash to the
+// client-declared digest. It is surfaced to the client as InvalidArgument: the
+// request is malformed (the declared key does not describe the content), and no
+// bytes are ever written to the backend.
+type digestMismatchError struct{ got, want string }
 
-func (streamingReader) String() string { return "cas-streaming-upload" }
+func (e *digestMismatchError) Error() string {
+	return fmt.Sprintf("uploaded content does not match the declared digest: got=%s, want=%s", e.got, e.want)
+}
 
-// feedPipe forwards the artifact from the client stream into pw, enforcing the
-// max upload size as it goes. firstData is the payload already read from the
-// first request. It returns the total number of bytes forwarded.
-func feedPipe(ctx context.Context, stream bytestream.ByteStream_WriteServer, pw *io.PipeWriter, firstData []byte, maxSize int64, log *log.Helper, digest string) (int64, error) {
+// spillStream forwards the artifact from the client stream into w (the staging
+// file tee'd into a SHA256 hasher), enforcing the max upload size as it goes.
+// firstData is the payload already read from the first request. It returns the
+// total number of bytes written. It reads the client stream straight into w.
+func spillStream(ctx context.Context, stream bytestream.ByteStream_WriteServer, w io.Writer, firstData []byte, maxSize int64, log *log.Helper, digest string) (int64, error) {
 	var size int64
 	write := func(data []byte) error {
 		if len(data) == 0 {
@@ -290,16 +264,13 @@ func feedPipe(ctx context.Context, stream bytestream.ByteStream_WriteServer, pw
 		if err := checkUploadSize(size, maxSize); err != nil {
 			return err
 		}
-		if _, err := pw.Write(data); err != nil {
-			// A write only fails once the reader has gone away; surface it as the
-			// consumer-gone sentinel so streamUpload defers to the backend result
-			// rather than treating this as a client-side stream failure.
-			return errPipeConsumerGone
+		if _, err := w.Write(data); err != nil {
+			return fmt.Errorf("writing to staging file: %w", err)
 		}
 		return nil
 	}
 
-	// Forward the data from the first request.
+	// Write the data from the first request.
 	if err := write(firstData); err != nil {
 		return size, err
 	}
@@ -320,14 +291,14 @@ func feedPipe(ctx context.Context, stream bytestream.ByteStream_WriteServer, pw
 				return size, err
 			}
 
-			// Forward this request's data first: a spec-compliant client may set
+			// Write this request's data first: a spec-compliant client may set
 			// finish_write=true on the same message that carries the final chunk,
 			// so the data must be written before the finish check or it is lost.
 			if err := write(req.GetData()); err != nil {
 				return size, err
 			}
 
-			log.Debugw("msg", "upload chunk received (streaming)", "digest", digest, "currentSize", size, "maxSize", maxSize, "chunkSize", len(req.GetData()))
+			log.Debugw("msg", "upload chunk received", "digest", digest, "currentSize", size, "maxSize", maxSize, "chunkSize", len(req.GetData()))
 
 			// Check if the client has finished sending data
 			if req.GetFinishWrite() {
@@ -395,85 +366,8 @@ func (s *ByteStreamService) Read(req *bytestream.ReadRequest, stream bytestream.
 	return nil
 }
 
-// Store the data received from the stream in a buffer and send a signal when finished
-// This is done in a separate goroutine to avoid blocking the stream
-func bufferStream(ctx context.Context, stream bytestream.ByteStream_WriteServer, buffer *streamReader, log *log.Helper) {
-	// Send termination signal when finished receiving data
-	var bufferErr error
-	defer func() {
-		buffer.errorChan <- bufferErr
-	}()
-
-	for {
-		select {
-		case <-ctx.Done():
-			// DeadlineExceeded, or Canceled
-			bufferErr = ctx.Err()
-			return
-		default:
-			// Extract the next chunk of data from the stream request
-			req, err := getWriteRequest(stream)
-			if err != nil {
-				// If we have finished reading the stream we don't consider it a real error
-				if !errors.Is(err, io.EOF) {
-					bufferErr = err
-				}
-				return
-			}
-
-			// Write the data first: a spec-compliant client may set
-			// finish_write=true on the same message that carries the final chunk,
-			// so the data must be buffered before the finish check or it is lost.
-			if err = buffer.Write(req.GetData()); err != nil {
-				bufferErr = err
-				return
-			}
-
-			log.Debugw("msg", "upload chunk received", "digest", req.resource.Digest, "currentSize", buffer.size, "maxSize", buffer.maxSize, "chunkSize", len(req.GetData()))
-
-			// Check if the client has finished sending data
-			if req.GetFinishWrite() {
-				return
-			}
-		}
-	}
-}
-
-type streamReader struct {
-	*bytes.Buffer
-	// total size of the in-memory buffer in bytes
-	size int64
-	// Max size allowed to be uploaded
-	maxSize int64
-	// there was an error during stream data filling
-	errorChan chan error
-}
-
-// Wrapper around a buffer that adds
-// the ability to record the total size of the data that went through it
-// and a channel to be used by the clients to signal when the buffer has been filled
-func newStreamReader(maxSize int64) *streamReader {
-	return &streamReader{
-		Buffer:    bytes.NewBuffer(nil),
-		errorChan: make(chan error),
-		maxSize:   maxSize,
-	}
-}
-
-func (r *streamReader) Write(data []byte) error {
-	r.size += int64(len(data))
-
-	if err := checkUploadSize(r.size, r.maxSize); err != nil {
-		return err
-	}
-
-	_, err := r.Buffer.Write(data)
-	return err
-}
-
 // checkUploadSize returns an ErrUploadSizeExceeded when total exceeds maxSize.
-// maxSize == 0 means no limit. It is shared by the buffered (streamReader) and
-// streaming (feedPipe) paths so their cap semantics cannot drift.
+// maxSize == 0 means no limit.
 func checkUploadSize(total, maxSize int64) error {
 	if maxSize != 0 && total > maxSize {
 		return backend.NewErrUploadSizeExceeded(total, maxSize)
diff --git a/app/artifact-cas/internal/service/bytestream_download_test.go b/app/artifact-cas/internal/service/bytestream_download_test.go
index 95abb97ba..3025cab29 100644
--- a/app/artifact-cas/internal/service/bytestream_download_test.go
+++ b/app/artifact-cas/internal/service/bytestream_download_test.go
@@ -31,12 +31,10 @@ import (
 	"google.golang.org/grpc/codes"
 )
 
-// These tests lock down the DOWNLOAD digest-verification behavior. The download
-// path is intentionally unchanged by the streaming-upload work (PFM-6923); this
-// battery guards it against regressions — the server must stream the stored
-// bytes back, compute their sha256 across however many chunks the backend
-// produces, and reject any content whose digest does not match the requested
-// resource name.
+// These tests lock down the DOWNLOAD digest-verification behavior — the server
+// must stream the stored bytes back, compute their sha256 across however many
+// chunks the backend produces, and reject any content whose digest does not
+// match the requested resource name.
 
 // fakeReadServer is a minimal bytestream.ByteStream_ReadServer that records the
 // data chunks the streamWriter sends. Only Send is exercised by streamWriter.
diff --git a/app/artifact-cas/internal/service/bytestream_streaming_test.go b/app/artifact-cas/internal/service/bytestream_streaming_test.go
index 777790e17..2931abc2b 100644
--- a/app/artifact-cas/internal/service/bytestream_streaming_test.go
+++ b/app/artifact-cas/internal/service/bytestream_streaming_test.go
@@ -21,12 +21,11 @@ import (
 	"encoding/hex"
 	"fmt"
 	"io"
+	"os"
 	"syscall"
 	"testing"
-	"time"
 
 	v1 "github.com/chainloop-dev/chainloop/app/artifact-cas/api/cas/v1"
-	"github.com/chainloop-dev/chainloop/pkg/blobmanager/mocks"
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
 	"google.golang.org/genproto/googleapis/bytestream"
@@ -36,15 +35,6 @@ import (
 
 const streamingBackendType = "streaming-backend-type"
 
-// streamingUploaderDownloader wraps a mock backend and advertises streaming
-// support, so the CAS service pipes the upload straight through instead of
-// buffering it. Used only in tests to exercise the streaming code path.
-type streamingUploaderDownloader struct {
-	*mocks.UploaderDownloader
-}
-
-func (streamingUploaderDownloader) SupportsStreaming() bool { return true }
-
 // --- test helpers ---------------------------------------------------------
 
 // streamingUpCtx returns an uploader context routed to the streaming backend.
@@ -115,6 +105,98 @@ func (s *bytestreamSuite) expectStreamingUpload(resource *v1.CASResource, upload
 	return received
 }
 
+// --- upload integrity verification -----------------------------------------
+
+// TestWriteDigestMismatchRejected is the core upload-integrity guarantee: bytes
+// that do not hash to the client-declared digest are rejected with
+// InvalidArgument, nothing is ever sent to the backend, and no audit event is
+// emitted. Without verification an attacker could store arbitrary content under
+// an arbitrary digest key.
+func (s *bytestreamSuite) TestWriteDigestMismatchRejected() {
+	content := []byte("this is the real content that was actually streamed")
+	// The declared digest belongs to DIFFERENT content than what is sent.
+	resource := &v1.CASResource{Digest: sha256Hex([]byte("something else entirely")), FileName: "artifact.bin"}
+	s.streamingBackend.On("Exists", mock.Anything, resource.Digest).Return(false, nil)
+	// Upload must NOT be called for a mismatching digest.
+
+	stream, err := s.client.Write(streamingUpCtx(""))
+	s.NoError(err)
+	sendInChunks(s.T(), stream, encodeResource(s.T(), resource), content, 7)
+
+	_, err = stream.CloseAndRecv()
+	assertGRPCError(s.T(), err, codes.InvalidArgument, "does not match the declared digest")
+	s.streamingBackend.AssertNotCalled(s.T(), "Upload", mock.Anything, mock.Anything, mock.Anything)
+	s.Empty(s.audit.published)
+}
+
+// TestWriteBackendReceivesSeekableFile asserts the backend's Upload is handed a
+// value satisfying io.ReaderAt+io.Seeker (an *os.File), so the AWS SDK's
+// zero-buffer fast path is taken. Wrapping the file on the way to Upload (a
+// TeeReader, progress reader, LimitReader) would silently reinstate part
+// buffering, so this guards against that regression.
+func (s *bytestreamSuite) TestWriteBackendReceivesSeekableFile() {
+	content := deterministicBytes(64 * 1024)
+	resource := resourceWithDigest(content, "artifact.bin")
+	var isReaderAt, isSeeker bool
+	s.streamingBackend.On("Exists", mock.Anything, resource.Digest).Return(false, nil)
+	s.streamingBackend.On("Upload", mock.Anything, mock.Anything, resource).Return(nil).Run(func(args mock.Arguments) {
+		r := args.Get(1)
+		_, isReaderAt = r.(io.ReaderAt)
+		_, isSeeker = r.(io.Seeker)
+		_, _ = io.ReadAll(r.(io.Reader))
+	})
+
+	stream, err := s.client.Write(streamingUpCtx(""))
+	s.NoError(err)
+	sendInChunks(s.T(), stream, encodeResource(s.T(), resource), content, 8192)
+
+	_, err = stream.CloseAndRecv()
+	s.NoError(err)
+	s.True(isReaderAt, "backend must receive an io.ReaderAt for the SDK zero-buffer fast path")
+	s.True(isSeeker, "backend must receive an io.Seeker for the SDK zero-buffer fast path")
+}
+
+// requireStagingEmpty asserts the per-test staging directory holds no files, so
+// verified-and-uploaded or rejected content never accumulates on disk.
+func (s *bytestreamSuite) requireStagingEmpty() {
+	entries, err := os.ReadDir(s.stagingDir)
+	s.Require().NoError(err)
+	s.Emptyf(entries, "staging dir must be left empty, found: %v", entries)
+}
+
+// TestWriteStagingCleanupOnSuccess: after a successful upload the staging file
+// is removed.
+func (s *bytestreamSuite) TestWriteStagingCleanupOnSuccess() {
+	content := []byte("staged, verified, then uploaded")
+	resource := resourceWithDigest(content, "ok.bin")
+	received := s.expectStreamingUpload(resource, nil)
+
+	stream, err := s.client.Write(streamingUpCtx(""))
+	s.NoError(err)
+	sendInChunks(s.T(), stream, encodeResource(s.T(), resource), content, 8)
+
+	_, err = stream.CloseAndRecv()
+	s.NoError(err)
+	<-received
+	s.requireStagingEmpty()
+}
+
+// TestWriteStagingCleanupOnMismatch: after a rejected digest mismatch the
+// staging file is removed too — unverified bytes never linger.
+func (s *bytestreamSuite) TestWriteStagingCleanupOnMismatch() {
+	content := []byte("content that will not match the declared digest")
+	resource := &v1.CASResource{Digest: sha256Hex([]byte("different")), FileName: "bad.bin"}
+	s.streamingBackend.On("Exists", mock.Anything, resource.Digest).Return(false, nil)
+
+	stream, err := s.client.Write(streamingUpCtx(""))
+	s.NoError(err)
+	sendInChunks(s.T(), stream, encodeResource(s.T(), resource), content, 8)
+
+	_, err = stream.CloseAndRecv()
+	assertGRPCError(s.T(), err, codes.InvalidArgument, "does not match the declared digest")
+	s.requireStagingEmpty()
+}
+
 // --- integrity / normal cases --------------------------------------------
 
 // TestWriteStreamingSingleChunkOK: a single-chunk streaming upload stores the
@@ -181,50 +263,6 @@ func (s *bytestreamSuite) TestWriteStreamingManyChunksIntegrity() {
 	s.Len(gotBytes, len(content))
 }
 
-// TestWriteStreamingConsumesBeforeFinish is the core bounded-memory regression
-// test (PFM-6923): the backend must begin consuming the upload BEFORE the client
-// finishes sending. The handshake blocks the client's tail chunk until Upload
-// has started; with a buffering implementation Upload is never entered until the
-// whole stream is received, so this deadlocks and fails via timeout.
-func (s *bytestreamSuite) TestWriteStreamingConsumesBeforeFinish() {
-	data := []byte("hello streaming world")
-	resource := resourceWithDigest(data, "artifact.bin")
-
-	uploadStarted := make(chan struct{})
-	received := make(chan []byte, 1)
-	s.streamingBackend.On("Exists", mock.Anything, resource.Digest).Return(false, nil)
-	s.streamingBackend.On("Upload", mock.Anything, mock.Anything, resource).
-		Return(nil).Run(func(args mock.Arguments) {
-		close(uploadStarted)
-		got, err := io.ReadAll(args.Get(1).(io.Reader))
-		s.NoError(err)
-		received <- got
-	})
-
-	stream, err := s.client.Write(streamingUpCtx(""))
-	s.NoError(err)
-	s.NoError(stream.Send(&bytestream.WriteRequest{
-		ResourceName: encodeResource(s.T(), resource),
-		Data:         data[:6],
-	}))
-
-	select {
-	case <-uploadStarted:
-	case <-time.After(5 * time.Second):
-		s.FailNow("backend Upload was not started before the stream finished — upload is being buffered, not streamed")
-	}
-
-	s.NoError(stream.Send(&bytestream.WriteRequest{
-		ResourceName: encodeResource(s.T(), resource),
-		Data:         data[6:],
-	}))
-
-	got, err := stream.CloseAndRecv()
-	s.NoError(err)
-	s.Equal(int64(len(data)), got.CommittedSize)
-	s.Equal(data, <-received)
-}
-
 // TestWriteStreamingEmptyArtifact: a zero-byte artifact streams cleanly and is
 // committed with size 0.
 func (s *bytestreamSuite) TestWriteStreamingEmptyArtifact() {
@@ -359,8 +397,10 @@ func (s *bytestreamSuite) TestWriteStreamingMaxSizeUnlimited() {
 // TestWriteStreamingBackendError: a generic backend Upload failure surfaces as
 // Internal and emits no audit event.
 func (s *bytestreamSuite) TestWriteStreamingBackendError() {
-	s.streamingBackend.On("Exists", mock.Anything, s.resource.Digest).Return(false, nil)
-	s.streamingBackend.On("Upload", mock.Anything, mock.Anything, s.resource).
+	data := []byte("hello world")
+	resource := resourceWithDigest(data, "artifact.bin")
+	s.streamingBackend.On("Exists", mock.Anything, resource.Digest).Return(false, nil)
+	s.streamingBackend.On("Upload", mock.Anything, mock.Anything, resource).
 		Return(fmt.Errorf("object store rejected the upload")).Run(func(args mock.Arguments) {
 		_, _ = io.ReadAll(args.Get(1).(io.Reader))
 	})
@@ -368,8 +408,8 @@ func (s *bytestreamSuite) TestWriteStreamingBackendError() {
 	stream, err := s.client.Write(streamingUpCtx(""))
 	s.NoError(err)
 	s.NoError(stream.Send(&bytestream.WriteRequest{
-		ResourceName: encodeResource(s.T(), s.resource),
-		Data:         []byte("hello world"),
+		ResourceName: encodeResource(s.T(), resource),
+		Data:         data,
 	}))
 
 	_, err = stream.CloseAndRecv()
@@ -381,16 +421,18 @@ func (s *bytestreamSuite) TestWriteStreamingBackendError() {
 // wrapping a network reset must be masked as Internal, NOT mistaken for a client
 // disconnect (which would falsely report success and silently drop the blob).
 func (s *bytestreamSuite) TestWriteStreamingBackendResetNotTreatedAsDisconnect() {
-	s.streamingBackend.On("Exists", mock.Anything, s.resource.Digest).Return(false, nil)
-	s.streamingBackend.On("Upload", mock.Anything, mock.Anything, s.resource).
+	data := []byte("hello world")
+	resource := resourceWithDigest(data, "artifact.bin")
+	s.streamingBackend.On("Exists", mock.Anything, resource.Digest).Return(false, nil)
+	s.streamingBackend.On("Upload", mock.Anything, mock.Anything, resource).
 		Return(fmt.Errorf("connection to object store failed: %w", syscall.ECONNRESET)).
 		Run(func(args mock.Arguments) { _, _ = io.ReadAll(args.Get(1).(io.Reader)) })
 
 	stream, err := s.client.Write(streamingUpCtx(""))
 	s.NoError(err)
 	s.NoError(stream.Send(&bytestream.WriteRequest{
-		ResourceName: encodeResource(s.T(), s.resource),
-		Data:         []byte("hello world"),
+		ResourceName: encodeResource(s.T(), resource),
+		Data:         data,
 	}))
 
 	_, err = stream.CloseAndRecv()
@@ -403,16 +445,18 @@ func (s *bytestreamSuite) TestWriteStreamingBackendResetNotTreatedAsDisconnect()
 // TestWriteStreamingBackendCanceledNotTreatedAsDisconnect: same guard for an
 // Upload error wrapping context.Canceled originating backend-side.
 func (s *bytestreamSuite) TestWriteStreamingBackendCanceledNotTreatedAsDisconnect() {
-	s.streamingBackend.On("Exists", mock.Anything, s.resource.Digest).Return(false, nil)
-	s.streamingBackend.On("Upload", mock.Anything, mock.Anything, s.resource).
+	data := []byte("hello world")
+	resource := resourceWithDigest(data, "artifact.bin")
+	s.streamingBackend.On("Exists", mock.Anything, resource.Digest).Return(false, nil)
+	s.streamingBackend.On("Upload", mock.Anything, mock.Anything, resource).
 		Return(fmt.Errorf("backend deadline: %w", context.Canceled)).
 		Run(func(args mock.Arguments) { _, _ = io.ReadAll(args.Get(1).(io.Reader)) })
 
 	stream, err := s.client.Write(streamingUpCtx(""))
 	s.NoError(err)
 	s.NoError(stream.Send(&bytestream.WriteRequest{
-		ResourceName: encodeResource(s.T(), s.resource),
-		Data:         []byte("hello world"),
+		ResourceName: encodeResource(s.T(), resource),
+		Data:         data,
 	}))
 
 	_, err = stream.CloseAndRecv()
@@ -425,13 +469,14 @@ func (s *bytestreamSuite) TestWriteStreamingBackendCanceledNotTreatedAsDisconnec
 // service's own reader-close must not surface as an Internal error.
 func (s *bytestreamSuite) TestWriteStreamingBackendSuccessWithoutDrain() {
 	data := []byte("hello world")
-	s.streamingBackend.On("Exists", mock.Anything, s.resource.Digest).Return(false, nil)
-	s.streamingBackend.On("Upload", mock.Anything, mock.Anything, s.resource).Return(nil) // no drain
+	resource := resourceWithDigest(data, "artifact.bin")
+	s.streamingBackend.On("Exists", mock.Anything, resource.Digest).Return(false, nil)
+	s.streamingBackend.On("Upload", mock.Anything, mock.Anything, resource).Return(nil) // no drain
 
 	stream, err := s.client.Write(streamingUpCtx(""))
 	s.NoError(err)
 	s.NoError(stream.Send(&bytestream.WriteRequest{
-		ResourceName: encodeResource(s.T(), s.resource),
+		ResourceName: encodeResource(s.T(), resource),
 		Data:         data,
 	}))
 
@@ -445,18 +490,12 @@ func (s *bytestreamSuite) TestWriteStreamingBackendSuccessWithoutDrain() {
 }
 
 // TestWriteStreamingClientDisconnect: when the client cancels mid-upload, the
-// server treats it as a disconnect (no error masking, no audit) and the backend
-// sees the stream abort through the pipe.
+// server aborts before verification completes, so nothing is ever sent to the
+// backend and no audit event is emitted.
 func (s *bytestreamSuite) TestWriteStreamingClientDisconnect() {
-	uploadStarted := make(chan struct{})
-	readErr := make(chan error, 1)
-	s.streamingBackend.On("Exists", mock.Anything, s.resource.Digest).Return(false, nil)
-	s.streamingBackend.On("Upload", mock.Anything, mock.Anything, s.resource).Maybe().
-		Return(nil).Run(func(args mock.Arguments) {
-		close(uploadStarted)
-		_, err := io.ReadAll(args.Get(1).(io.Reader))
-		readErr <- err
-	})
+	// Exists may or may not be reached depending on how fast the cancellation
+	// races the handler; the invariant under test is that Upload never is.
+	s.streamingBackend.On("Exists", mock.Anything, s.resource.Digest).Maybe().Return(false, nil)
 
 	ctx, cancel := context.WithCancel(streamingUpCtx(""))
 	stream, err := s.client.Write(ctx)
@@ -465,26 +504,13 @@ func (s *bytestreamSuite) TestWriteStreamingClientDisconnect() {
 		ResourceName: encodeResource(s.T(), s.resource),
 		Data:         []byte("partial upload"),
 	}))
-
-	// Wait until the backend is actively consuming, then cancel the client so the
-	// disconnect happens deterministically mid-stream.
-	select {
-	case <-uploadStarted:
-	case <-time.After(5 * time.Second):
-		s.FailNow("backend Upload was not started")
-	}
 	cancel()
 
-	select {
-	case err := <-readErr:
-		// The backend saw the aborted stream (pipe closed with the cancellation).
-		s.Error(err)
-	case <-time.After(5 * time.Second):
-		s.FailNow("backend Upload did not observe the client disconnect")
-	}
-
-	// A disconnect is not a successful upload: no audit event is emitted.
+	_, err = stream.CloseAndRecv()
+	// The client canceled: the RPC ends in error and nothing is stored.
+	s.Error(err)
 	s.Empty(s.audit.published)
+	s.streamingBackend.AssertNotCalled(s.T(), "Upload", mock.Anything, mock.Anything, mock.Anything)
 }
 
 // --- dedup ----------------------------------------------------------------
diff --git a/app/artifact-cas/internal/service/bytestream_test.go b/app/artifact-cas/internal/service/bytestream_test.go
index 096a467ea..968fe00c9 100644
--- a/app/artifact-cas/internal/service/bytestream_test.go
+++ b/app/artifact-cas/internal/service/bytestream_test.go
@@ -48,42 +48,6 @@ import (
 	"google.golang.org/grpc/test/bufconn"
 )
 
-func (s *bytestreamSuite) TestStreamReader() {
-	buffer := newStreamReader(0)
-	// Write twice and check the length
-	err := buffer.Write([]byte("hello"))
-	s.NoError(err)
-	s.Equal(int64(5), buffer.size)
-	err = buffer.Write([]byte("chainloop"))
-	s.NoError(err)
-	s.Equal(int64(14), buffer.size)
-	// The buffer length also matches
-	s.Equal(14, buffer.Len())
-
-	// Start reading
-	writer := bytes.NewBuffer(nil)
-	copied, err := io.Copy(writer, buffer)
-	s.Equal(int64(14), copied)
-	s.NoError(err)
-	// The buffer length is still 14 to indicate what it has processed
-	s.Equal(int64(14), buffer.size)
-	// but the internal one is 0
-	s.Equal(0, buffer.Len())
-}
-
-func (s *bytestreamSuite) TestStreamReaderOverflow() {
-	// a buffer with 8 bytes limit
-	buffer := newStreamReader(8)
-	// Write twice and check the length
-	err := buffer.Write([]byte("hello"))
-	s.NoError(err)
-	s.Equal(int64(5), buffer.size)
-	err = buffer.Write([]byte("chainloop"))
-	s.Error(err)
-	s.True(backend.IsUploadSizeExceeded(err))
-	s.ErrorContains(err, "max size of upload exceeded")
-}
-
 func (s *bytestreamSuite) TestWrite() {
 	ctx := s.upCtx
 
@@ -210,19 +174,20 @@ func (s *bytestreamSuite) TestWriteExistInternalTraffic() {
 
 func (s *bytestreamSuite) TestWriteOK() {
 	data := []byte("hello world")
-	s.ociBackend.On("Exists", mock.Anything, s.resource.Digest).Return(false, nil)
-	s.ociBackend.On("Upload", mock.Anything, mock.Anything, s.resource).Return(nil)
+	resource := resourceWithDigest(data, "skynet.exe")
+	s.ociBackend.On("Exists", mock.Anything, resource.Digest).Return(false, nil)
+	s.ociBackend.On("Upload", mock.Anything, mock.Anything, resource).Return(nil)
 
 	stream, err := s.client.Write(s.upCtx)
 	s.NoError(err)
 	// Multiple chunks
 	s.NoError(stream.Send(&bytestream.WriteRequest{
-		ResourceName: encodeResource(s.T(), s.resource),
+		ResourceName: encodeResource(s.T(), resource),
 		Data:         data[:5],
 	}))
 
 	s.NoError(stream.Send(&bytestream.WriteRequest{
-		ResourceName: encodeResource(s.T(), s.resource),
+		ResourceName: encodeResource(s.T(), resource),
 		Data:         data[5:],
 	}))
 
@@ -234,20 +199,22 @@ func (s *bytestreamSuite) TestWriteOK() {
 	s.Require().Len(s.audit.published, 1)
 	info := decodeArtifactEvent(s.T(), s.audit.published[0])
 	s.False(info.Skipped)
-	s.Equal(s.resource.Digest, info.Digest)
+	s.Equal(resource.Digest, info.Digest)
 	s.Equal(int64(len(data)), info.SizeBytes)
-	s.Equal(s.resource.FileName, info.FileName)
+	s.Equal(resource.FileName, info.FileName)
 }
 
 func (s *bytestreamSuite) TestWriteErrorUploading() {
-	s.ociBackend.On("Exists", mock.Anything, s.resource.Digest).Return(false, nil)
-	s.ociBackend.On("Upload", mock.Anything, mock.Anything, s.resource).Return(errors.New("error uploading"))
+	data := []byte("hello world")
+	resource := resourceWithDigest(data, "skynet.exe")
+	s.ociBackend.On("Exists", mock.Anything, resource.Digest).Return(false, nil)
+	s.ociBackend.On("Upload", mock.Anything, mock.Anything, resource).Return(errors.New("error uploading"))
 
 	stream, err := s.client.Write(s.upCtx)
 	s.NoError(err)
 	s.NoError(stream.Send(&bytestream.WriteRequest{
-		ResourceName: encodeResource(s.T(), s.resource),
-		Data:         []byte("hello world"),
+		ResourceName: encodeResource(s.T(), resource),
+		Data:         data,
 	}))
 
 	_, err = stream.CloseAndRecv()
@@ -256,9 +223,9 @@ func (s *bytestreamSuite) TestWriteErrorUploading() {
 	s.Empty(s.audit.published)
 }
 
-// TestWriteBufferedMultiChunkIntegrity: the buffered/OCI path reassembles a
-// multi-chunk upload byte-for-byte and hands the backend content whose sha256
-// matches the declared digest — parity with the streaming path.
+// TestWriteBufferedMultiChunkIntegrity: an OCI-backed multi-chunk upload is
+// reassembled byte-for-byte and the backend receives content whose sha256
+// matches the declared digest.
 func (s *bytestreamSuite) TestWriteBufferedMultiChunkIntegrity() {
 	content := []byte("chainloop attestation payload spanning multiple stream chunks")
 	resource := resourceWithDigest(content, "artifact.bin")
@@ -314,15 +281,17 @@ func (s *bytestreamSuite) TestWriteBufferedFinishWriteWithData() {
 // masks a backend-side failure wrapping a network reset as Internal, never
 // mistaking it for a client disconnect (which would falsely report success).
 func (s *bytestreamSuite) TestWriteBufferedBackendResetNotTreatedAsDisconnect() {
-	s.ociBackend.On("Exists", mock.Anything, s.resource.Digest).Return(false, nil)
-	s.ociBackend.On("Upload", mock.Anything, mock.Anything, s.resource).
+	data := []byte("hello world")
+	resource := resourceWithDigest(data, "artifact.bin")
+	s.ociBackend.On("Exists", mock.Anything, resource.Digest).Return(false, nil)
+	s.ociBackend.On("Upload", mock.Anything, mock.Anything, resource).
 		Return(fmt.Errorf("connection to registry failed: %w", syscall.ECONNRESET))
 
 	stream, err := s.client.Write(s.upCtx)
 	s.NoError(err)
 	s.NoError(stream.Send(&bytestream.WriteRequest{
-		ResourceName: encodeResource(s.T(), s.resource),
-		Data:         []byte("hello world"),
+		ResourceName: encodeResource(s.T(), resource),
+		Data:         data,
 	}))
 
 	_, err = stream.CloseAndRecv()
@@ -335,15 +304,17 @@ func (s *bytestreamSuite) TestWriteBufferedBackendResetNotTreatedAsDisconnect()
 // TestWriteBufferedBackendCanceledNotTreatedAsDisconnect: same guard for an
 // Upload error wrapping context.Canceled originating backend-side.
 func (s *bytestreamSuite) TestWriteBufferedBackendCanceledNotTreatedAsDisconnect() {
-	s.ociBackend.On("Exists", mock.Anything, s.resource.Digest).Return(false, nil)
-	s.ociBackend.On("Upload", mock.Anything, mock.Anything, s.resource).
+	data := []byte("hello world")
+	resource := resourceWithDigest(data, "artifact.bin")
+	s.ociBackend.On("Exists", mock.Anything, resource.Digest).Return(false, nil)
+	s.ociBackend.On("Upload", mock.Anything, mock.Anything, resource).
 		Return(fmt.Errorf("registry deadline: %w", context.Canceled))
 
 	stream, err := s.client.Write(s.upCtx)
 	s.NoError(err)
 	s.NoError(stream.Send(&bytestream.WriteRequest{
-		ResourceName: encodeResource(s.T(), s.resource),
-		Data:         []byte("hello world"),
+		ResourceName: encodeResource(s.T(), resource),
+		Data:         data,
 	}))
 
 	_, err = stream.CloseAndRecv()
@@ -464,6 +435,9 @@ type bytestreamSuite struct {
 	audit            *fakePublisher
 	upCtx            context.Context
 	downCtx          context.Context
+	// stagingDir is the per-test upload staging directory the service is
+	// configured with, so tests can assert it is left clean.
+	stagingDir string
 }
 
 // Run after each test
@@ -522,21 +496,21 @@ func (s *bytestreamSuite) SetupTest() {
 	ociBackend := mocks.NewUploaderDownloader(s.T())
 	ociBackendProvider.On("FromCredentials", mock.Anything, mock.Anything).Maybe().Return(ociBackend, nil)
 
-	// A streaming-capable backend (object stores like S3/Azure). It wraps a mock
-	// so tests can set expectations on it while the service detects it as
-	// streaming via the backend.StreamingUploader interface.
+	// A second, object-store-like backend reachable via the "backend-streaming"
+	// metadata that tests can set expectations on.
 	streamingBackend := mocks.NewUploaderDownloader(s.T())
 	streamingBackendProvider := mocks.NewProvider(s.T())
 	streamingBackendProvider.On("FromCredentials", mock.Anything, mock.Anything).Maybe().
-		Return(&streamingUploaderDownloader{streamingBackend}, nil)
+		Return(streamingBackend, nil)
 
 	s.audit = &fakePublisher{}
+	s.stagingDir = s.T().TempDir()
 	bytestream.RegisterByteStreamServer(
 		server,
 		NewByteStreamService(backend.Providers{
 			backendType:          ociBackendProvider,
 			streamingBackendType: streamingBackendProvider,
-		}, WithLogger(log.DefaultLogger), WithAuditDispatcher(newTestDispatcher(s.audit))),
+		}, WithLogger(log.DefaultLogger), WithAuditDispatcher(newTestDispatcher(s.audit)), WithStagingDir(s.stagingDir)),
 	)
 	go func() {
 		_ = server.Serve(l)
diff --git a/app/artifact-cas/internal/service/service.go b/app/artifact-cas/internal/service/service.go
index a1a24c2ef..6e3cfd7b7 100644
--- a/app/artifact-cas/internal/service/service.go
+++ b/app/artifact-cas/internal/service/service.go
@@ -19,6 +19,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"os"
 	"syscall"
 
 	backend "github.com/chainloop-dev/chainloop/pkg/blobmanager"
@@ -38,6 +39,10 @@ type commonService struct {
 	backends backend.Providers
 	// best-effort audit events publisher, nil-safe
 	audit *AuditDispatcher
+	// stagingDir is the local directory where uploads are staged on disk while
+	// their SHA256 is verified against the declared digest before reaching the
+	// backend. It must be writable; when unset it defaults to the OS temp dir.
+	stagingDir string
 }
 
 func (s *commonService) loadBackend(ctx context.Context, providerType, secretID string) (backend.UploaderDownloader, error) {
@@ -72,10 +77,23 @@ func WithAuditDispatcher(d *AuditDispatcher) NewOpt {
 	}
 }
 
+// WithStagingDir sets the local directory where uploads are spilled and
+// verified before being sent to the backend. An empty path leaves the default
+// (the OS temp dir), which is only appropriate for tests — production must
+// point this at a dedicated writable volume.
+func WithStagingDir(dir string) NewOpt {
+	return func(s *commonService) {
+		if dir != "" {
+			s.stagingDir = dir
+		}
+	}
+}
+
 func newCommonService(backends backend.Providers, opts ...NewOpt) *commonService {
 	s := &commonService{
-		log:      servicelogger.EmptyLogger(),
-		backends: backends,
+		log:        servicelogger.EmptyLogger(),
+		backends:   backends,
+		stagingDir: os.TempDir(),
 	}
 
 	for _, opt := range opts {
diff --git a/app/artifact-cas/internal/service/staging.go b/app/artifact-cas/internal/service/staging.go
new file mode 100644
index 000000000..46aa53949
--- /dev/null
+++ b/app/artifact-cas/internal/service/staging.go
@@ -0,0 +1,61 @@
+//
+// Copyright 2026 The Chainloop Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package service
+
+import (
+	"errors"
+	"fmt"
+	"os"
+	"path/filepath"
+	"strings"
+
+	"github.com/go-kratos/kratos/v2/log"
+)
+
+// SweepStagingDir removes leftover upload staging files from a previous run
+// (e.g. a crash mid-transfer that skipped the deferred cleanup). It only removes
+// files carrying the staging prefix — anything else in the directory is left
+// untouched — and is a no-op when the directory does not exist. It returns the
+// number of files removed. Intended to run once at startup, before serving
+// traffic, so the staging volume starts clean.
+func SweepStagingDir(dir string, logger *log.Helper) (int, error) {
+	entries, err := os.ReadDir(dir)
+	if err != nil {
+		// A not-yet-populated volume is fine; nothing to sweep.
+		if errors.Is(err, os.ErrNotExist) {
+			return 0, nil
+		}
+		return 0, fmt.Errorf("reading staging dir %q: %w", dir, err)
+	}
+
+	var removed int
+	for _, e := range entries {
+		if e.IsDir() || !strings.HasPrefix(e.Name(), stagingFilePrefix) {
+			continue
+		}
+		p := filepath.Join(dir, e.Name())
+		if err := os.Remove(p); err != nil {
+			// Best-effort: log and continue so one stuck file doesn't block boot.
+			logger.Warnw("msg", "failed to remove leftover staging file", "path", p, "error", err.Error())
+			continue
+		}
+		removed++
+	}
+	if removed > 0 {
+		logger.Infow("msg", "swept leftover staging files", "dir", dir, "removed", removed)
+	}
+	return removed, nil
+}
diff --git a/app/artifact-cas/internal/service/staging_test.go b/app/artifact-cas/internal/service/staging_test.go
new file mode 100644
index 000000000..51e27c6ba
--- /dev/null
+++ b/app/artifact-cas/internal/service/staging_test.go
@@ -0,0 +1,58 @@
+//
+// Copyright 2026 The Chainloop Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package service
+
+import (
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/chainloop-dev/chainloop/pkg/servicelogger"
+	"github.com/stretchr/testify/require"
+)
+
+// TestSweepStagingDir: a boot-time sweep removes leftover upload temp files from
+// a previous crash while preserving unrelated files and the directory itself.
+func TestSweepStagingDir(t *testing.T) {
+	dir := t.TempDir()
+
+	writeFile := func(name string) string {
+		p := filepath.Join(dir, name)
+		require.NoError(t, os.WriteFile(p, []byte("x"), 0o600))
+		return p
+	}
+
+	leftover1 := writeFile(stagingFilePrefix + "abc123")
+	leftover2 := writeFile(stagingFilePrefix + "def456")
+	unrelated := writeFile("some-other-file.txt")
+
+	removed, err := SweepStagingDir(dir, servicelogger.EmptyLogger())
+	require.NoError(t, err)
+	require.Equal(t, 2, removed, "both leftover upload temp files must be removed")
+
+	require.NoFileExists(t, leftover1)
+	require.NoFileExists(t, leftover2)
+	require.FileExists(t, unrelated, "unrelated files must be preserved")
+	require.DirExists(t, dir, "the staging directory itself must be kept")
+}
+
+// TestSweepStagingDirMissing: sweeping a non-existent directory is not an error
+// (the volume may not have been populated yet).
+func TestSweepStagingDirMissing(t *testing.T) {
+	removed, err := SweepStagingDir(filepath.Join(t.TempDir(), "does-not-exist"), servicelogger.EmptyLogger())
+	require.NoError(t, err)
+	require.Zero(t, removed)
+}
diff --git a/deployment/chainloop/Chart.yaml b/deployment/chainloop/Chart.yaml
index c274c70ba..fb3225d20 100644
--- a/deployment/chainloop/Chart.yaml
+++ b/deployment/chainloop/Chart.yaml
@@ -7,7 +7,7 @@ description: Chainloop is an open source software supply chain control plane, a
 
 type: application
 # Bump the patch (not minor, not major) version on each change in the Chart Source code
-version: 1.427.0
+version: 1.427.1
 # Do not update appVersion, this is handled automatically by the release process
 appVersion: v1.106.1
 
diff --git a/deployment/chainloop/README.md b/deployment/chainloop/README.md
index 976984545..666ca7f17 100644
--- a/deployment/chainloop/README.md
+++ b/deployment/chainloop/README.md
@@ -882,6 +882,9 @@ Once done, you can access with [two predefined users](https://github.com/chainlo
 | `cas.containerSecurityContext.allowPrivilegeEscalation` | Set allowPrivilegeEscalation in cas container' Security Context                                                                                                                                                                                     | `false`          |
 | `cas.containerSecurityContext.capabilities.drop`        | List of capabilities to be dropped in cas container                                                                                                                                                                                                 | `["ALL"]`        |
 | `cas.containerSecurityContext.seccompProfile.type`      | Set seccomp profile in cas container                                                                                                                                                                                                                | `RuntimeDefault` |
+| `cas.staging.enabled`                                   | Mount a dedicated emptyDir for upload staging and point staging_dir at it. Required when readOnlyRootFilesystem is true.                                                                                                                              | `true`           |
+| `cas.staging.mountPath`                                 | Directory where uploads are staged and verified. Must not be /tmp (used by the jwt-public-key secret mount).                                                                                                                                         | `/staging`       |
+| `cas.staging.sizeLimit`                                 | Size limit for the staging emptyDir. Budget roughly (concurrent uploads + downloads) × max artifact size per replica. NOTE: a breach triggers kubelet POD EVICTION, not a clean error, so keep this generous and rely on per-request size caps.      | `10Gi`           |
 | `cas.automountServiceAccountToken`                      | Mount Service Account token in cas pods                                                                                                                                                                                                             | `false`          |
 | `cas.hostAliases`                                       | cas pods host aliases                                                                                                                                                                                                                               | `[]`             |
 | `cas.deploymentAnnotations`                             | Annotations for cas deployment                                                                                                                                                                                                                      | `{}`             |
diff --git a/deployment/chainloop/templates/cas/configmap.yaml b/deployment/chainloop/templates/cas/configmap.yaml
index ae17ba695..8d20902b7 100644
--- a/deployment/chainloop/templates/cas/configmap.yaml
+++ b/deployment/chainloop/templates/cas/configmap.yaml
@@ -14,6 +14,11 @@ metadata:
   {{- end }}
 data:
   server.yaml: |
+    {{- if .Values.cas.staging.enabled }}
+    # Local directory where uploads are staged and verified against the declared
+    # digest before being sent to the backend (see cas.staging in values.yaml).
+    staging_dir: {{ .Values.cas.staging.mountPath | quote }}
+    {{- end }}
     server:
       http:
         addr: "0.0.0.0:{{ .Values.cas.containerPorts.http }}"
diff --git a/deployment/chainloop/templates/cas/deployment.yaml b/deployment/chainloop/templates/cas/deployment.yaml
index 96a57e2de..32f20f9e3 100644
--- a/deployment/chainloop/templates/cas/deployment.yaml
+++ b/deployment/chainloop/templates/cas/deployment.yaml
@@ -126,6 +126,12 @@ spec:
               mountPath: "/data/conf"
             - name: jwt-public-key
               mountPath: "/tmp"
+            {{- if .Values.cas.staging.enabled }}
+            # Writable scratch volume for staging + verifying uploads before they
+            # reach the backend (the container root filesystem is read-only).
+            - name: staging
+              mountPath: {{ .Values.cas.staging.mountPath | quote }}
+            {{- end }}
             {{- if eq "gcpSecretManager" .Values.secretsBackend.backend  }}
             - name: gcp-secretmanager-serviceaccountkey
               mountPath: /gcp-secrets
@@ -158,6 +164,12 @@ spec:
         - name: jwt-public-key
           secret:
             secretName: {{ include "chainloop.cas.fullname" . }}-jwt-public-key
+        {{- if .Values.cas.staging.enabled }}
+        # Node-disk (NOT tmpfs/RAM) scratch space for staging + verifying uploads.
+        - name: staging
+          emptyDir:
+            sizeLimit: {{ .Values.cas.staging.sizeLimit }}
+        {{- end }}
         {{- if include "cas.tls-secret-name" .  }}
         - name: server-certs
           secret:
diff --git a/deployment/chainloop/values.yaml b/deployment/chainloop/values.yaml
index 7a9eaab52..cf3b80353 100644
--- a/deployment/chainloop/values.yaml
+++ b/deployment/chainloop/values.yaml
@@ -1473,7 +1473,34 @@ cas:
       drop: ["ALL"]
     seccompProfile:
       type: "RuntimeDefault"
-  
+
+  ## CAS upload staging volume.
+  ## Uploads are streamed to this directory on local disk and verified against
+  ## the declared digest before being sent to the storage backend; nothing
+  ## unverified ever reaches the backend. The container root filesystem is
+  ## read-only, so a writable emptyDir is mounted here.
+  ##
+  ## Optional: when disabled, the CAS falls back to the OS temporary directory.
+  ## Keep it enabled while readOnlyRootFilesystem is true, since neither the root
+  ## filesystem nor the /tmp secret mount are writable in that case.
+  ##
+  ## WARNING: this MUST be a per-pod volume (an emptyDir, as configured below).
+  ## On startup each CAS pod sweeps leftover staging files from its own volume.
+  ## The sweep assumes exclusive, per-pod ownership: pointing it at a volume
+  ## SHARED between pods (a PVC or hostPath) would let one pod's boot sweep delete
+  ## another pod's in-flight upload. Do not use shared storage here.
+  ## @param cas.staging.enabled Mount a dedicated emptyDir for upload staging and point staging_dir at it. Required when readOnlyRootFilesystem is true.
+  ## @param cas.staging.mountPath Directory where uploads are staged and verified. Must not be /tmp (used by the jwt-public-key secret mount).
+  ## @param cas.staging.sizeLimit Size limit for the staging emptyDir. Budget roughly (concurrent uploads) × max artifact size per replica. NOTE: a breach triggers kubelet POD EVICTION, not a clean error, so keep this generous and rely on per-request size caps as the real bound.
+  ##
+  staging:
+    enabled: true
+    mountPath: /staging
+    ## IMPORTANT: this is a node-disk emptyDir. Do NOT set `medium: Memory` — a
+    ## tmpfs volume is backed by RAM and would reintroduce the OOM that staging
+    ## to disk is meant to avoid.
+    sizeLimit: 10Gi
+
   ## @param cas.automountServiceAccountToken Mount Service Account token in cas pods
   ##
   automountServiceAccountToken: false
diff --git a/pkg/blobmanager/azureblob/backend.go b/pkg/blobmanager/azureblob/backend.go
index d70893553..edc3f72b4 100644
--- a/pkg/blobmanager/azureblob/backend.go
+++ b/pkg/blobmanager/azureblob/backend.go
@@ -20,6 +20,7 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"os"
 
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore/to"
 	"github.com/Azure/azure-sdk-for-go/sdk/azidentity"
@@ -37,15 +38,7 @@ type Backend struct {
 	endpoint           string
 }
 
-var (
-	_ backend.UploaderDownloader = (*Backend)(nil)
-	_ backend.StreamingUploader  = (*Backend)(nil)
-)
-
-// SupportsStreaming reports that the azureblob backend can upload directly from
-// a streaming reader. Upload uses the SDK's UploadStream, which reads the artifact
-// in bounded-size blocks, so CAS never needs to buffer the whole blob in memory.
-func (b *Backend) SupportsStreaming() bool { return true }
+var _ backend.UploaderDownloader = (*Backend)(nil)
 
 func NewBackend(creds *Credentials) (*Backend, error) {
 	credential, err := azidentity.NewClientSecretCredential(creds.TenantID, creds.ClientID, creds.ClientSecret, nil)
@@ -128,11 +121,25 @@ func (b *Backend) Upload(ctx context.Context, r io.Reader, resource *pb.CASResou
 		return fmt.Errorf("failed to create Blob storage Container: %w", err)
 	}
 
+	metadata := map[string]*string{
+		annotationNameAuthor:   to.Ptr(backend.AuthorAnnotation),
+		annotationNameFilename: to.Ptr(resource.FileName),
+	}
+
+	// The CAS service stages verified content on local disk and hands us an
+	// *os.File. UploadFile reads it via ReadAt in bounded blocks, avoiding the
+	// intermediate block buffering that UploadStream needs for a non-seekable
+	// reader. Fall back to UploadStream for any other reader (still bounded by
+	// block size × concurrency).
+	if f, ok := r.(*os.File); ok {
+		_, err = client.UploadFile(ctx, b.container, resourceName(resource.Digest), f, &azblob.UploadFileOptions{
+			Metadata: metadata,
+		})
+		return err
+	}
+
 	_, err = client.UploadStream(ctx, b.container, resourceName(resource.Digest), r, &azblob.UploadStreamOptions{
-		Metadata: map[string]*string{
-			annotationNameAuthor:   to.Ptr(backend.AuthorAnnotation),
-			annotationNameFilename: to.Ptr(resource.FileName),
-		},
+		Metadata: metadata,
 	})
 
 	return err
diff --git a/pkg/blobmanager/azureblob/backend_test.go b/pkg/blobmanager/azureblob/backend_test.go
deleted file mode 100644
index f3a17b20d..000000000
--- a/pkg/blobmanager/azureblob/backend_test.go
+++ /dev/null
@@ -1,34 +0,0 @@
-//
-// Copyright 2026 The Chainloop Authors.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package azureblob
-
-import (
-	"testing"
-
-	backend "github.com/chainloop-dev/chainloop/pkg/blobmanager"
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-)
-
-// TestBackend_SupportsStreaming asserts the azureblob backend opts into
-// streaming uploads so the CAS service feeds it directly from the client stream
-// instead of buffering the whole artifact in memory (PFM-6923).
-func TestBackend_SupportsStreaming(t *testing.T) {
-	var b backend.UploaderDownloader = &Backend{}
-	su, ok := b.(backend.StreamingUploader)
-	require.True(t, ok, "azureblob backend must implement backend.StreamingUploader")
-	assert.True(t, su.SupportsStreaming())
-}
diff --git a/pkg/blobmanager/backend.go b/pkg/blobmanager/backend.go
index c4c4f355a..03c868b04 100644
--- a/pkg/blobmanager/backend.go
+++ b/pkg/blobmanager/backend.go
@@ -44,22 +44,6 @@ type UploaderDownloader interface {
 	Describer
 }
 
-// StreamingUploader is an optional interface implemented by backends whose
-// Upload can consume the artifact directly from a streaming io.Reader without
-// requiring the whole blob to be buffered in memory first.
-//
-// The Artifact CAS service type-asserts uploaders against this interface: when
-// a backend reports SupportsStreaming()==true the upload is piped straight from
-// the client stream to the backend, bounding CAS memory usage independently of
-// artifact size (PFM-6923). Backends that do not implement it (e.g. the OCI
-// backend, whose push path needs the full layer content up front) keep the
-// buffered code path.
-type StreamingUploader interface {
-	// SupportsStreaming reports whether Upload can be fed a streaming reader
-	// without the caller buffering the full artifact in memory first.
-	SupportsStreaming() bool
-}
-
 type Describer interface {
 	Describe(ctx context.Context, digest string) (*v1.CASResource, error)
 }
diff --git a/pkg/blobmanager/oci/backend_test.go b/pkg/blobmanager/oci/backend_test.go
index c0d47638d..4b509b30d 100644
--- a/pkg/blobmanager/oci/backend_test.go
+++ b/pkg/blobmanager/oci/backend_test.go
@@ -27,7 +27,6 @@ import (
 	"testing"
 
 	pb "github.com/chainloop-dev/chainloop/app/artifact-cas/api/cas/v1"
-	backend "github.com/chainloop-dev/chainloop/pkg/blobmanager"
 	"github.com/google/go-containerregistry/pkg/authn"
 	"github.com/google/go-containerregistry/pkg/name"
 	"github.com/google/go-containerregistry/pkg/registry"
@@ -409,14 +408,3 @@ func (s *testSuite) TearDownTest() {
 func TestOCIBackend(t *testing.T) {
 	suite.Run(t, new(testSuite))
 }
-
-// TestBackend_DoesNotSupportStreaming pins the OCI backend as NON-streaming.
-// go-containerregistry's push path needs the whole layer content in memory up
-// front (see Backend.Upload), so the CAS service must keep buffering OCI
-// uploads. If OCI ever grows a SupportsStreaming method it would be fed a
-// streaming reader and silently break; this test fails closed against that.
-func TestBackend_DoesNotSupportStreaming(t *testing.T) {
-	var b backend.UploaderDownloader = &Backend{}
-	_, ok := b.(backend.StreamingUploader)
-	require.False(t, ok, "oci backend must NOT implement backend.StreamingUploader; it requires full in-memory buffering")
-}
diff --git a/pkg/blobmanager/s3/backend.go b/pkg/blobmanager/s3/backend.go
index 8fdc9546a..49d4782e3 100644
--- a/pkg/blobmanager/s3/backend.go
+++ b/pkg/blobmanager/s3/backend.go
@@ -46,19 +46,10 @@ type Backend struct {
 	customEndpoint string
 }
 
-var (
-	_ backend.UploaderDownloader = (*Backend)(nil)
-	_ backend.StreamingUploader  = (*Backend)(nil)
-)
+var _ backend.UploaderDownloader = (*Backend)(nil)
 
 const defaultRegion = "us-east-1"
 
-// SupportsStreaming reports that the s3 backend can upload directly from a
-// streaming reader. The AWS SDK's manager.Uploader consumes the reader in
-// bounded-size parts (multipart upload), so CAS never needs to buffer the whole
-// artifact in memory.
-func (b *Backend) SupportsStreaming() bool { return true }
-
 func NewBackend(creds *Credentials) (*Backend, error) {
 	if creds == nil {
 		return nil, errors.New("credentials cannot be nil")
@@ -173,10 +164,12 @@ func (b *Backend) Upload(ctx context.Context, r io.Reader, resource *pb.CASResou
 		},
 	}
 
-	// if b.checksumVerificationEnabled() {
-	// 	// Check that the object is uploaded correctly
-	// 	input.ChecksumSHA256 = aws.String(hexSha256ToBinaryB64(resource.Digest))
-	// }
+	// No per-object ChecksumSHA256 precondition is set here: a whole-object SHA256
+	// precondition cannot be expressed for multipart uploads (S3 only supports
+	// FULL_OBJECT checksums for CRC variants) and is unsupported on some
+	// S3-compatible endpoints (e.g. R2). Upload integrity is guaranteed by the CAS
+	// service, which hashes the content and verifies it against the declared
+	// digest before calling Upload.
 
 	if _, err := uploader.Upload(ctx, input); err != nil {
 		return fmt.Errorf("failed to upload to bucket: %w", err)
diff --git a/pkg/blobmanager/s3/backend_test.go b/pkg/blobmanager/s3/backend_test.go
index 91ee4c6da..6957eb62b 100644
--- a/pkg/blobmanager/s3/backend_test.go
+++ b/pkg/blobmanager/s3/backend_test.go
@@ -244,12 +244,6 @@ func (s *testSuite) TestDownload() {
 		s.NoError(err)
 		s.Equal("test", buf.String())
 	})
-
-	// s.T().Run("it's been tampered", func(t *testing.T) {
-	// 	buf := bytes.NewBuffer(nil)
-	// 	err := s.backend.Download(context.Background(), buf, s.tamperedObjectDigest)
-	// 	s.ErrorContains(err, "failed to validate integrity of object")
-	// })
 }
 
 type testSuite struct {
@@ -258,7 +252,6 @@ type testSuite struct {
 	backend, invalidBackend *Backend
 	ownedObjectDigest       string
 	externalObjectDigest    string
-	tamperedObjectDigest    string
 }
 
 func TestS3Backend(t *testing.T) {
@@ -313,15 +306,6 @@ func (s *testSuite) SetupTest() {
 	err = s.backend.Upload(context.Background(), buf, &pb.CASResource{Digest: s.ownedObjectDigest, FileName: "test.txt"})
 	require.NoError(s.T(), err)
 
-	// Copy an existing object but reference it from somewhere else
-	s.tamperedObjectDigest = "b5bb9d8014a0f9b1d61e21e796d78dccdf1352f23cd32812f4850b878ae4944c"
-	_, err = minioClient.CopyObject(context.Background(), minio.CopyDestOptions{
-		Bucket: testBucket, Object: fmt.Sprintf("sha256:%s", s.tamperedObjectDigest),
-	}, minio.CopySrcOptions{
-		Bucket: testBucket, Object: fmt.Sprintf("sha256:%s", s.ownedObjectDigest),
-	})
-	require.NoError(s.T(), err)
-
 	// upload another one but by the client directly
 	reader := bytes.NewReader([]byte("hello world"))
 	s.externalObjectDigest = "external-deadbeef"
@@ -376,13 +360,3 @@ func (c *minioInstance) ConnectionString(t *testing.T) string {
 type minioInstance struct {
 	instance testcontainers.Container
 }
-
-// TestBackend_SupportsStreaming asserts the s3 backend opts into streaming
-// uploads so the CAS service feeds it directly from the client stream instead
-// of buffering the whole artifact in memory (PFM-6923).
-func TestBackend_SupportsStreaming(t *testing.T) {
-	var b backend.UploaderDownloader = &Backend{}
-	su, ok := b.(backend.StreamingUploader)
-	require.True(t, ok, "s3 backend must implement backend.StreamingUploader")
-	assert.True(t, su.SupportsStreaming())
-}
diff --git a/pkg/blobmanager/s3accesspoint/backend.go b/pkg/blobmanager/s3accesspoint/backend.go
index 71d2524cc..6c30e663a 100644
--- a/pkg/blobmanager/s3accesspoint/backend.go
+++ b/pkg/blobmanager/s3accesspoint/backend.go
@@ -73,16 +73,7 @@ type Backend struct {
 	s3Client *s3.Client
 }
 
-var (
-	_ backend.UploaderDownloader = (*Backend)(nil)
-	_ backend.StreamingUploader  = (*Backend)(nil)
-)
-
-// SupportsStreaming reports that the s3accesspoint backend can upload directly
-// from a streaming reader. Like the plain s3 backend it uses the AWS SDK's
-// manager.Uploader, which consumes the reader in bounded-size multipart parts,
-// so CAS never needs to buffer the whole artifact in memory.
-func (b *Backend) SupportsStreaming() bool { return true }
+var _ backend.UploaderDownloader = (*Backend)(nil)
 
 // NewBackend constructs a *Backend wired to an STS-backed credentials
 // provider. ctx is used only for the initial AWS config load (DNS lookups,
diff --git a/pkg/blobmanager/s3accesspoint/backend_test.go b/pkg/blobmanager/s3accesspoint/backend_test.go
index 1b929bd7e..57cbd34a8 100644
--- a/pkg/blobmanager/s3accesspoint/backend_test.go
+++ b/pkg/blobmanager/s3accesspoint/backend_test.go
@@ -26,7 +26,6 @@ import (
 	ststypes "github.com/aws/aws-sdk-go-v2/service/sts/types"
 	pb "github.com/chainloop-dev/chainloop/app/artifact-cas/api/cas/v1"
 	robotaccount "github.com/chainloop-dev/chainloop/internal/robotaccount/cas"
-	backend "github.com/chainloop-dev/chainloop/pkg/blobmanager"
 	jwtmiddleware "github.com/go-kratos/kratos/v2/middleware/auth/jwt"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -295,13 +294,3 @@ func assertFailedClosed(t *testing.T, err error) {
 	require.Containsf(t, err.Error(), ErrMissingRequestingOrg.Error(),
 		"expected fail-closed missing-org error, got %q", err)
 }
-
-// TestBackend_SupportsStreaming asserts the s3accesspoint backend opts into
-// streaming uploads so the CAS service feeds it directly from the client stream
-// instead of buffering the whole artifact in memory (PFM-6923).
-func TestBackend_SupportsStreaming(t *testing.T) {
-	var b backend.UploaderDownloader = &Backend{}
-	su, ok := b.(backend.StreamingUploader)
-	require.True(t, ok, "s3accesspoint backend must implement backend.StreamingUploader")
-	assert.True(t, su.SupportsStreaming())
-}