diff --git a/internal/quic/config.go b/internal/quic/config.go
index 7d1b743..df49357 100644
--- a/internal/quic/config.go
+++ b/internal/quic/config.go
@@ -17,4 +17,29 @@
 	// TLSConfig is the endpoint's TLS configuration.
 	// It must be non-nil and include at least one certificate or else set GetCertificate.
 	TLSConfig *tls.Config
+
+	// StreamReadBufferSize is the maximum amount of data sent by the peer that a
+	// stream will buffer for reading.
+	// If zero, the default value of 1MiB is used.
+	// If negative, the limit is zero.
+	StreamReadBufferSize int64
+
+	// StreamWriteBufferSize is the maximum amount of data a stream will buffer for
+	// sending to the peer.
+	// If zero, the default value of 1MiB is used.
+	// If negative, the limit is zero.
+	StreamWriteBufferSize int64
 }
+
+func configDefault(v, def int64) int64 {
+	switch v {
+	case -1:
+		return 0
+	case 0:
+		return def
+	}
+	return v
+}
+
+func (c *Config) streamReadBufferSize() int64  { return configDefault(c.StreamReadBufferSize, 1<<20) }
+func (c *Config) streamWriteBufferSize() int64 { return configDefault(c.StreamWriteBufferSize, 1<<20) }
diff --git a/internal/quic/conn.go b/internal/quic/conn.go
index 90e6739..0952a79 100644
--- a/internal/quic/conn.go
+++ b/internal/quic/conn.go
@@ -160,6 +160,9 @@
 
 // receiveTransportParameters applies transport parameters sent by the peer.
 func (c *Conn) receiveTransportParameters(p transportParameters) error {
+	c.streams.peerInitialMaxStreamDataBidiLocal = p.initialMaxStreamDataBidiLocal
+	c.streams.peerInitialMaxStreamDataRemote[bidiStream] = p.initialMaxStreamDataBidiRemote
+	c.streams.peerInitialMaxStreamDataRemote[uniStream] = p.initialMaxStreamDataUni
 	c.peerAckDelayExponent = p.ackDelayExponent
 	c.loss.setMaxAckDelay(p.maxAckDelay)
 	if err := c.connIDState.setPeerActiveConnIDLimit(p.activeConnIDLimit, c.newConnIDFunc()); err != nil {
diff --git a/internal/quic/conn_loss.go b/internal/quic/conn_loss.go
index ca17808..f42f7e5 100644
--- a/internal/quic/conn_loss.go
+++ b/internal/quic/conn_loss.go
@@ -44,6 +44,14 @@
 		case frameTypeCrypto:
 			start, end := sent.nextRange()
 			c.crypto[space].ackOrLoss(start, end, fate)
+		case frameTypeMaxStreamData,
+			frameTypeStreamDataBlocked:
+			id := streamID(sent.nextInt())
+			s := c.streamForID(id)
+			if s == nil {
+				continue
+			}
+			s.ackOrLoss(sent.num, f, fate)
 		case frameTypeStreamBase,
 			frameTypeStreamBase | streamFinBit:
 			id := streamID(sent.nextInt())
diff --git a/internal/quic/conn_loss_test.go b/internal/quic/conn_loss_test.go
index e3d16a7..d944515 100644
--- a/internal/quic/conn_loss_test.go
+++ b/internal/quic/conn_loss_test.go
@@ -7,7 +7,9 @@
 package quic
 
 import (
+	"context"
 	"crypto/tls"
+	"fmt"
 	"testing"
 )
 
@@ -145,7 +147,275 @@
 				data: []byte{},
 			})
 	})
+}
 
