diff --git a/app/artifact-cas/internal/service/bytestream.go b/app/artifact-cas/internal/service/bytestream.go
index b533c1cf6..86c588141 100644
--- a/app/artifact-cas/internal/service/bytestream.go
+++ b/app/artifact-cas/internal/service/bytestream.go
@@ -115,29 +115,33 @@ 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)
-	// Create a buffer that will be filled in the background before sending its content to the backend
-	buffer := newStreamReader(info.MaxBytes)
-	// Add data from first request
-	if err = buffer.Write(req.GetData()); err != nil {
-		if backend.IsUploadSizeExceeded(err) {
-			return status.Error(codes.ResourceExhausted, err.Error())
-		}
-		return sl.LogAndMaskErr(err, s.log)
-	}
 
-	// Start a goroutine that will fill the buffer in the background
-	go bufferStream(ctx, stream, buffer, s.log)
+	// 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)
+	}
 
-	// Block until the buffer has been filled or the upload process has been canceled
-	// This implementation is suboptimal since it requires the content to be uploaded in memory before pushing it
-	// This is due to the fact that our OCI push implementation does not support streaming/chunks for uncompressed layers
-	// We can not use stream.Layer since it only supports compressed layers, we want to store raw data and set custom mimetypes
-	// https://github.com/google/go-containerregistry/blob/main/pkg/v1/stream/README.md
-	// TODO: Split content in multiple layers and do concurrent uploads/downloads
-	err = <-buffer.errorChan
-
-	// Now it's time to check if the data provider has sent an error
+	// Classify the outcome. The error may come from two distinct stages, which
+	// must be treated differently:
+	//   - 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
+	//     failure, an exceeded size cap maps to ResourceExhausted, anything else
+	//     is masked.
 	if err != nil {
+		var backendErr *backendUploadError
+		if errors.As(err, &backendErr) {
+			return sl.LogAndMaskErr(backendErr.err, s.log)
+		}
 		if isClientDisconnect(err) {
 			s.log.Infow("msg", "upload canceled", "digest", req.resource.Digest, "name", req.resource.FileName)
 			return nil
@@ -148,22 +152,189 @@ func (s *ByteStreamService) Write(stream bytestream.ByteStream_WriteServer) erro
 		return sl.LogAndMaskErr(err, s.log)
 	}
 
-	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 sl.LogAndMaskErr(err, s.log)
-	}
-
-	s.log.Infow("msg", "upload finished", "name", req.resource.FileName, "digest", req.resource.Digest, "size", buffer.size)
+	s.log.Infow("msg", "upload finished", "name", req.resource.FileName, "digest", req.resource.Digest, "size", committedSize)
 	s.audit.Dispatch(&events.CASArtifactUploaded{
 		CASArtifactBase: &events.CASArtifactBase{
 			Digest:      req.resource.Digest,
-			SizeBytes:   buffer.size,
+			SizeBytes:   committedSize,
 			FileName:    req.resource.FileName,
 			BackendType: info.BackendType,
 		},
 	}, info)
 
-	return stream.SendAndClose(&bytestream.WriteResponse{CommittedSize: buffer.size})
+	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
+	}
+
+	// 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 {
+		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}
+	}
+
+	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
+	}
+
+	// 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}
+	}
+
+	return uploadedSize, nil
+}
+
+// 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
+// a size-cap violation, both of which only originate on the stream-read side.
+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
+}
+
+func (streamingReader) String() string { return "cas-streaming-upload" }
+
+// 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) {
+	var size int64
+	write := func(data []byte) error {
+		if len(data) == 0 {
+			return nil
+		}
+		size += int64(len(data))
+		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
+		}
+		return nil
+	}
+
+	// Forward the data from the first request.
+	if err := write(firstData); err != nil {
+		return size, err
+	}
+
+	for {
+		select {
+		case <-ctx.Done():
+			// DeadlineExceeded, or Canceled
+			return size, ctx.Err()
+		default:
+			// Extract the next chunk of data from the stream request
+			req, err := getWriteRequest(stream)
+			if err != nil {
+				// Finished reading the stream is not a real error
+				if errors.Is(err, io.EOF) {
+					return size, nil
+				}
+				return size, err
+			}
+
+			// Forward 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()))
+
+			// Check if the client has finished sending data
+			if req.GetFinishWrite() {
+				return size, nil
+			}
+		}
+	}
 }
 
 // Server-side streaming RPC for reading blobs, implements the bytestream interface
@@ -250,18 +421,20 @@ func bufferStream(ctx context.Context, stream bytestream.ByteStream_WriteServer,
 				return
 			}
 
-			// Check if the client has finished sending data
-			if req.GetFinishWrite() {
-				return
-			}
-
-			// Write the data to the buffer
+			// 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
+			}
 		}
 	}
 }