+func TestLostStreamWithData(t *testing.T) {
+	// "Application data sent in STREAM frames is retransmitted in new STREAM
+	// frames unless the endpoint has sent a RESET_STREAM for that stream."
+	// https://www.rfc-editor.org/rfc/rfc9000#section-13.3-3.2
+	//
+	// TODO: Lost stream frame after RESET_STREAM
+	lostFrameTest(t, func(t *testing.T, pto bool) {
+		data := []byte{0, 1, 2, 3, 4, 5, 6, 7}
+		tc, s := newTestConnAndLocalStream(t, serverSide, uniStream, func(p *transportParameters) {
+			p.initialMaxStreamsUni = 1
+			p.initialMaxData = 1 << 20
+			p.initialMaxStreamDataUni = 1 << 20
+		})
+		s.Write(data[:4])
+		tc.wantFrame("send [0,4)",
+			packetType1RTT, debugFrameStream{
+				id:   s.id,
+				off:  0,
+				data: data[:4],
+			})
+		s.Write(data[4:8])
+		tc.wantFrame("send [4,8)",
+			packetType1RTT, debugFrameStream{
+				id:   s.id,
+				off:  4,
+				data: data[4:8],
+			})
+		s.Close()
+		tc.wantFrame("send FIN",
+			packetType1RTT, debugFrameStream{
+				id:   s.id,
+				off:  8,
+				fin:  true,
+				data: []byte{},
+			})
+
+		tc.triggerLossOrPTO(packetType1RTT, pto)
+		tc.wantFrame("resend data",
+			packetType1RTT, debugFrameStream{
+				id:   s.id,
+				off:  0,
+				fin:  true,
+				data: data[:8],
+			})
+	})
+}
+
+func TestLostStreamPartialLoss(t *testing.T) {
+	// Conn sends four STREAM packets.
+	// ACKs are received for the packets containing bytes 0 and 2.
+	// The remaining packets are declared lost.
+	// The Conn resends only the lost data.
+	//
+	// This test doesn't have a PTO mode, because the ACK for the packet containing byte 2
+	// starts the loss timer for the packet containing byte 1, and the PTO timer is not
+	// armed when the loss timer is.
+	data := []byte{0, 1, 2, 3}
+	tc, s := newTestConnAndLocalStream(t, serverSide, uniStream, func(p *transportParameters) {
+		p.initialMaxStreamsUni = 1
+		p.initialMaxData = 1 << 20
+		p.initialMaxStreamDataUni = 1 << 20
+	})
+	for i := range data {
+		s.Write(data[i : i+1])
+		tc.wantFrame(fmt.Sprintf("send STREAM frame with byte %v", i),
+			packetType1RTT, debugFrameStream{
+				id:   s.id,
+				off:  int64(i),
+				data: data[i : i+1],
+			})
+		if i%2 == 0 {
+			num := tc.sentFramePacket.num
+			tc.writeFrames(packetType1RTT, debugFrameAck{
+				ranges: []i64range[packetNumber]{
+					{num, num + 1},
+				},
+			})
+		}
+	}
+	const pto = false
+	tc.triggerLossOrPTO(packetType1RTT, pto)
+	tc.wantFrame("resend byte 1",
+		packetType1RTT, debugFrameStream{
+			id:   s.id,
+			off:  1,
+			data: data[1:2],
+		})
+	tc.wantFrame("resend byte 3",
+		packetType1RTT, debugFrameStream{
+			id:   s.id,
+			off:  3,
+			data: data[3:4],
+		})
+	tc.wantIdle("no more frames sent after packet loss")
+}
+
+func TestLostMaxStreamDataFrame(t *testing.T) {
+	// "[...] an updated value is sent when the packet containing
+	// the most recent MAX_STREAM_DATA frame for a stream is lost"
+	// https://www.rfc-editor.org/rfc/rfc9000#section-13.3-3.8
+	lostFrameTest(t, func(t *testing.T, pto bool) {
+		const maxWindowSize = 10
+		buf := make([]byte, maxWindowSize)
+		tc, s := newTestConnAndRemoteStream(t, serverSide, uniStream, func(c *Config) {
+			c.StreamReadBufferSize = maxWindowSize
+		})
+
+		// We send MAX_STREAM_DATA = 19.
+		tc.writeFrames(packetType1RTT, debugFrameStream{
+			id:   s.id,
+			off:  0,
+			data: make([]byte, maxWindowSize),
+		})
+		if n, err := s.Read(buf[:maxWindowSize-1]); err != nil || n != maxWindowSize-1 {
+			t.Fatalf("Read() = %v, %v; want %v, nil", n, err, maxWindowSize-1)
+		}
+		tc.wantFrame("stream window is extended after reading data",
+			packetType1RTT, debugFrameMaxStreamData{
+				id:  s.id,
+				max: (maxWindowSize * 2) - 1,
+			})
+
+		// MAX_STREAM_DATA = 20, which is only one more byte, so we don't send the frame.
+		if n, err := s.Read(buf); err != nil || n != 1 {
+			t.Fatalf("Read() = %v, %v; want %v, nil", n, err, 1)
+		}
+		tc.wantIdle("read doesn't extend window enough to send another MAX_STREAM_DATA")
+
+		// The MAX_STREAM_DATA = 19 packet was lost, so we send 20.
+		tc.triggerLossOrPTO(packetType1RTT, pto)
+		tc.wantFrame("resent MAX_STREAM_DATA includes most current value",
+			packetType1RTT, debugFrameMaxStreamData{
+				id:  s.id,
+				max: maxWindowSize * 2,
+			})
+	})
+}
+
+func TestLostMaxStreamDataFrameAfterStreamFinReceived(t *testing.T) {
+	// "An endpoint SHOULD stop sending MAX_STREAM_DATA frames when
+	// the receiving part of the stream enters a "Size Known" or "Reset Recvd" state."
+	// https://www.rfc-editor.org/rfc/rfc9000#section-13.3-3.8
+	lostFrameTest(t, func(t *testing.T, pto bool) {
+		const maxWindowSize = 10
+		buf := make([]byte, maxWindowSize)
+		tc, s := newTestConnAndRemoteStream(t, serverSide, uniStream, func(c *Config) {
+			c.StreamReadBufferSize = maxWindowSize
+		})
+
+		tc.writeFrames(packetType1RTT, debugFrameStream{
+			id:   s.id,
+			off:  0,
+			data: make([]byte, maxWindowSize),
+		})
+		if n, err := s.Read(buf); err != nil || n != maxWindowSize {
+			t.Fatalf("Read() = %v, %v; want %v, nil", n, err, maxWindowSize)
+		}
+		tc.wantFrame("stream window is extended after reading data",
+			packetType1RTT, debugFrameMaxStreamData{
+				id:  s.id,
+				max: 2 * maxWindowSize,
+			})
+
+		tc.writeFrames(packetType1RTT, debugFrameStream{
+			id:  s.id,
+			off: maxWindowSize,
+			fin: true,
+		})
+
+		tc.ignoreFrame(frameTypePing)
+		tc.triggerLossOrPTO(packetType1RTT, pto)
+		tc.wantIdle("lost MAX_STREAM_DATA not resent for stream in 'size known'")
+	})
+}
+
+func TestLostStreamDataBlockedFrame(t *testing.T) {
+	// "A new [STREAM_DATA_BLOCKED] frame is sent if a packet containing
+	// the most recent frame for a scope is lost [...]"
+	// https://www.rfc-editor.org/rfc/rfc9000#section-13.3-3.10
+	lostFrameTest(t, func(t *testing.T, pto bool) {
+		tc, s := newTestConnAndLocalStream(t, serverSide, uniStream, func(p *transportParameters) {
+			p.initialMaxStreamsUni = 1
+			p.initialMaxData = 1 << 20
+		})
+
+		w := runAsync(tc, func(ctx context.Context) (int, error) {
+			return s.WriteContext(ctx, []byte{0, 1, 2, 3})
+		})
+		defer w.cancel()
+		tc.wantFrame("write is blocked by flow control",
+			packetType1RTT, debugFrameStreamDataBlocked{
+				id:  s.id,
+				max: 0,
+			})
+
+		tc.writeFrames(packetType1RTT, debugFrameMaxStreamData{
+			id:  s.id,
+			max: 1,
+		})
+		tc.wantFrame("write makes some progress, but is still blocked by flow control",
+			packetType1RTT, debugFrameStreamDataBlocked{
+				id:  s.id,
+				max: 1,
+			})
+		tc.wantFrame("write consuming available window",
+			packetType1RTT, debugFrameStream{
+				id:   s.id,
+				off:  0,
+				data: []byte{0},
+			})
+
+		tc.triggerLossOrPTO(packetType1RTT, pto)
+		tc.wantFrame("STREAM_DATA_BLOCKED is resent",
+			packetType1RTT, debugFrameStreamDataBlocked{
+				id:  s.id,
+				max: 1,
+			})
+		tc.wantFrame("STREAM is resent as well",
+			packetType1RTT, debugFrameStream{
+				id:   s.id,
+				off:  0,
+				data: []byte{0},
+			})
+	})
+}
+
+func TestLostStreamDataBlockedFrameAfterStreamUnblocked(t *testing.T) {
+	// "A new [STREAM_DATA_BLOCKED] frame is sent [...] only while
+	// the endpoint is blocked on the corresponding limit."
+	// https://www.rfc-editor.org/rfc/rfc9000#section-13.3-3.10
+	lostFrameTest(t, func(t *testing.T, pto bool) {
+		tc, s := newTestConnAndLocalStream(t, serverSide, uniStream, func(p *transportParameters) {
+			p.initialMaxStreamsUni = 1
+			p.initialMaxData = 1 << 20
+		})
+
+		data := []byte{0, 1, 2, 3}
+		w := runAsync(tc, func(ctx context.Context) (int, error) {
+			return s.WriteContext(ctx, data)
+		})
+		defer w.cancel()
+		tc.wantFrame("write is blocked by flow control",
+			packetType1RTT, debugFrameStreamDataBlocked{
+				id:  s.id,
+				max: 0,
+			})
+
+		tc.writeFrames(packetType1RTT, debugFrameMaxStreamData{
+			id:  s.id,
+			max: 10,
+		})
+		tc.wantFrame("write completes after flow control available",
+			packetType1RTT, debugFrameStream{
+				id:   s.id,
+				off:  0,
+				data: data,
+			})
+
+		tc.triggerLossOrPTO(packetType1RTT, pto)
+		tc.wantFrame("STREAM data is resent",
+			packetType1RTT, debugFrameStream{
+				id:   s.id,
+				off:  0,
+				data: data,
+			})
+		tc.wantIdle("STREAM_DATA_BLOCKED is not resent, since the stream is not blocked")
+	})
 }
 
 func TestLostNewConnectionIDFrame(t *testing.T) {
diff --git a/internal/quic/conn_recv.go b/internal/quic/conn_recv.go
index 45ef384..00985b6 100644
--- a/internal/quic/conn_recv.go
+++ b/internal/quic/conn_recv.go
@@ -191,7 +191,7 @@
 			if !frameOK(c, ptype, __01) {
 				return
 			}
-			_, _, n = consumeMaxStreamDataFrame(payload)
+			n = c.handleMaxStreamDataFrame(now, payload)
 		case frameTypeMaxStreamsBidi, frameTypeMaxStreamsUni:
 			if !frameOK(c, ptype, __01) {
 				return
@@ -280,6 +280,17 @@
 	return n
 }
 
+func (c *Conn) handleMaxStreamDataFrame(now time.Time, payload []byte) int {
+	id, maxStreamData, n := consumeMaxStreamDataFrame(payload)
+	if s := c.streamForFrame(now, id, sendStream); s != nil {
+		if err := s.handleMaxStreamData(maxStreamData); err != nil {
+			c.abort(now, err)
+			return -1
+		}
+	}
+	return n
+}
+
 func (c *Conn) handleCryptoFrame(now time.Time, space numberSpace, payload []byte) int {
 	off, data, n := consumeCryptoFrame(payload)
 	err := c.handleCrypto(now, space, off, data)
diff --git a/internal/quic/conn_streams.go b/internal/quic/conn_streams.go
index f626323..7a531f5 100644
--- a/internal/quic/conn_streams.go
+++ b/internal/quic/conn_streams.go
@@ -20,6 +20,10 @@
 	streams   map[streamID]*Stream
 	opened    [streamTypeCount]int64 // number of streams opened by us
 
+	// Peer configuration provided in transport parameters.
+	peerInitialMaxStreamDataRemote    [streamTypeCount]int64 // streams opened by us
+	peerInitialMaxStreamDataBidiLocal int64                  // streams opened by them
+
 	// Streams with frames to send are stored in a circular linked list.
 	// sendHead is the next stream to write, or nil if there are no streams
 	// with data to send. sendTail is the last stream to write.
@@ -55,15 +59,24 @@
 	return c.newLocalStream(ctx, uniStream)
 }
 
-func (c *Conn) newLocalStream(ctx context.Context, typ streamType) (*Stream, error) {
+func (c *Conn) newLocalStream(ctx context.Context, styp streamType) (*Stream, error) {
 	// TODO: Stream limits.
 	c.streams.streamsMu.Lock()
 	defer c.streams.streamsMu.Unlock()
 
-	num := c.streams.opened[typ]
-	c.streams.opened[typ]++
+	num := c.streams.opened[styp]
+	c.streams.opened[styp]++
 
-	s := newStream(c, newStreamID(c.side, typ, num))
+	s := newStream(c, newStreamID(c.side, styp, num))
+	s.outmaxbuf = c.config.streamWriteBufferSize()
+	s.outwin = c.streams.peerInitialMaxStreamDataRemote[styp]
+	if styp == bidiStream {
+		s.inmaxbuf = c.config.streamReadBufferSize()
+		s.inwin = c.config.streamReadBufferSize()
+	}
+	s.inUnlock()
+	s.outUnlock()
+
 	c.streams.streams[s.id] = s
 	return s, nil
 }
@@ -117,7 +130,17 @@
 		c.abort(now, localTransportError(errStreamState))
 		return nil
 	}
+
 	s := newStream(c, id)
+	s.inmaxbuf = c.config.streamReadBufferSize()
+	s.inwin = c.config.streamReadBufferSize()
+	if id.streamType() == bidiStream {
+		s.outmaxbuf = c.config.streamWriteBufferSize()
+		s.outwin = c.streams.peerInitialMaxStreamDataBidiLocal
+	}
+	s.inUnlock()
+	s.outUnlock()
+
 	c.streams.streams[id] = s
 	c.streams.queue.put(s)
 	return s
diff --git a/internal/quic/conn_test.go b/internal/quic/conn_test.go
index 5aad69f..2480f9c 100644
--- a/internal/quic/conn_test.go
+++ b/internal/quic/conn_test.go
@@ -179,6 +179,8 @@
 	peerProvidedParams := defaultTransportParameters()
 	for _, o := range opts {
 		switch o := o.(type) {
+		case func(*Config):
+			o(config)
 		case func(*tls.Config):
 			o(config.TLSConfig)
 		case func(p *transportParameters):
diff --git a/internal/quic/crypto_stream.go b/internal/quic/crypto_stream.go
index 6cda657..75dea87 100644
--- a/internal/quic/crypto_stream.go
+++ b/internal/quic/crypto_stream.go
@@ -118,28 +118,7 @@
 // copy the data it wants into position.
 func (s *cryptoStream) dataToSend(pto bool, f func(off, size int64) (sent int64)) {
 	for {
-		var off, size int64
-		if pto {
-			// On PTO, resend unacked data that fits in the probe packet.
-			// For simplicity, we send the range starting at s.out.start
-			// (which is definitely unacked, or else we would have discarded it)
-			// up to the next acked byte (if any).
-			//
-			// This may miss unacked data starting after that acked byte,
-			// but avoids resending data the peer has acked.
-			off = s.out.start
-			end := s.out.end
-			for _, r := range s.outacked {
-				if r.start > off {
-					end = r.start
-					break
-				}
-			}
-			size = end - s.out.start
-		} else if s.outunsent.numRanges() > 0 {
-			off = s.outunsent.min()
-			size = s.outunsent[0].size()
-		}
+		off, size := dataToSend(s.out, s.outunsent, s.outacked, pto)
 		if size == 0 {
 			return
 		}
diff --git a/internal/quic/gate.go b/internal/quic/gate.go
index efb28da..27ab07a 100644
--- a/internal/quic/gate.go
+++ b/internal/quic/gate.go
@@ -20,13 +20,19 @@
 	unset chan struct{}
 }
 
+// newGate returns a new, unlocked gate with the condition unset.
 func newGate() gate {
-	g := gate{
+	g := newLockedGate()
+	g.unlock(false)
+	return g
+}
+
+// newLocked gate returns a new, locked gate.
+func newLockedGate() gate {
+	return gate{
 		set:   make(chan struct{}, 1),
 		unset: make(chan struct{}, 1),
 	}
-	g.unset <- struct{}{}
-	return g
 }
 
 // lock acquires the gate unconditionally.
diff --git a/internal/quic/quic_test.go b/internal/quic/quic_test.go
new file mode 100644
index 0000000..1281b54
--- /dev/null
+++ b/internal/quic/quic_test.go
@@ -0,0 +1,37 @@
+// Copyright 2023 The Go Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+//go:build go1.21
+
+package quic
+
+import (
+	"testing"
+)
+
+func testSides(t *testing.T, name string, f func(*testing.T, connSide)) {
+	if name != "" {
+		name += "/"
+	}
+	t.Run(name+"server", func(t *testing.T) { f(t, serverSide) })
+	t.Run(name+"client", func(t *testing.T) { f(t, clientSide) })
+}
+
+func testStreamTypes(t *testing.T, name string, f func(*testing.T, streamType)) {
+	if name != "" {
+		name += "/"
+	}
+	t.Run(name+"bidi", func(t *testing.T) { f(t, bidiStream) })
+	t.Run(name+"uni", func(t *testing.T) { f(t, uniStream) })
+}
+
+func testSidesAndStreamTypes(t *testing.T, name string, f func(*testing.T, connSide, streamType)) {
+	if name != "" {
+		name += "/"
+	}
+	t.Run(name+"server/bidi", func(t *testing.T) { f(t, serverSide, bidiStream) })
+	t.Run(name+"client/bidi", func(t *testing.T) { f(t, clientSide, bidiStream) })
+	t.Run(name+"server/uni", func(t *testing.T) { f(t, serverSide, uniStream) })
+	t.Run(name+"client/uni", func(t *testing.T) { f(t, clientSide, uniStream) })
+}
diff --git a/internal/quic/stream.go b/internal/quic/stream.go
index b55f927..83215df 100644
--- a/internal/quic/stream.go
+++ b/internal/quic/stream.go
@@ -9,34 +9,57 @@
 import (
 	"context"
 	"errors"
+	"io"
 )
 
 type Stream struct {
 	id   streamID
 	conn *Conn
 
+	// ingate's lock guards all receive-related state.
+	//
+	// The gate condition is set if a read from the stream will not block,
+	// either because the stream has available data or because the read will fail.
+	ingate    gate
+	in        pipe            // received data
+	inwin     int64           // last MAX_STREAM_DATA sent to the peer
+	insendmax sentVal         // set when we should send MAX_STREAM_DATA to the peer
+	inmaxbuf  int64           // maximum amount of data we will buffer
+	insize    int64           // stream final size; -1 before this is known
+	inset     rangeset[int64] // received ranges
+
 	// outgate's lock guards all send-related state.
 	//
 	// The gate condition is set if a write to the stream will not block,
 	// either because the stream has available flow control or because
 	// the write will fail.
-	outgate   gate
-	outopened sentVal // set if we should open the stream
+	outgate    gate
+	out        pipe            // buffered data to send
+	outwin     int64           // maximum MAX_STREAM_DATA received from the peer
+	outmaxbuf  int64           // maximum amount of data we will buffer
+	outunsent  rangeset[int64] // ranges buffered but not yet sent
+	outacked   rangeset[int64] // ranges sent and acknowledged
+	outopened  sentVal         // set if we should open the stream
+	outclosed  sentVal         // set by CloseWrite
+	outblocked sentVal         // set when a write to the stream is blocked by flow control
 
 	prev, next *Stream // guarded by streamsState.sendMu
 }
 
+// newStream returns a new stream.
+//
+// The stream's ingate and outgate are locked.
+// (We create the stream with locked gates so after the caller
+// initializes the flow control window,
+// unlocking outgate will set the stream writability state.)
 func newStream(c *Conn, id streamID) *Stream {
 	s := &Stream{
 		conn:    c,
 		id:      id,
-		outgate: newGate(),
+		insize:  -1, // -1 indicates the stream size is unknown
+		ingate:  newLockedGate(),
+		outgate: newLockedGate(),
 	}
-
-	// Lock and unlock outgate to update the stream writability state.
-	s.outgate.lock()
-	s.outUnlock()
-
 	return s
 }
 
@@ -66,8 +89,48 @@
 // returning all data sent by the peer.
 // If the peer terminates reads abruptly, ReadContext returns StreamResetError.
 func (s *Stream) ReadContext(ctx context.Context, b []byte) (n int, err error) {
-	// TODO: implement
-	return 0, errors.New("unimplemented")
+	if s.IsWriteOnly() {
+		return 0, errors.New("read from write-only stream")
+	}
+	// Wait until data is available.
+	if err := s.conn.waitAndLockGate(ctx, &s.ingate); err != nil {
+		return 0, err
+	}
+	defer s.inUnlock()
+	if s.insize == s.in.start {
+		return 0, io.EOF
+	}
+	// Getting here indicates the stream contains data to be read.
+	if len(s.inset) < 1 || s.inset[0].start != 0 || s.inset[0].end <= s.in.start {
+		panic("BUG: inconsistent input stream state")
+	}
+	if size := int(s.inset[0].end - s.in.start); size < len(b) {
+		b = b[:size]
+	}
+	start := s.in.start
+	end := start + int64(len(b))
+	s.in.copy(start, b)
+	s.in.discardBefore(end)
+	if s.insize == -1 || s.insize > s.inwin {
+		if shouldUpdateFlowControl(s.inwin-s.in.start, s.inmaxbuf) {
+			// Update stream flow control with a STREAM_MAX_DATA frame.
+			s.insendmax.setUnsent()
+		}
+	}
+	if end == s.insize {
+		return len(b), io.EOF
+	}
+	return len(b), nil
+}
+
+// shouldUpdateFlowControl determines whether to send a flow control window update.
+//
+// We want to balance keeping the peer well-supplied with flow control with not sending
+// many small updates.
+func shouldUpdateFlowControl(curwin, maxwin int64) bool {
+	// Update flow control if doing so gives the peer at least 64k tokens,
+	// or if it will double the current window.
+	return maxwin-curwin >= 64<<10 || curwin*2 < maxwin
 }
 
 // Write writes data to the stream.
@@ -87,65 +150,330 @@
 	if s.IsReadOnly() {
 		return 0, errors.New("write to read-only stream")
 	}
-	if len(b) > 0 {
-		// TODO: implement
-		return 0, errors.New("unimplemented")
+	canWrite := s.outgate.lock()
+	if s.outclosed.isSet() {
+		s.outUnlock()
+		return 0, errors.New("write to closed stream")
 	}
-	if err := s.outgate.waitAndLockContext(ctx); err != nil {
-		return 0, err
+	if len(b) == 0 {
+		// We aren't writing any data, but send a STREAM frame to open the stream
+		// if we haven't done so already.
+		s.outopened.set()
 	}
-	defer s.outUnlock()
-
-	// Set outopened to send a STREAM frame with no data,
-	// opening the stream on the peer.
-	s.outopened.set()
-
+	for len(b) > 0 {
+		// The first time through this loop, we may or may not be write blocked.
+		// We exit the loop after writing all data, so on subsequent passes through
+		// the loop we are always write blocked.
+		if !canWrite {
+			// We're blocked, either by flow control or by our own buffer limit.
+			// We either need the peer to extend our flow control window,
+			// or ack some of our outstanding packets.
+			if s.out.end == s.outwin {
+				// We're blocked by flow control.
+				// Send a STREAM_DATA_BLOCKED frame to let the peer know.
+				s.outblocked.setUnsent()
+			}
+			s.outUnlock()
+			if err := s.conn.waitAndLockGate(ctx, &s.outgate); err != nil {
+				return n, err
+			}
+			// Successfully returning from waitAndLockGate means we are no longer
+			// write blocked. (Unlike traditional condition variables, gates do not
+			// have spurious wakeups.)
+		}
+		s.outblocked.clear()
+		// Write limit is min(our own buffer limit, the peer-provided flow control window).
+		// This is a stream offset.
+		lim := min(s.out.start+s.outmaxbuf, s.outwin)
+		// Amount to write is min(the full buffer, data up to the write limit).
+		// This is a number of bytes.
+		nn := min(int64(len(b)), lim-s.out.end)
+		// Copy the data into the output buffer and mark it as unsent.
+		s.outunsent.add(s.out.end, s.out.end+nn)
+		s.out.writeAt(b[:nn], s.out.end)
+		s.outopened.set()
+		b = b[nn:]
+		n += int(nn)
+		// If we have bytes left to send, we're blocked.
+		canWrite = false
+	}
+	s.outUnlock()
 	return n, nil
 }
 
+// Close closes the stream.
+// See CloseContext for more details.
+func (s *Stream) Close() error {
+	return s.CloseContext(context.Background())
+}
+
+// CloseContext closes the stream.
+// Any blocked stream operations will be unblocked and return errors.
+//
+// CloseContext flushes any data in the stream write buffer and waits for the peer to
+// acknowledge receipt of the data.
+// If the stream has been reset, it waits for the peer to acknowledge the reset.
+// If the context expires before the peer receives the stream's data,
+// CloseContext discards the buffer and returns the context error.
+func (s *Stream) CloseContext(ctx context.Context) error {
+	s.CloseRead()
+	s.CloseWrite()
+	// TODO: wait for peer to acknowledge data
+	// TODO: Return code from peer's RESET_STREAM frame?
+	return nil
+}
+
+// CloseRead aborts reads on the stream.
+// Any blocked reads will be unblocked and return errors.
+//
+// CloseRead notifies the peer that the stream has been closed for reading.
+// It does not wait for the peer to acknowledge the closure.
+// Use CloseContext to wait for the peer's acknowledgement.
+func (s *Stream) CloseRead() {
+	if s.IsWriteOnly() {
+		return
+	}
+	// TODO: support read-closing streams with a STOP_SENDING frame
+}
+
+// CloseWrite aborts writes on the stream.
+// Any blocked writes will be unblocked and return errors.
+//
+// CloseWrite sends any data in the stream write buffer to the peer.
+// It does not wait for the peer to acknowledge receipt of the data.
+// Use CloseContext to wait for the peer's acknowledgement.
+func (s *Stream) CloseWrite() {
+	if s.IsReadOnly() {
+		return
+	}
+	s.outgate.lock()
+	defer s.outUnlock()
+	s.outclosed.set()
+}
+
+// inUnlock unlocks s.ingate.
+// It sets the gate condition if reads from s will not block.
+// If s has receive-related frames to write, it notifies the Conn.
+func (s *Stream) inUnlock() {
+	if s.inUnlockNoQueue() {
+		s.conn.queueStreamForSend(s)
+	}
+}
+
+// inUnlockNoQueue is inUnlock,
+// but reports whether s has frames to write rather than notifying the Conn.
+func (s *Stream) inUnlockNoQueue() (shouldSend bool) {
+	// TODO: STOP_SENDING
+	canRead := s.inset.contains(s.in.start) || // data available to read
+		s.insize == s.in.start // at EOF
+	s.ingate.unlock(canRead)
+	return s.insendmax.shouldSend() // STREAM_MAX_DATA
+}
+
 // outUnlock unlocks s.outgate.
 // It sets the gate condition if writes to s will not block.
-// If s has frames to write, it notifies the Conn.
+// If s has send-related frames to write, it notifies the Conn.
 func (s *Stream) outUnlock() {
-	if s.outopened.shouldSend() {
+	if s.outUnlockNoQueue() {
 		s.conn.queueStreamForSend(s)
 	}
-	canSend := true // TODO: set sendability status based on flow control
-	s.outgate.unlock(canSend)
+}
+
+// outUnlockNoQueue is outUnlock,
+// but reports whether s has frames to write rather than notifying the Conn.
+func (s *Stream) outUnlockNoQueue() (shouldSend bool) {
+	lim := min(s.out.start+s.outmaxbuf, s.outwin)
+	canWrite := lim > s.out.end || // available flow control
+		s.outclosed.isSet() // closed
+	s.outgate.unlock(canWrite)
+	return len(s.outunsent) > 0 || // STREAM frame with data
+		s.outclosed.shouldSend() || // STREAM frame with FIN bit
+		s.outopened.shouldSend() || // STREAM frame with no data
+		s.outblocked.shouldSend() // STREAM_DATA_BLOCKED
 }
 
 // handleData handles data received in a STREAM frame.
 func (s *Stream) handleData(off int64, b []byte, fin bool) error {
-	// TODO
+	s.ingate.lock()
+	defer s.inUnlock()
+	end := off + int64(len(b))
+	if end > s.inwin {
+		// The peer sent us data past the maximum flow control window we gave them.
+		return localTransportError(errFlowControl)
+	}
+	if s.insize != -1 && end > s.insize {
+		// The peer sent us data past the final size of the stream they previously gave us.
+		return localTransportError(errFinalSize)
+	}
+	s.in.writeAt(b, off)
+	s.inset.add(off, end)
+	if fin {
+		if s.insize != -1 && s.insize != end {
+			// The peer changed the final size of the stream.
+			return localTransportError(errFinalSize)
+		}
+		s.insize = end
+		// The peer has enough flow control window to send the entire stream.
+		s.insendmax.clear()
+	}
 	return nil
 }
 
+// handleMaxStreamData handles an update received in a MAX_STREAM_DATA frame.
+func (s *Stream) handleMaxStreamData(maxStreamData int64) error {
+	s.outgate.lock()
+	defer s.outUnlock()
+	s.outwin = max(maxStreamData, s.outwin)
+	return nil
+}
+
+// ackOrLoss handles the fate of stream frames other than STREAM.
+func (s *Stream) ackOrLoss(pnum packetNumber, ftype byte, fate packetFate) {
+	// Frames which carry new information each time they are sent
+	// (MAX_STREAM_DATA, STREAM_DATA_BLOCKED) must only be marked
+	// as received if the most recent packet carrying this frame is acked.
+	//
+	// Frames which are always the same (STOP_SENDING, RESET_STREAM)
+	// can be marked as received if any packet carrying this frame is acked.
+	switch ftype {
+	case frameTypeMaxStreamData:
+		s.ingate.lock()
+		s.insendmax.ackLatestOrLoss(pnum, fate)
+		s.inUnlock()
+	case frameTypeStreamDataBlocked:
+		s.outgate.lock()
+		s.outblocked.ackLatestOrLoss(pnum, fate)
+		s.outUnlock()
+	default:
+		// TODO: Handle STOP_SENDING, RESET_STREAM.
+		panic("unhandled frame type")
+	}
+}
+
 // ackOrLossData handles the fate of a STREAM frame.
 func (s *Stream) ackOrLossData(pnum packetNumber, start, end int64, fin bool, fate packetFate) {
 	s.outgate.lock()
 	defer s.outUnlock()
 	s.outopened.ackOrLoss(pnum, fate)
+	if fin {
+		s.outclosed.ackOrLoss(pnum, fate)
+	}
+	switch fate {
+	case packetAcked:
+		s.outacked.add(start, end)
+		s.outunsent.sub(start, end)
+		// If this ack is for data at the start of the send buffer, we can now discard it.
+		if s.outacked.contains(s.out.start) {
+			s.out.discardBefore(s.outacked[0].end)
+		}
+	case packetLost:
+		// Mark everything lost, but not previously acked, as needing retransmission.
+		// We do this by adding all the lost bytes to outunsent, and then
+		// removing everything already acked.
+		s.outunsent.add(start, end)
+		for _, a := range s.outacked {
+			s.outunsent.sub(a.start, a.end)
+		}
+	}
 }
 
+// appendInFrames appends STOP_SENDING and MAX_STREAM_DATA frames
+// to the current packet.
+//
+// It returns true if no more frames need appending,
+// false if not everything fit in the current packet.
 func (s *Stream) appendInFrames(w *packetWriter, pnum packetNumber, pto bool) bool {
+	s.ingate.lock()
+	defer s.inUnlockNoQueue()
 	// TODO: STOP_SENDING
-	// TODO: MAX_STREAM_DATA
+	if s.insendmax.shouldSendPTO(pto) {
+		// MAX_STREAM_DATA
+		maxStreamData := s.in.start + s.inmaxbuf
+		if !w.appendMaxStreamDataFrame(s.id, maxStreamData) {
+			return false
+		}
+		s.inwin = maxStreamData
+		s.insendmax.setSent(pnum)
+	}
 	return true
 }
 
+// appendOutFrames appends RESET_STREAM, STREAM_DATA_BLOCKED, and STREAM frames
+// to the current packet.
+//
+// It returns true if no more frames need appending,
+// false if not everything fit in the current packet.
 func (s *Stream) appendOutFrames(w *packetWriter, pnum packetNumber, pto bool) bool {
+	s.outgate.lock()
+	defer s.outUnlockNoQueue()
 	// TODO: RESET_STREAM
-	// TODO: STREAM_DATA_BLOCKED
-	// TODO: STREAM frames with data
-	if s.outopened.shouldSendPTO(pto) {
-		off := int64(0)
-		size := 0
-		fin := false
-		_, added := w.appendStreamFrame(s.id, off, size, fin)
+	if s.outblocked.shouldSendPTO(pto) {
+		// STREAM_DATA_BLOCKED
+		if !w.appendStreamDataBlockedFrame(s.id, s.out.end) {
+			return false
+		}
+		s.outblocked.setSent(pnum)
+		s.frameOpensStream(pnum)
+	}
+	// STREAM
+	for {
+		off, size := dataToSend(s.out, s.outunsent, s.outacked, pto)
+		fin := s.outclosed.isSet() && off+size == s.out.end
+		shouldSend := size > 0 || // have data to send
+			s.outopened.shouldSendPTO(pto) || // should open the stream
+			(fin && s.outclosed.shouldSendPTO(pto)) // should close the stream
+		if !shouldSend {
+			return true
+		}
+		b, added := w.appendStreamFrame(s.id, off, int(size), fin)
 		if !added {
 			return false
 		}
+		s.out.copy(off, b)
+		s.outunsent.sub(off, off+int64(len(b)))
+		s.frameOpensStream(pnum)
+		if fin {
+			s.outclosed.setSent(pnum)
+		}
+		if pto {
+			return true
+		}
+		if int64(len(b)) < size {
+			return false
+		}
+	}
+}
+
+// frameOpensStream records that we're sending a frame that will open the stream.
+//
+// If we don't have an acknowledgement from the peer for a previous frame opening the stream,
+// record this packet as being the latest one to open it.
+func (s *Stream) frameOpensStream(pnum packetNumber) {
+	if !s.outopened.isReceived() {
 		s.outopened.setSent(pnum)
 	}
-	return true
+}
+
+// dataToSend returns the next range of data to send in a STREAM or CRYPTO_STREAM.
+func dataToSend(out pipe, outunsent, outacked rangeset[int64], pto bool) (start, size int64) {
+	switch {
+	case pto:
+		// On PTO, resend unacked data that fits in the probe packet.
+		// For simplicity, we send the range starting at s.out.start
+		// (which is definitely unacked, or else we would have discarded it)
+		// up to the next acked byte (if any).
+		//
+		// This may miss unacked data starting after that acked byte,
+		// but avoids resending data the peer has acked.
+		for _, r := range outacked {
+			if r.start > out.start {
+				return out.start, r.start - out.start
+			}
+		}
+		return out.start, out.end - out.start
+	case outunsent.numRanges() > 0:
+		return outunsent.min(), outunsent[0].size()
+	default:
+		return out.end, 0
+	}
 }
diff --git a/internal/quic/stream_test.go b/internal/quic/stream_test.go
index 8ae9dbc..d158e72 100644
--- a/internal/quic/stream_test.go
+++ b/internal/quic/stream_test.go
@@ -7,10 +7,703 @@
 package quic
 
 import (
+	"bytes"
+	"context"
+	"crypto/rand"
+	"fmt"
+	"io"
 	"reflect"
+	"strings"
 	"testing"
 )
 
+func TestStreamWriteBlockedByStreamFlowControl(t *testing.T) {
+	testStreamTypes(t, "", func(t *testing.T, styp streamType) {
+		ctx := canceledContext()
+		want := []byte{0, 1, 2, 3, 4, 5, 6, 7, 8, 9}
+		tc := newTestConn(t, clientSide, func(p *transportParameters) {
+			p.initialMaxStreamsBidi = 100
+			p.initialMaxStreamsUni = 100
+			p.initialMaxData = 1 << 20
+		})
+		tc.handshake()
+		tc.ignoreFrame(frameTypeAck)
+
+		// Non-blocking write with no flow control.
+		s, err := tc.conn.newLocalStream(ctx, styp)
+		if err != nil {
+			t.Fatal(err)
+		}
+		_, err = s.WriteContext(ctx, want)
+		if err != context.Canceled {
+			t.Fatalf("write to stream with no flow control: err = %v, want context.Canceled", err)
+		}
+		tc.wantFrame("write blocked by flow control triggers a STREAM_DATA_BLOCKED frame",
+			packetType1RTT, debugFrameStreamDataBlocked{
+				id:  s.id,
+				max: 0,
+			})
+
+		// Blocking write waiting for flow control.
+		w := runAsync(tc, func(ctx context.Context) (int, error) {
+			return s.WriteContext(ctx, want)
+		})
+		tc.wantFrame("second blocked write triggers another STREAM_DATA_BLOCKED",
+			packetType1RTT, debugFrameStreamDataBlocked{
+				id:  s.id,
+				max: 0,
+			})
+		tc.writeFrames(packetType1RTT, debugFrameMaxStreamData{
+			id:  s.id,
+			max: 4,
+		})
+		tc.wantFrame("stream window extended, but still more data to write",
+			packetType1RTT, debugFrameStreamDataBlocked{
+				id:  s.id,
+				max: 4,
+			})
+		tc.wantFrame("stream window extended to 4, expect blocked write to progress",
+			packetType1RTT, debugFrameStream{
+				id:   s.id,
+				data: want[:4],
+			})
+
+		tc.writeFrames(packetType1RTT, debugFrameMaxStreamData{
+			id:  s.id,
+			max: int64(len(want)),
+		})
+		tc.wantFrame("stream window extended further, expect blocked write to finish",
+			packetType1RTT, debugFrameStream{
+				id:   s.id,
+				off:  4,
+				data: want[4:],
+			})
+		n, err := w.result()
+		if n != len(want) || err != nil {
+			t.Errorf("Write() = %v, %v; want %v, nil", n, err, len(want))
+		}
+	})
+}
+
+func TestStreamIgnoresMaxStreamDataReduction(t *testing.T) {
+	// "A sender MUST ignore any MAX_STREAM_DATA [...] frames that
+	// do not increase flow control limits."
+	// https://www.rfc-editor.org/rfc/rfc9000#section-4.1-9
+	testStreamTypes(t, "", func(t *testing.T, styp streamType) {
+		ctx := canceledContext()
+		want := []byte{0, 1, 2, 3, 4, 5, 6, 7, 8, 9}
+		tc := newTestConn(t, clientSide, func(p *transportParameters) {
+			if styp == uniStream {
+				p.initialMaxStreamsUni = 1
+				p.initialMaxStreamDataUni = 4
+			} else {
+				p.initialMaxStreamsBidi = 1
+				p.initialMaxStreamDataBidiRemote = 4
+			}
+			p.initialMaxData = 1 << 20
+		})
+		tc.handshake()
+		tc.ignoreFrame(frameTypeAck)
+		tc.ignoreFrame(frameTypeStreamDataBlocked)
+
+		// Write [0,1).
+		s, err := tc.conn.newLocalStream(ctx, styp)
+		if err != nil {
+			t.Fatal(err)
+		}
+		s.WriteContext(ctx, want[:1])
+		tc.wantFrame("sent data (1 byte) fits within flow control limit",
+			packetType1RTT, debugFrameStream{
+				id:   s.id,
+				off:  0,
+				data: want[:1],
+			})
+
+		// MAX_STREAM_DATA tries to decrease limit, and is ignored.
+		tc.writeFrames(packetType1RTT, debugFrameMaxStreamData{
+			id:  s.id,
+			max: 2,
+		})
+
+		// Write [1,4).
+		s.WriteContext(ctx, want[1:])
+		tc.wantFrame("stream limit is 4 bytes, ignoring decrease in MAX_STREAM_DATA",
+			packetType1RTT, debugFrameStream{
+				id:   s.id,
+				off:  1,
+				data: want[1:4],
+			})
+
+		// MAX_STREAM_DATA increases limit.
+		// Second MAX_STREAM_DATA decreases it, and is ignored.
+		tc.writeFrames(packetType1RTT, debugFrameMaxStreamData{
+			id:  s.id,
+			max: 8,
+		})
+		tc.writeFrames(packetType1RTT, debugFrameMaxStreamData{
+			id:  s.id,
+			max: 6,
+		})
+
+		// Write [1,4).
+		s.WriteContext(ctx, want[4:])
+		tc.wantFrame("stream limit is 8 bytes, ignoring decrease in MAX_STREAM_DATA",
+			packetType1RTT, debugFrameStream{
+				id:   s.id,
+				off:  4,
+				data: want[4:8],
+			})
+	})
+}
+
+func TestStreamWriteBlockedByWriteBufferLimit(t *testing.T) {
+	testStreamTypes(t, "", func(t *testing.T, styp streamType) {
+		ctx := canceledContext()
+		want := []byte{0, 1, 2, 3, 4, 5, 6, 7, 8, 9}
+		const maxWriteBuffer = 4
+		tc := newTestConn(t, clientSide, func(p *transportParameters) {
+			p.initialMaxStreamsBidi = 100
+			p.initialMaxStreamsUni = 100
+			p.initialMaxData = 1 << 20
+			p.initialMaxStreamDataBidiRemote = 1 << 20
+			p.initialMaxStreamDataUni = 1 << 20
+		}, func(c *Config) {
+			c.StreamWriteBufferSize = maxWriteBuffer
+		})
+		tc.handshake()
+		tc.ignoreFrame(frameTypeAck)
+
+		// Write more data than StreamWriteBufferSize.
+		// The peer has given us plenty of flow control,
+		// so we're just blocked by our local limit.
+		s, err := tc.conn.newLocalStream(ctx, styp)
+		if err != nil {
+			t.Fatal(err)
+		}
+		w := runAsync(tc, func(ctx context.Context) (int, error) {
+			return s.WriteContext(ctx, want)
+		})
+		tc.wantFrame("stream write should send as much data as write buffer allows",
+			packetType1RTT, debugFrameStream{
+				id:   s.id,
+				off:  0,
+				data: want[:maxWriteBuffer],
+			})
+		tc.wantIdle("no STREAM_DATA_BLOCKED, we're blocked locally not by flow control")
+
+		// ACK for previously-sent data allows making more progress.
+		tc.writeFrames(packetType1RTT, debugFrameAck{
+			ranges: []i64range[packetNumber]{{0, tc.sentFramePacket.num + 1}},
+		})
+		tc.wantFrame("ACK for previous data allows making progress",
+			packetType1RTT, debugFrameStream{
+				id:   s.id,
+				off:  maxWriteBuffer,
+				data: want[maxWriteBuffer:][:maxWriteBuffer],
+			})
+
+		// Cancel the write with data left to send.
+		w.cancel()
+		n, err := w.result()
+		if n != 2*maxWriteBuffer || err == nil {
+			t.Fatalf("WriteContext() = %v, %v; want %v bytes, error", n, err, 2*maxWriteBuffer)
+		}
+	})
+}
+
+func TestStreamReceive(t *testing.T) {
+	// "Endpoints MUST be able to deliver stream data to an application as
+	// an ordered byte stream."
+	// https://www.rfc-editor.org/rfc/rfc9000#section-2.2-2
+	want := make([]byte, 5000)
+	for i := range want {
+		want[i] = byte(i)
+	}
+	type frame struct {
+		start   int64
+		end     int64
+		fin     bool
+		want    int
+		wantEOF bool
+	}
+	for _, test := range []struct {
+		name   string
+		frames []frame
+	}{{
+		name: "linear",
+		frames: []frame{{
+			start: 0,
+			end:   1000,
+			want:  1000,
+		}, {
+			start: 1000,
+			end:   2000,
+			want:  2000,
+		}, {
+			start:   2000,
+			end:     3000,
+			want:    3000,
+			fin:     true,
+			wantEOF: true,
+		}},
+	}, {
+		name: "out of order",
+		frames: []frame{{
+			start: 1000,
+			end:   2000,
+		}, {
+			start: 2000,
+			end:   3000,
+		}, {
+			start: 0,
+			end:   1000,
+			want:  3000,
+		}},
+	}, {
+		name: "resent",
+		frames: []frame{{
+			start: 0,
+			end:   1000,
+			want:  1000,
+		}, {
+			start: 0,
+			end:   1000,
+			want:  1000,
+		}, {
+			start: 1000,
+			end:   2000,
+			want:  2000,
+		}, {
+			start: 0,
+			end:   1000,
+			want:  2000,
+		}, {
+			start: 1000,
+			end:   2000,
+			want:  2000,
+		}},
+	}, {
+		name: "overlapping",
+		frames: []frame{{
+			start: 0,
+			end:   1000,
+			want:  1000,
+		}, {
+			start: 3000,
+			end:   4000,
+			want:  1000,
+		}, {
+			start: 2000,
+			end:   3000,
+			want:  1000,
+		}, {
+			start: 1000,
+			end:   3000,
+			want:  4000,
+		}},
+	}, {
+		name: "early eof",
+		frames: []frame{{
+			start: 3000,
+			end:   3000,
+			fin:   true,
+			want:  0,
+		}, {
+			start: 1000,
+			end:   2000,
+			want:  0,
+		}, {
+			start: 0,
+			end:   1000,
+			want:  2000,
+		}, {
+			start:   2000,
+			end:     3000,
+			want:    3000,
+			wantEOF: true,
+		}},
+	}, {
+		name: "empty eof",
+		frames: []frame{{
+			start: 0,
+			end:   1000,
+			want:  1000,
+		}, {
+			start:   1000,
+			end:     1000,
+			fin:     true,
+			want:    1000,
+			wantEOF: true,
+		}},
+	}} {
+		testStreamTypes(t, test.name, func(t *testing.T, styp streamType) {
+			ctx := canceledContext()
+			tc := newTestConn(t, serverSide)
+			tc.handshake()
+			sid := newStreamID(clientSide, styp, 0)
+			var s *Stream
+			got := make([]byte, len(want))
+			var total int
+			for _, f := range test.frames {
+				t.Logf("receive [%v,%v)", f.start, f.end)
+				tc.writeFrames(packetType1RTT, debugFrameStream{
+					id:   sid,
+					off:  f.start,
+					data: want[f.start:f.end],
+					fin:  f.fin,
+				})
+				if s == nil {
+					var err error
+					s, err = tc.conn.AcceptStream(ctx)
+					if err != nil {
+						tc.t.Fatalf("conn.AcceptStream() = %v", err)
+					}
+				}
+				for {
+					n, err := s.ReadContext(ctx, got[total:])
+					t.Logf("s.ReadContext() = %v, %v", n, err)
+					total += n
+					if f.wantEOF && err != io.EOF {
+						t.Fatalf("ReadContext() error = %v; want io.EOF", err)
+					}
+					if !f.wantEOF && err == io.EOF {
+						t.Fatalf("ReadContext() error = io.EOF, want something else")
+					}
+					if err != nil {
+						break
+					}
+				}
+				if total != f.want {
+					t.Fatalf("total bytes read = %v, want %v", total, f.want)
+				}
+				for i := 0; i < total; i++ {
+					if got[i] != want[i] {
+						t.Fatalf("byte %v differs: got %v, want %v", i, got[i], want[i])
+					}
+				}
+			}
+		})
+	}
+
+}
+
+func TestStreamReceiveExtendsStreamWindow(t *testing.T) {
+	testStreamTypes(t, "", func(t *testing.T, styp streamType) {
+		const maxWindowSize = 20
+		ctx := canceledContext()
+		tc := newTestConn(t, serverSide, func(c *Config) {
+			c.StreamReadBufferSize = maxWindowSize
+		})
+		tc.handshake()
+		tc.ignoreFrame(frameTypeAck)
+		sid := newStreamID(clientSide, styp, 0)
+		tc.writeFrames(packetType1RTT, debugFrameStream{
+			id:   sid,
+			off:  0,
+			data: make([]byte, maxWindowSize),
+		})
+		s, err := tc.conn.AcceptStream(ctx)
+		if err != nil {
+			t.Fatalf("AcceptStream: %v", err)
+		}
+		tc.wantIdle("stream window is not extended before data is read")
+		buf := make([]byte, maxWindowSize+1)
+		if n, err := s.ReadContext(ctx, buf); n != maxWindowSize || err != nil {
+			t.Fatalf("s.ReadContext() = %v, %v; want %v, nil", n, err, maxWindowSize)
+		}
+		tc.wantFrame("stream window is extended after reading data",
+			packetType1RTT, debugFrameMaxStreamData{
+				id:  sid,
+				max: maxWindowSize * 2,
+			})
+		tc.writeFrames(packetType1RTT, debugFrameStream{
+			id:   sid,
+			off:  maxWindowSize,
+			data: make([]byte, maxWindowSize),
+			fin:  true,
+		})
+		if n, err := s.ReadContext(ctx, buf); n != maxWindowSize || err != io.EOF {
+			t.Fatalf("s.ReadContext() = %v, %v; want %v, io.EOF", n, err, maxWindowSize)
+		}
+		tc.wantIdle("stream window is not extended after FIN")
+	})
+}
+
+func TestStreamReceiveViolatesStreamDataLimit(t *testing.T) {
+	// "A receiver MUST close the connection with an error of type FLOW_CONTROL_ERROR if
+	// the sender violates the advertised [...] stream data limits [...]"
+	// https://www.rfc-editor.org/rfc/rfc9000#section-4.1-8
+	testStreamTypes(t, "", func(t *testing.T, styp streamType) {
+		const maxStreamData = 10
+		for _, test := range []struct {
+			off  int64
+			size int64
+		}{{
+			off:  maxStreamData,
+			size: 1,
+		}, {
+			off:  0,
+			size: maxStreamData + 1,
+		}, {
+			off:  maxStreamData - 1,
+			size: 2,
+		}} {
+			tc := newTestConn(t, serverSide, func(c *Config) {
+				c.StreamReadBufferSize = maxStreamData
+			})
+			tc.handshake()
+			tc.ignoreFrame(frameTypeAck)
+			tc.writeFrames(packetType1RTT, debugFrameStream{
+				id:   newStreamID(clientSide, styp, 0),
+				off:  test.off,
+				data: make([]byte, test.size),
+			})
+			tc.wantFrame(
+				fmt.Sprintf("data [%v,%v) violates stream data limit and closes connection",
+					test.off, test.off+test.size),
+				packetType1RTT, debugFrameConnectionCloseTransport{
+					code: errFlowControl,
+				},
+			)
+		}
+	})
+}
+
+func TestStreamReceiveDuplicateDataDoesNotViolateLimits(t *testing.T) {
+	testStreamTypes(t, "", func(t *testing.T, styp streamType) {
+		const maxData = 10
+		tc := newTestConn(t, serverSide, func(c *Config) {
+			// TODO: Add connection-level maximum data here as well.
+			c.StreamReadBufferSize = maxData
+		})
+		tc.handshake()
+		tc.ignoreFrame(frameTypeAck)
+		for i := 0; i < 3; i++ {
+			tc.writeFrames(packetType1RTT, debugFrameStream{
+				id:   newStreamID(clientSide, styp, 0),
+				off:  0,
+				data: make([]byte, maxData),
+			})
+			tc.wantIdle(fmt.Sprintf("conn sends no frames after receiving data frame %v", i))
+		}
+	})
+}
+
+func TestStreamFinalSizeChangedByStreamFrame(t *testing.T) {
+	// "If a [...] STREAM frame is received indicating a change
+	// in the final size for the stream, an endpoint SHOULD
+	// respond with an error of type FINAL_SIZE_ERROR [...]"
+	// https://www.rfc-editor.org/rfc/rfc9000#section-4.5-5
+	testStreamTypes(t, "", func(t *testing.T, styp streamType) {
+		tc := newTestConn(t, serverSide)
+		tc.handshake()
+		sid := newStreamID(clientSide, styp, 0)
+
+		const write1size = 4
+		tc.writeFrames(packetType1RTT, debugFrameStream{
+			id:  sid,
+			off: 10,
+			fin: true,
+		})
+		tc.writeFrames(packetType1RTT, debugFrameStream{
+			id:  sid,
+			off: 9,
+			fin: true,
+		})
+		tc.wantFrame("change in final size of stream is an error",
+			packetType1RTT, debugFrameConnectionCloseTransport{
+				code: errFinalSize,
+			},
+		)
+	})
+}
+
+func TestStreamDataBeyondFinalSize(t *testing.T) {
+	// "A receiver SHOULD treat receipt of data at or beyond
+	// the final size as an error of type FINAL_SIZE_ERROR [...]"
+	// https://www.rfc-editor.org/rfc/rfc9000#section-4.5-5
+	testStreamTypes(t, "", func(t *testing.T, styp streamType) {
+		tc := newTestConn(t, serverSide)
+		tc.handshake()
+		sid := newStreamID(clientSide, styp, 0)
+
+		const write1size = 4
+		tc.writeFrames(packetType1RTT, debugFrameStream{
+			id:   sid,
+			off:  0,
+			data: make([]byte, 16),
+			fin:  true,
+		})
+		tc.writeFrames(packetType1RTT, debugFrameStream{
+			id:   sid,
+			off:  16,
+			data: []byte{0},
+		})
+		tc.wantFrame("received data past final size of stream",
+			packetType1RTT, debugFrameConnectionCloseTransport{
+				code: errFinalSize,
+			},
+		)
+	})
+}
+
+func TestStreamReceiveUnblocksReader(t *testing.T) {
+	testStreamTypes(t, "", func(t *testing.T, styp streamType) {
+		tc := newTestConn(t, serverSide)
+		tc.handshake()
+		want := []byte{0, 1, 2, 3, 4, 5, 6, 7, 8, 9}
+		sid := newStreamID(clientSide, styp, 0)
+
+		// AcceptStream blocks until a STREAM frame is received.
+		accept := runAsync(tc, func(ctx context.Context) (*Stream, error) {
+			return tc.conn.AcceptStream(ctx)
+		})
+		const write1size = 4
+		tc.writeFrames(packetType1RTT, debugFrameStream{
+			id:   sid,
+			off:  0,
+			data: want[:write1size],
+		})
+		s, err := accept.result()
+		if err != nil {
+			t.Fatalf("AcceptStream() = %v", err)
+		}
+
+		// ReadContext succeeds immediately, since we already have data.
+		got := make([]byte, len(want))
+		read := runAsync(tc, func(ctx context.Context) (int, error) {
+			return s.ReadContext(ctx, got)
+		})
+		if n, err := read.result(); n != write1size || err != nil {
+			t.Fatalf("ReadContext = %v, %v; want %v, nil", n, err, write1size)
+		}
+
+		// ReadContext blocks waiting for more data.
+		read = runAsync(tc, func(ctx context.Context) (int, error) {
+			return s.ReadContext(ctx, got[write1size:])
+		})
+		tc.writeFrames(packetType1RTT, debugFrameStream{
+			id:   sid,
+			off:  write1size,
+			data: want[write1size:],
+			fin:  true,
+		})
+		if n, err := read.result(); n != len(want)-write1size || err != io.EOF {
+			t.Fatalf("ReadContext = %v, %v; want %v, io.EOF", n, err, len(want)-write1size)
+		}
+		if !bytes.Equal(got, want) {
+			t.Fatalf("read bytes %x, want %x", got, want)
+		}
+	})
+}
+
+// testStreamSendFrameInvalidState calls the test func with a stream ID for:
+//
+//   - a remote bidirectional stream that the peer has not created
+//   - a remote unidirectional stream
+//
+// It then sends the returned frame (STREAM, STREAM_DATA_BLOCKED, etc.)
+// to the conn and expects a STREAM_STATE_ERROR.
+func testStreamSendFrameInvalidState(t *testing.T, f func(sid streamID) debugFrame) {
+	testSides(t, "stream_not_created", func(t *testing.T, side connSide) {
+		tc := newTestConn(t, side)
+		tc.handshake()
+		tc.writeFrames(packetType1RTT, f(newStreamID(side, bidiStream, 0)))
+		tc.wantFrame("frame for local stream which has not been created",
+			packetType1RTT, debugFrameConnectionCloseTransport{
+				code: errStreamState,
+			})
+	})
+	testSides(t, "uni_stream", func(t *testing.T, side connSide) {
+		ctx := canceledContext()
+		tc := newTestConn(t, side)
+		tc.handshake()
+		sid := newStreamID(side, uniStream, 0)
+		s, err := tc.conn.NewSendOnlyStream(ctx)
+		if err != nil {
+			t.Fatal(err)
+		}
+		s.Write(nil) // open the stream
+		tc.wantFrame("new stream is opened",
+			packetType1RTT, debugFrameStream{
+				id:   sid,
+				data: []byte{},
+			})
+		tc.writeFrames(packetType1RTT, f(sid))
+		tc.wantFrame("send-oriented frame for send-only stream",
+			packetType1RTT, debugFrameConnectionCloseTransport{
+				code: errStreamState,
+			})
+	})
+}
+
+func TestStreamStreamFrameInvalidState(t *testing.T) {
+	// "An endpoint MUST terminate the connection with error STREAM_STATE_ERROR
+	// if it receives a STREAM frame for a locally initiated stream
+	// that has not yet been created, or for a send-only stream."
+	// https://www.rfc-editor.org/rfc/rfc9000.html#section-19.8-3
+	testStreamSendFrameInvalidState(t, func(sid streamID) debugFrame {
+		return debugFrameStream{
+			id: sid,
+		}
+	})
+}
+
+func TestStreamDataBlockedInvalidState(t *testing.T) {
+	// "An endpoint MUST terminate the connection with error STREAM_STATE_ERROR
+	// if it receives a STREAM frame for a locally initiated stream
+	// that has not yet been created, or for a send-only stream."
+	// https://www.rfc-editor.org/rfc/rfc9000.html#section-19.8-3
+	testStreamSendFrameInvalidState(t, func(sid streamID) debugFrame {
+		return debugFrameStream{
+			id: sid,
+		}
+	})
+}
+
+// testStreamReceiveFrameInvalidState calls the test func with a stream ID for:
+//
+//   - a remote bidirectional stream that the peer has not created
+//   - a local unidirectional stream
+//
+// It then sends the returned frame (MAX_STREAM_DATA, STOP_SENDING, etc.)
+// to the conn and expects a STREAM_STATE_ERROR.
+func testStreamReceiveFrameInvalidState(t *testing.T, f func(sid streamID) debugFrame) {
+	testSides(t, "stream_not_created", func(t *testing.T, side connSide) {
+		tc := newTestConn(t, side)
+		tc.handshake()
+		tc.writeFrames(packetType1RTT, f(newStreamID(side, bidiStream, 0)))
+		tc.wantFrame("frame for local stream which has not been created",
+			packetType1RTT, debugFrameConnectionCloseTransport{
+				code: errStreamState,
+			})
+	})
+	testSides(t, "uni_stream", func(t *testing.T, side connSide) {
+		tc := newTestConn(t, side)
+		tc.handshake()
+		tc.writeFrames(packetType1RTT, f(newStreamID(side.peer(), uniStream, 0)))
+		tc.wantFrame("receive-oriented frame for receive-only stream",
+			packetType1RTT, debugFrameConnectionCloseTransport{
+				code: errStreamState,
+			})
+	})
+}
+
+func TestStreamMaxStreamDataInvalidState(t *testing.T) {
+	// "Receiving a MAX_STREAM_DATA frame for a locally initiated stream
+	// that has not yet been created MUST be treated as a connection error
+	// of type STREAM_STATE_ERROR. An endpoint that receives a MAX_STREAM_DATA
+	// frame for a receive-only stream MUST terminate the connection
+	// with error STREAM_STATE_ERROR."
+	// https://www.rfc-editor.org/rfc/rfc9000#section-19.10-2
+	testStreamReceiveFrameInvalidState(t, func(sid streamID) debugFrame {
+		return debugFrameMaxStreamData{
+			id:  sid,
+			max: 1000,
+		}
+	})
+}
+
 func TestStreamOffsetTooLarge(t *testing.T) {
 	// "Receipt of a frame that exceeds [2^62-1] MUST be treated as a
 	// connection error of type FRAME_ENCODING_ERROR or FLOW_CONTROL_ERROR."
@@ -31,3 +724,104 @@
 		t.Fatalf("STREAM offset exceeds 2^62-1\ngot:  %v\nwant: %v\n  or: %v", got, want1, want2)
 	}
 }
+
+func TestStreamReadFromWriteOnlyStream(t *testing.T) {
+	_, s := newTestConnAndLocalStream(t, serverSide, uniStream)
+	buf := make([]byte, 10)
+	wantErr := "read from write-only stream"
+	if n, err := s.Read(buf); err == nil || !strings.Contains(err.Error(), wantErr) {
+		t.Errorf("s.Read() = %v, %v; want error %q", n, err, wantErr)
+	}
+}
+
+func TestStreamWriteToReadOnlyStream(t *testing.T) {
+	_, s := newTestConnAndRemoteStream(t, serverSide, uniStream)
+	buf := make([]byte, 10)
+	wantErr := "write to read-only stream"
+	if n, err := s.Write(buf); err == nil || !strings.Contains(err.Error(), wantErr) {
+		t.Errorf("s.Write() = %v, %v; want error %q", n, err, wantErr)
+	}
+}
+
+func TestStreamWriteToClosedStream(t *testing.T) {
+	tc, s := newTestConnAndLocalStream(t, serverSide, bidiStream, func(p *transportParameters) {
+		p.initialMaxStreamsBidi = 1
+		p.initialMaxData = 1 << 20
+		p.initialMaxStreamDataBidiRemote = 1 << 20
+	})
+	s.Close()
+	tc.wantFrame("stream is opened after being closed",
+		packetType1RTT, debugFrameStream{
+			id:   s.id,
+			off:  0,
+			fin:  true,
+			data: []byte{},
+		})
+	wantErr := "write to closed stream"
+	if n, err := s.Write([]byte{}); err == nil || !strings.Contains(err.Error(), wantErr) {
+		t.Errorf("s.Write() = %v, %v; want error %q", n, err, wantErr)
+	}
+}
+
+func TestStreamWriteMoreThanOnePacketOfData(t *testing.T) {
+	tc, s := newTestConnAndLocalStream(t, serverSide, uniStream, func(p *transportParameters) {
+		p.initialMaxStreamsUni = 1
+		p.initialMaxData = 1 << 20
+		p.initialMaxStreamDataUni = 1 << 20
+	})
+	want := make([]byte, 4096)
+	rand.Read(want) // doesn't need to be crypto/rand, but non-deprecated and harmless
+	w := runAsync(tc, func(ctx context.Context) (int, error) {
+		return s.WriteContext(ctx, want)
+	})
+	got := make([]byte, 0, len(want))
+	for {
+		f, _ := tc.readFrame()
+		if f == nil {
+			break
+		}
+		sf, ok := f.(debugFrameStream)
+		if !ok {
+			t.Fatalf("unexpected frame: %v", sf)
+		}
+		if len(got) != int(sf.off) {
+			t.Fatalf("got frame: %v\nwant offset %v", sf, len(got))
+		}
+		got = append(got, sf.data...)
+	}
+	if n, err := w.result(); n != len(want) || err != nil {
+		t.Fatalf("s.WriteContext() = %v, %v; want %v, nil", n, err, len(want))
+	}
+	if !bytes.Equal(got, want) {
+		t.Fatalf("mismatch in received stream data")
+	}
+}
+
+func newTestConnAndLocalStream(t *testing.T, side connSide, styp streamType, opts ...any) (*testConn, *Stream) {
+	t.Helper()
+	ctx := canceledContext()
+	tc := newTestConn(t, side, opts...)
+	tc.handshake()
+	tc.ignoreFrame(frameTypeAck)
+	s, err := tc.conn.newLocalStream(ctx, styp)
+	if err != nil {
+		t.Fatalf("conn.newLocalStream(%v) = %v", styp, err)
+	}
+	return tc, s
+}
+
+func newTestConnAndRemoteStream(t *testing.T, side connSide, styp streamType, opts ...any) (*testConn, *Stream) {
+	t.Helper()
+	ctx := canceledContext()
+	tc := newTestConn(t, side, opts...)
+	tc.handshake()
+	tc.ignoreFrame(frameTypeAck)
+	tc.writeFrames(packetType1RTT, debugFrameStream{
+		id: newStreamID(side.peer(), styp, 0),
+	})
+	s, err := tc.conn.AcceptStream(ctx)
+	if err != nil {
+		t.Fatalf("conn.AcceptStream() = %v", err)
+	}
+	return tc, s
+}