@@ -290,16 +463,24 @@ func newStreamReader(maxSize int64) *streamReader {
 func (r *streamReader) Write(data []byte) error {
 	r.size += int64(len(data))
 
-	// Check if the size of the buffer has exceeded the maximum allowed size
-	// if maxSize is 0, then there is no limit
-	if r.maxSize != 0 && r.size > r.maxSize {
-		return backend.NewErrUploadSizeExceeded(r.size, r.maxSize)
+	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.
+func checkUploadSize(total, maxSize int64) error {
+	if maxSize != 0 && total > maxSize {
+		return backend.NewErrUploadSizeExceeded(total, maxSize)
+	}
+	return nil
+}
+
 type writeRequest struct {
 	*bytestream.WriteRequest
 	resource *v1.CASResource
diff --git a/app/artifact-cas/internal/service/bytestream_download_test.go b/app/artifact-cas/internal/service/bytestream_download_test.go
new file mode 100644
index 000000000..95abb97ba
--- /dev/null
+++ b/app/artifact-cas/internal/service/bytestream_download_test.go
@@ -0,0 +1,230 @@
+//
+// 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 (
+	"context"
+	"crypto/sha256"
+	"errors"
+	"fmt"
+	"io"
+	"testing"
+
+	"github.com/chainloop-dev/chainloop/pkg/servicelogger"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/genproto/googleapis/bytestream"
+	"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.
+
+// fakeReadServer is a minimal bytestream.ByteStream_ReadServer that records the
+// data chunks the streamWriter sends. Only Send is exercised by streamWriter.
+type fakeReadServer struct {
+	bytestream.ByteStream_ReadServer
+	sent [][]byte
+}
+
+func (f *fakeReadServer) Send(r *bytestream.ReadResponse) error {
+	// copy: the underlying buffer may be reused by the caller
+	cp := make([]byte, len(r.Data))
+	copy(cp, r.Data)
+	f.sent = append(f.sent, cp)
+	return nil
+}
+
+// TestStreamWriter_ChecksumAcrossWrites verifies the download writer computes
+// the sha256 over the full stream regardless of how the content is chunked, and
+// forwards every byte to the client in order.
+func TestStreamWriter_ChecksumAcrossWrites(t *testing.T) {
+	content := []byte("chainloop download payload delivered in several writes")
+	digest := sha256Hex(content)
+
+	fake := &fakeReadServer{}
+	sw := &streamWriter{
+		stream:       fake,
+		log:          servicelogger.EmptyLogger(),
+		wantChecksum: digest,
+		gotChecksum:  sha256.New(),
+	}
+
+	// Feed the content in uneven pieces to mimic a backend emitting arbitrary
+	// chunk boundaries.
+	for off := 0; off < len(content); off += 11 {
+		end := off + 11
+		if end > len(content) {
+			end = len(content)
+		}
+		n, err := sw.Write(content[off:end])
+		require.NoError(t, err)
+		assert.Equal(t, end-off, n)
+	}
+
+	assert.Equal(t, digest, sw.GetChecksum(), "checksum must be computed over the whole stream")
+	assert.Equal(t, int64(len(content)), sw.size)
+
+	// The bytes forwarded to the client must reassemble to the exact content.
+	forwarded := make([]byte, 0, len(content))
+	for _, c := range fake.sent {
+		forwarded = append(forwarded, c...)
+	}
+	assert.Equal(t, content, forwarded)
+}
+
+// TestStreamWriter_EmptyStreamChecksum: with no writes the computed checksum is
+// the sha256 of the empty input.
+func TestStreamWriter_EmptyStreamChecksum(t *testing.T) {
+	sw := &streamWriter{stream: &fakeReadServer{}, log: servicelogger.EmptyLogger(), gotChecksum: sha256.New()}
+	assert.Equal(t, sha256Hex([]byte{}), sw.GetChecksum())
+	assert.Equal(t, int64(0), sw.size)
+}
+
+// recvAllDownload drains a Read stream, returning the concatenated payload and
+// the terminating error (io.EOF becomes nil).
+func recvAllDownload(reader bytestream.ByteStream_ReadClient) ([]byte, error) {
+	var buf []byte
+	for {
+		resp, err := reader.Recv()
+		if resp != nil {
+			buf = append(buf, resp.Data...)
+		}
+		if errors.Is(err, io.EOF) {
+			return buf, nil
+		}
+		if err != nil {
+			return buf, err
+		}
+	}
+}
+
+// TestDownloadMultiChunkChecksumOK: content the backend emits across several
+// writes is streamed back intact and passes digest verification.
+func (s *bytestreamSuite) TestDownloadMultiChunkChecksumOK() {
+	content := deterministicBytes(200 * 1024)
+	digest := sha256Hex(content)
+
+	s.ociBackend.On("Download", mock.Anything, mock.Anything, digest).Return(nil).
+		Run(func(args mock.Arguments) {
+			w := args.Get(1).(io.Writer)
+			for off := 0; off < len(content); off += 8192 {
+				end := off + 8192
+				if end > len(content) {
+					end = len(content)
+				}
+				_, err := w.Write(content[off:end])
+				s.NoError(err)
+			}
+		})
+
+	reader, err := s.client.Read(s.downCtx, &bytestream.ReadRequest{ResourceName: digest})
+	s.NoError(err)
+
+	got, err := recvAllDownload(reader)
+	s.NoError(err)
+	s.Equal(content, got)
+	s.Equal(digest, sha256Hex(got))
+
+	s.Require().Len(s.audit.published, 1)
+	info := decodeArtifactEvent(s.T(), s.audit.published[0])
+	s.Equal(digest, info.Digest)
+	s.Equal(int64(len(content)), info.SizeBytes)
+	s.False(info.Skipped)
+}
+
+// TestDownloadEmptyContentChecksum: a zero-byte artifact whose requested digest
+// is sha256("") verifies and streams back empty.
+func (s *bytestreamSuite) TestDownloadEmptyContentChecksum() {
+	digest := sha256Hex([]byte{})
+
+	s.ociBackend.On("Download", mock.Anything, mock.Anything, digest).Return(nil).
+		Run(func(_ mock.Arguments) {
+			// write nothing
+		})
+
+	reader, err := s.client.Read(s.downCtx, &bytestream.ReadRequest{ResourceName: digest})
+	s.NoError(err)
+
+	got, err := recvAllDownload(reader)
+	s.NoError(err)
+	s.Empty(got)
+
+	s.Require().Len(s.audit.published, 1)
+	info := decodeArtifactEvent(s.T(), s.audit.published[0])
+	s.Equal(int64(0), info.SizeBytes)
+}
+
+// TestDownloadTamperedAcrossChunksRejected: if the streamed bytes do not hash to
+// the requested digest, the server reports a checksum mismatch and emits no
+// audit event — the tamper-detection guarantee of a CAS.
+func (s *bytestreamSuite) TestDownloadTamperedAcrossChunksRejected() {
+	// The client asks for this digest, but the backend returns different bytes.
+	requested := sha256Hex([]byte("the authentic artifact contents"))
+	tampered := []byte("tampered artifact contents split across chunks!!")
+
+	s.ociBackend.On("Download", mock.Anything, mock.Anything, requested).Return(nil).
+		Run(func(args mock.Arguments) {
+			w := args.Get(1).(io.Writer)
+			_, _ = w.Write(tampered[:20])
+			_, _ = w.Write(tampered[20:])
+		})
+
+	reader, err := s.client.Read(s.downCtx, &bytestream.ReadRequest{ResourceName: requested})
+	s.NoError(err)
+
+	_, err = recvAllDownload(reader)
+	s.ErrorContains(err, "checksum mismatch")
+	// tampered downloads emit no events
+	s.Empty(s.audit.published)
+}
+
+// TestDownloadClientDisconnect: a backend Download failure that indicates the
+// client went away is treated as a cancellation (no error to a gone client, no
+// audit). This is pre-existing download behavior; the test pins it.
+func (s *bytestreamSuite) TestDownloadClientDisconnect() {
+	const digest = "deadbeef"
+	s.ociBackend.On("Download", mock.Anything, mock.Anything, digest).
+		Return(fmt.Errorf("stream send failed: %w", context.Canceled))
+
+	reader, err := s.client.Read(s.downCtx, &bytestream.ReadRequest{ResourceName: digest})
+	s.NoError(err)
+
+	// Server returns nil (treats as disconnect); the client sees a clean EOF.
+	_, err = recvAllDownload(reader)
+	s.NoError(err)
+	s.Empty(s.audit.published)
+}
+
+// TestDownloadBackendErrorMasked: a genuine backend Download failure is masked
+// as Internal and emits no audit event.
+func (s *bytestreamSuite) TestDownloadBackendErrorMasked() {
+	const digest = "deadbeef"
+	s.ociBackend.On("Download", mock.Anything, mock.Anything, digest).
+		Return(fmt.Errorf("object store unreachable"))
+
+	reader, err := s.client.Read(s.downCtx, &bytestream.ReadRequest{ResourceName: digest})
+	s.NoError(err)
+
+	_, err = recvAllDownload(reader)
+	assertGRPCError(s.T(), err, codes.Internal, "server error")
+	s.Empty(s.audit.published)
+}
diff --git a/app/artifact-cas/internal/service/bytestream_streaming_test.go b/app/artifact-cas/internal/service/bytestream_streaming_test.go
new file mode 100644
index 000000000..777790e17
--- /dev/null
+++ b/app/artifact-cas/internal/service/bytestream_streaming_test.go
@@ -0,0 +1,531 @@
+//
+// 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 (
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"fmt"
+	"io"
+	"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"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/metadata"
+)
+
+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.
+// maxBytes == "" leaves the per-upload cap unset (unlimited).
+func streamingUpCtx(maxBytes string) context.Context {
+	pairs := []string{"role", "uploader", "backend-streaming", "true"}
+	if maxBytes != "" {
+		pairs = append(pairs, "max-bytes", maxBytes)
+	}
+	return metadata.NewOutgoingContext(context.Background(), metadata.Pairs(pairs...))
+}
+
+func sha256Hex(b []byte) string {
+	sum := sha256.Sum256(b)
+	return hex.EncodeToString(sum[:])
+}
+
+// resourceWithDigest builds a CASResource whose Digest is the real sha256 of
+// content, so integrity assertions can compare the bytes the backend received
+// against the declared digest.
+func resourceWithDigest(content []byte, fileName string) *v1.CASResource {
+	return &v1.CASResource{Digest: sha256Hex(content), FileName: fileName}
+}
+
+// deterministicBytes returns n reproducible, non-trivial bytes (a rolling
+// pattern) so multi-chunk integrity tests exercise real content, not zeros.
+func deterministicBytes(n int) []byte {
+	b := make([]byte, n)
+	for i := range b {
+		b[i] = byte((i*31 + 7) % 251)
+	}
+	return b
+}
+
+// sendInChunks writes content to the stream in chunkSize slices, re-sending the
+// resource name on every message (as the Chainloop CLI does). A zero-length
+// content still sends one message carrying the resource name.
+func sendInChunks(t *testing.T, stream bytestream.ByteStream_WriteClient, resourceB64 string, content []byte, chunkSize int) {
+	t.Helper()
+	if len(content) == 0 {
+		require.NoError(t, stream.Send(&bytestream.WriteRequest{ResourceName: resourceB64}))
+		return
+	}
+	for off := 0; off < len(content); off += chunkSize {
+		end := off + chunkSize
+		if end > len(content) {
+			end = len(content)
+		}
+		require.NoError(t, stream.Send(&bytestream.WriteRequest{
+			ResourceName: resourceB64,
+			Data:         content[off:end],
+		}))
+	}
+}
+
+// expectStreamingUpload stubs Exists=false and an Upload that fully drains the
+// reader, returning the captured bytes on the returned channel and reporting
+// uploadErr to the client.
+func (s *bytestreamSuite) expectStreamingUpload(resource *v1.CASResource, uploadErr error) <-chan []byte {
+	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(uploadErr).Run(func(args mock.Arguments) {
+		got, err := io.ReadAll(args.Get(1).(io.Reader))
+		s.NoError(err)
+		received <- got
+	})
+	return received
+}
+
+// --- integrity / normal cases --------------------------------------------
+
+// TestWriteStreamingSingleChunkOK: a single-chunk streaming upload stores the
+// exact bytes, reports the committed size, and emits an audit event.
+func (s *bytestreamSuite) TestWriteStreamingSingleChunkOK() {
+	content := []byte("hello streaming world")
+	resource := resourceWithDigest(content, "artifact.bin")
+	received := s.expectStreamingUpload(resource, nil)
+
+	stream, err := s.client.Write(streamingUpCtx(""))
+	s.NoError(err)
+	sendInChunks(s.T(), stream, encodeResource(s.T(), resource), content, len(content))
+
+	got, err := stream.CloseAndRecv()
+	s.NoError(err)
+	s.Equal(int64(len(content)), got.CommittedSize)
+	s.Equal(content, <-received)
+
+	s.Require().Len(s.audit.published, 1)
+	info := decodeArtifactEvent(s.T(), s.audit.published[0])
+	s.False(info.Skipped)
+	s.Equal(resource.Digest, info.Digest)
+	s.Equal(int64(len(content)), info.SizeBytes)
+}
+
+// TestWriteStreamingMultiChunkIntegrity: content split across several small
+// chunks is reassembled byte-for-byte, and the sha256 of what the backend
+// received matches the declared digest — no corruption/reordering/truncation.
+func (s *bytestreamSuite) TestWriteStreamingMultiChunkIntegrity() {
+	content := []byte("the quick brown fox jumps over the lazy dog, twice over for good measure")
+	resource := resourceWithDigest(content, "artifact.bin")
+	received := s.expectStreamingUpload(resource, nil)
+
+	stream, err := s.client.Write(streamingUpCtx(""))
+	s.NoError(err)
+	sendInChunks(s.T(), stream, encodeResource(s.T(), resource), content, 7)
+
+	got, err := stream.CloseAndRecv()
+	s.NoError(err)
+	s.Equal(int64(len(content)), got.CommittedSize)
+
+	gotBytes := <-received
+	s.Equal(content, gotBytes)
+	s.Equal(resource.Digest, sha256Hex(gotBytes), "backend content digest must match the declared digest")
+}
+
+// TestWriteStreamingManyChunksIntegrity exercises the pipe under many
+// iterations with a larger payload and verifies integrity end-to-end.
+func (s *bytestreamSuite) TestWriteStreamingManyChunksIntegrity() {
+	content := deterministicBytes(512 * 1024) // 512 KiB
+	resource := resourceWithDigest(content, "big.bin")
+	received := s.expectStreamingUpload(resource, nil)
+
+	stream, err := s.client.Write(streamingUpCtx(""))
+	s.NoError(err)
+	sendInChunks(s.T(), stream, encodeResource(s.T(), resource), content, 4096)
+
+	got, err := stream.CloseAndRecv()
+	s.NoError(err)
+	s.Equal(int64(len(content)), got.CommittedSize)
+
+	gotBytes := <-received
+	s.Equal(resource.Digest, sha256Hex(gotBytes))
+	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() {
+	content := []byte{}
+	resource := resourceWithDigest(content, "empty.bin")
+	received := s.expectStreamingUpload(resource, nil)
+
+	stream, err := s.client.Write(streamingUpCtx(""))
+	s.NoError(err)
+	sendInChunks(s.T(), stream, encodeResource(s.T(), resource), content, 1024)
+
+	got, err := stream.CloseAndRecv()
+	s.NoError(err)
+	s.Equal(int64(0), got.CommittedSize)
+	s.Empty(<-received)
+
+	s.Require().Len(s.audit.published, 1)
+	info := decodeArtifactEvent(s.T(), s.audit.published[0])
+	s.Equal(int64(0), info.SizeBytes)
+}
+
+// TestWriteStreamingFinishWriteWithData: a spec-compliant client may set
+// finish_write=true on the final message while it still carries Data; that
+// trailing chunk must be stored, not dropped/truncated (google.bytestream
+// allows the terminal message to carry payload).
+func (s *bytestreamSuite) TestWriteStreamingFinishWriteWithData() {
+	content := []byte("hello world")
+	resource := resourceWithDigest(content, "artifact.bin")
+	received := s.expectStreamingUpload(resource, nil)
+
+	stream, err := s.client.Write(streamingUpCtx(""))
+	s.NoError(err)
+	rb := encodeResource(s.T(), resource)
+	// First chunk with no finish, then the final chunk carrying data AND finish.
+	s.NoError(stream.Send(&bytestream.WriteRequest{ResourceName: rb, Data: content[:5]}))
+	s.NoError(stream.Send(&bytestream.WriteRequest{ResourceName: rb, Data: content[5:], FinishWrite: true}))
+
+	got, err := stream.CloseAndRecv()
+	s.NoError(err)
+	s.Equal(int64(len(content)), got.CommittedSize)
+
+	gotBytes := <-received
+	s.Equal(content, gotBytes)
+	s.Equal(resource.Digest, sha256Hex(gotBytes), "the finish_write chunk's data must be stored, not dropped")
+}
+
+// --- size-cap boundaries --------------------------------------------------
+
+// TestWriteStreamingMaxSizeExceededMultiChunk: the cap trips across chunks and
+// surfaces as ResourceExhausted, emitting no audit event.
+func (s *bytestreamSuite) TestWriteStreamingMaxSizeExceededMultiChunk() {
+	s.streamingBackend.On("Exists", mock.Anything, s.resource.Digest).Return(false, nil)
+	// The backend may or may not observe the truncated stream; drain if called.
+	s.streamingBackend.On("Upload", mock.Anything, mock.Anything, s.resource).Maybe().
+		Return(nil).Run(func(args mock.Arguments) {
+		_, _ = io.ReadAll(args.Get(1).(io.Reader))
+	})
+
+	stream, err := s.client.Write(streamingUpCtx("8"))
+	s.NoError(err)
+	s.NoError(stream.Send(&bytestream.WriteRequest{
+		ResourceName: encodeResource(s.T(), s.resource),
+		Data:         []byte("hello"),
+	}))
+	s.NoError(stream.Send(&bytestream.WriteRequest{
+		ResourceName: encodeResource(s.T(), s.resource),
+		Data:         []byte("chainloop"),
+	}))
+
+	_, err = stream.CloseAndRecv()
+	assertGRPCError(s.T(), err, codes.ResourceExhausted, "max size of upload exceeded")
+	s.Empty(s.audit.published)
+}
+
+// TestWriteStreamingMaxSizeExceededFirstChunk: a first chunk that alone exceeds
+// the cap is rejected with ResourceExhausted.
+func (s *bytestreamSuite) TestWriteStreamingMaxSizeExceededFirstChunk() {
+	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) {
+		_, _ = io.ReadAll(args.Get(1).(io.Reader))
+	})
+
+	stream, err := s.client.Write(streamingUpCtx("4"))
+	s.NoError(err)
+	s.NoError(stream.Send(&bytestream.WriteRequest{
+		ResourceName: encodeResource(s.T(), s.resource),
+		Data:         []byte("too-long"),
+	}))
+
+	_, err = stream.CloseAndRecv()
+	assertGRPCError(s.T(), err, codes.ResourceExhausted, "max size of upload exceeded")
+	s.Empty(s.audit.published)
+}
+
+// TestWriteStreamingMaxSizeBoundaryExact: content exactly equal to the cap is
+// accepted (the cap is exceeded only when size > max).
+func (s *bytestreamSuite) TestWriteStreamingMaxSizeBoundaryExact() {
+	content := []byte("12345678") // 8 bytes
+	resource := resourceWithDigest(content, "artifact.bin")
+	received := s.expectStreamingUpload(resource, nil)
+
+	stream, err := s.client.Write(streamingUpCtx("8"))
+	s.NoError(err)
+	sendInChunks(s.T(), stream, encodeResource(s.T(), resource), content, 3)
+
+	got, err := stream.CloseAndRecv()
+	s.NoError(err)
+	s.Equal(int64(8), got.CommittedSize)
+	s.Equal(content, <-received)
+}
+
+// TestWriteStreamingMaxSizeUnlimited: max-bytes==0 imposes no cap.
+func (s *bytestreamSuite) TestWriteStreamingMaxSizeUnlimited() {
+	content := deterministicBytes(200 * 1024)
+	resource := resourceWithDigest(content, "artifact.bin")
+	received := s.expectStreamingUpload(resource, nil)
+
+	// note: streamingUpCtx("") leaves max-bytes unset → claims.MaxBytes == 0
+	stream, err := s.client.Write(streamingUpCtx(""))
+	s.NoError(err)
+	sendInChunks(s.T(), stream, encodeResource(s.T(), resource), content, 8192)
+
+	got, err := stream.CloseAndRecv()
+	s.NoError(err)
+	s.Equal(int64(len(content)), got.CommittedSize)
+	s.Equal(resource.Digest, sha256Hex(<-received))
+}
+
+// --- error / disconnect classification ------------------------------------
+
+// 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).
+		Return(fmt.Errorf("object store rejected the upload")).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"),
+	}))
+
+	_, err = stream.CloseAndRecv()
+	assertGRPCError(s.T(), err, codes.Internal, "server error")
+	s.Empty(s.audit.published)
+}
+
+// TestWriteStreamingBackendResetNotTreatedAsDisconnect: a backend-side failure
+// 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).
+		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"),
+	}))
+
+	_, err = stream.CloseAndRecv()
+	// "server error" is the masked-error message; the buggy disconnect path
+	// would instead return nil and surface a gRPC "cardinality violation".
+	assertGRPCError(s.T(), err, codes.Internal, "server error")
+	s.Empty(s.audit.published)
+}
+
+// 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).
+		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"),
+	}))
+
+	_, err = stream.CloseAndRecv()
+	assertGRPCError(s.T(), err, codes.Internal, "server error")
+	s.Empty(s.audit.published)
+}
+
+// TestWriteStreamingBackendSuccessWithoutDrain: a backend that returns success
+// without reading the reader to EOF must be reported as a successful upload; the
+// 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
+
+	stream, err := s.client.Write(streamingUpCtx(""))
+	s.NoError(err)
+	s.NoError(stream.Send(&bytestream.WriteRequest{
+		ResourceName: encodeResource(s.T(), s.resource),
+		Data:         data,
+	}))
+
+	got, err := stream.CloseAndRecv()
+	s.NoError(err)
+	s.Equal(int64(len(data)), got.CommittedSize)
+
+	s.Require().Len(s.audit.published, 1)
+	info := decodeArtifactEvent(s.T(), s.audit.published[0])
+	s.False(info.Skipped)
+}
+
+// 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.
+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
+	})
+
+	ctx, cancel := context.WithCancel(streamingUpCtx(""))
+	stream, err := s.client.Write(ctx)
+	s.NoError(err)
+	s.NoError(stream.Send(&bytestream.WriteRequest{
+		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.
+	s.Empty(s.audit.published)
+}
+
+// --- dedup ----------------------------------------------------------------
+
+// TestWriteStreamingExist: an already-present artifact is not re-uploaded.
+func (s *bytestreamSuite) TestWriteStreamingExist() {
+	s.streamingBackend.On("Exists", mock.Anything, s.resource.Digest).Return(true, nil)
+	s.streamingBackend.On("Describe", mock.Anything, s.resource.Digest).Return(&v1.CASResource{
+		FileName: s.resource.FileName, Digest: s.resource.Digest, Size: 2048,
+	}, nil)
+
+	stream, err := s.client.Write(streamingUpCtx(""))
+	s.NoError(err)
+	s.NoError(stream.Send(&bytestream.WriteRequest{
+		ResourceName: encodeResource(s.T(), s.resource),
+	}))
+
+	got, err := stream.CloseAndRecv()
+	s.NoError(err)
+	s.Equal(int64(0), got.CommittedSize)
+	s.streamingBackend.AssertNotCalled(s.T(), "Upload", mock.Anything, mock.Anything, mock.Anything)
+
+	s.Require().Len(s.audit.published, 1)
+	info := decodeArtifactEvent(s.T(), s.audit.published[0])
+	s.True(info.Skipped)
+	s.Equal(int64(2048), info.SizeBytes)
+}
+
+// TestWriteStreamingExistError: an Exists lookup failure is masked as Internal.
+func (s *bytestreamSuite) TestWriteStreamingExistError() {
+	s.streamingBackend.On("Exists", mock.Anything, s.resource.Digest).
+		Return(false, fmt.Errorf("backend unreachable"))
+
+	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"),
+	}))
+
+	_, err = stream.CloseAndRecv()
+	assertGRPCError(s.T(), err, codes.Internal, "server error")
+	s.Empty(s.audit.published)
+}
diff --git a/app/artifact-cas/internal/service/bytestream_test.go b/app/artifact-cas/internal/service/bytestream_test.go
index 45c2cf451..096a467ea 100644
--- a/app/artifact-cas/internal/service/bytestream_test.go
+++ b/app/artifact-cas/internal/service/bytestream_test.go
@@ -21,9 +21,11 @@ import (
 	"encoding/base64"
 	"encoding/gob"
 	"errors"
+	"fmt"
 	"io"
 	"net"
 	"strconv"
+	"syscall"
 	"testing"
 
 	v1 "github.com/chainloop-dev/chainloop/app/artifact-cas/api/cas/v1"
@@ -254,6 +256,101 @@ 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.
+func (s *bytestreamSuite) TestWriteBufferedMultiChunkIntegrity() {
+	content := []byte("chainloop attestation payload spanning multiple stream chunks")
+	resource := resourceWithDigest(content, "artifact.bin")
+
+	received := make(chan []byte, 1)
+	s.ociBackend.On("Exists", mock.Anything, resource.Digest).Return(false, nil)
+	s.ociBackend.On("Upload", mock.Anything, mock.Anything, resource).Return(nil).Run(func(args mock.Arguments) {
+		got, err := io.ReadAll(args.Get(1).(io.Reader))
+		s.NoError(err)
+		received <- got
+	})
+
+	stream, err := s.client.Write(s.upCtx)
+	s.NoError(err)
+	sendInChunks(s.T(), stream, encodeResource(s.T(), resource), content, 9)
+
+	got, err := stream.CloseAndRecv()
+	s.NoError(err)
+	s.Equal(int64(len(content)), got.CommittedSize)
+
+	gotBytes := <-received
+	s.Equal(content, gotBytes)
+	s.Equal(resource.Digest, sha256Hex(gotBytes), "backend content digest must match the declared digest")
+}
+
+// TestWriteBufferedFinishWriteWithData: the buffered/OCI path must also store a
+// trailing chunk carried on a finish_write=true message, not drop it.
+func (s *bytestreamSuite) TestWriteBufferedFinishWriteWithData() {
+	content := []byte("hello world")
+	resource := resourceWithDigest(content, "artifact.bin")
+
+	received := make(chan []byte, 1)
+	s.ociBackend.On("Exists", mock.Anything, resource.Digest).Return(false, nil)
+	s.ociBackend.On("Upload", mock.Anything, mock.Anything, resource).Return(nil).Run(func(args mock.Arguments) {
+		got, err := io.ReadAll(args.Get(1).(io.Reader))
+		s.NoError(err)
+		received <- got
+	})
+
+	stream, err := s.client.Write(s.upCtx)
+	s.NoError(err)
+	rb := encodeResource(s.T(), resource)
+	s.NoError(stream.Send(&bytestream.WriteRequest{ResourceName: rb, Data: content[:5]}))
+	s.NoError(stream.Send(&bytestream.WriteRequest{ResourceName: rb, Data: content[5:], FinishWrite: true}))
+
+	got, err := stream.CloseAndRecv()
+	s.NoError(err)
+	s.Equal(int64(len(content)), got.CommittedSize)
+	s.Equal(content, <-received)
+}
+
+// TestWriteBufferedBackendResetNotTreatedAsDisconnect: the buffered/OCI path
+// 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).
+		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"),
+	}))
+
+	_, err = stream.CloseAndRecv()
+	// "server error" is the masked message; a buggy disconnect path would return
+	// nil and surface a gRPC "cardinality violation" instead.
+	assertGRPCError(s.T(), err, codes.Internal, "server error")
+	s.Empty(s.audit.published)
+}
+
+// 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).
+		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"),
+	}))
+
+	_, err = stream.CloseAndRecv()
+	assertGRPCError(s.T(), err, codes.Internal, "server error")
+	s.Empty(s.audit.published)
+}
+
 func (s *bytestreamSuite) TestRead() {
 	ctx := s.downCtx
 
@@ -360,10 +457,13 @@ type bytestreamSuite struct {
 	srv        *grpc.Server
 	client     bytestream.ByteStreamClient
 	ociBackend *mocks.UploaderDownloader
-	resource   *v1.CASResource
-	audit      *fakePublisher
-	upCtx      context.Context
-	downCtx    context.Context
+	// streamingBackend is a streaming-capable backend (implements
+	// backend.StreamingUploader) reachable via the "backend-streaming" metadata.
+	streamingBackend *mocks.UploaderDownloader
+	resource         *v1.CASResource
+	audit            *fakePublisher
+	upCtx            context.Context
+	downCtx          context.Context
 }
 
 // Run after each test
@@ -389,6 +489,12 @@ func (s *bytestreamSuite) SetupTest() {
 					claims := &casJWT.Claims{
 						StoredSecretID: "secret-id", BackendType: backendType, OrgID: testOrgID,
 					}
+					// Route to the streaming-capable backend when requested so
+					// the same suite can exercise both the buffered (OCI) and
+					// streaming (object-store) code paths.
+					if v := md.Get("backend-streaming"); len(v) > 0 {
+						claims.BackendType = streamingBackendType
+					}
 					if v := md.Get("source-internal"); len(v) > 0 {
 						claims.SourceInternal = true
 					}
@@ -416,11 +522,20 @@ 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.
+	streamingBackend := mocks.NewUploaderDownloader(s.T())
+	streamingBackendProvider := mocks.NewProvider(s.T())
+	streamingBackendProvider.On("FromCredentials", mock.Anything, mock.Anything).Maybe().
+		Return(&streamingUploaderDownloader{streamingBackend}, nil)
+
 	s.audit = &fakePublisher{}
 	bytestream.RegisterByteStreamServer(
 		server,
 		NewByteStreamService(backend.Providers{
-			backendType: ociBackendProvider,
+			backendType:          ociBackendProvider,
+			streamingBackendType: streamingBackendProvider,
 		}, WithLogger(log.DefaultLogger), WithAuditDispatcher(newTestDispatcher(s.audit))),
 	)
 	go func() {
@@ -437,6 +552,7 @@ func (s *bytestreamSuite) SetupTest() {
 	s.srv = server
 	s.conn = conn
 	s.ociBackend = ociBackend
+	s.streamingBackend = streamingBackend
 	s.client = bytestream.NewByteStreamClient(conn)
 	s.resource = &v1.CASResource{
 		Digest: "deadbeef", FileName: "skynet.exe",
diff --git a/pkg/blobmanager/azureblob/backend.go b/pkg/blobmanager/azureblob/backend.go
index 1f41486e0..d70893553 100644
--- a/pkg/blobmanager/azureblob/backend.go
+++ b/pkg/blobmanager/azureblob/backend.go
@@ -1,5 +1,5 @@
 //
-// Copyright 2023-2025 The Chainloop Authors.
+// Copyright 2023-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.
@@ -37,7 +37,15 @@ type Backend struct {
 	endpoint           string
 }
 
-var _ backend.UploaderDownloader = (*Backend)(nil)
+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 }
 
 func NewBackend(creds *Credentials) (*Backend, error) {
 	credential, err := azidentity.NewClientSecretCredential(creds.TenantID, creds.ClientID, creds.ClientSecret, nil)
diff --git a/pkg/blobmanager/azureblob/backend_test.go b/pkg/blobmanager/azureblob/backend_test.go
new file mode 100644
index 000000000..f3a17b20d
--- /dev/null
+++ b/pkg/blobmanager/azureblob/backend_test.go
@@ -0,0 +1,34 @@
+//
+// 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 37652c009..c4c4f355a 100644
--- a/pkg/blobmanager/backend.go
+++ b/pkg/blobmanager/backend.go
@@ -1,5 +1,5 @@
 //
-// Copyright 2023 The Chainloop Authors.
+// Copyright 2023-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.
@@ -44,6 +44,22 @@ 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 fbdb4922e..c0d47638d 100644
--- a/pkg/blobmanager/oci/backend_test.go
+++ b/pkg/blobmanager/oci/backend_test.go
@@ -1,5 +1,5 @@
 //
-// Copyright 2023 The Chainloop Authors.
+// Copyright 2023-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.
@@ -27,6 +27,7 @@ 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"
@@ -408,3 +409,14 @@ 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 9a0d47d55..8fdc9546a 100644
--- a/pkg/blobmanager/s3/backend.go
+++ b/pkg/blobmanager/s3/backend.go
@@ -1,5 +1,5 @@
 //
-// Copyright 2024-2025 The Chainloop Authors.
+// Copyright 2024-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.
@@ -46,10 +46,19 @@ type Backend struct {
 	customEndpoint string
 }
 
-var _ backend.UploaderDownloader = (*Backend)(nil)
+var (
+	_ backend.UploaderDownloader = (*Backend)(nil)
+	_ backend.StreamingUploader  = (*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")
diff --git a/pkg/blobmanager/s3/backend_test.go b/pkg/blobmanager/s3/backend_test.go
index 51942f05a..91ee4c6da 100644
--- a/pkg/blobmanager/s3/backend_test.go
+++ b/pkg/blobmanager/s3/backend_test.go
@@ -376,3 +376,13 @@ 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 6c30e663a..71d2524cc 100644
--- a/pkg/blobmanager/s3accesspoint/backend.go
+++ b/pkg/blobmanager/s3accesspoint/backend.go
@@ -73,7 +73,16 @@ type Backend struct {
 	s3Client *s3.Client
 }
 
-var _ backend.UploaderDownloader = (*Backend)(nil)
+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 }
 
 // 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 57cbd34a8..1b929bd7e 100644
--- a/pkg/blobmanager/s3accesspoint/backend_test.go
+++ b/pkg/blobmanager/s3accesspoint/backend_test.go
@@ -26,6 +26,7 @@ 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"
@@ -294,3 +295,13 @@ 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())
+}