aa93a57 fix(brokeripc): keep images under the byte window and share stream chunking · vev

M docs/architecture.md +2 -2

@@ -1,6 +1,6 @@
1	1		# Architecture
2	2
3		-	vev uses a hexagonal core with typed session messages at the client and daemon boundary. Binary framing and carriage selection stay outside the use cases. The current strict protocol version is 65.
	3	+	vev uses a hexagonal core with typed session messages at the client and daemon boundary. Binary framing and carriage selection stay outside the use cases. The current strict protocol version is 66.
4	4
5	5		`CommandResult` has a closed explicit outcome contract:
6	6

@@ -135,7 +135,7 @@ Use cases exchange `protocol.ClientMessage` and `protocol.ServerMessage` values.
135	135		`go tool buf generate`, and handle the new variant in `sessionwire`.
136	136		Never edit generated `*.pb.go`; never add manual IDs or dispatch tables.
137	137		- Implement I/O, queues, workers, environment integration, or technology selection in an adapter or `internal/app`.
138		-	- Bump `internal/protocol.Version` for negotiated wire layout changes (currently `65`). The preamble epoch (`wire.ProtocolEpoch`, QUIC ALPN `vev/1`) bumps only for an intentional clean break.
	138	+	- Bump `internal/protocol.Version` for negotiated wire layout changes (currently `66`). The preamble epoch (`wire.ProtocolEpoch`, QUIC ALPN `vev/1`) bumps only for an intentional clean break.
139	139
140	140		## Broker
141	141

@@ -1,6 +1,6 @@
1		# Architecture	1		# Architecture
2			2
3	-	vev uses a hexagonal core with typed session messages at the client and daemon boundary. Binary framing and carriage selection stay outside the use cases. The current strict protocol version is 65.	3	+	vev uses a hexagonal core with typed session messages at the client and daemon boundary. Binary framing and carriage selection stay outside the use cases. The current strict protocol version is 66.
4			4
5		`CommandResult` has a closed explicit outcome contract:	5		`CommandResult` has a closed explicit outcome contract:
6			6

@@ -135,7 +135,7 @@ Use cases exchange `protocol.ClientMessage` and `protocol.ServerMessage` values.
135		`go tool buf generate`, and handle the new variant in `sessionwire`.	135		`go tool buf generate`, and handle the new variant in `sessionwire`.
136		Never edit generated `*.pb.go`; never add manual IDs or dispatch tables.	136		Never edit generated `*.pb.go`; never add manual IDs or dispatch tables.
137		- Implement I/O, queues, workers, environment integration, or technology selection in an adapter or `internal/app`.	137		- Implement I/O, queues, workers, environment integration, or technology selection in an adapter or `internal/app`.
138	-	- Bump `internal/protocol.Version` for negotiated wire layout changes (currently `65`). The preamble epoch (`wire.ProtocolEpoch`, QUIC ALPN `vev/1`) bumps only for an intentional clean break.	138	+	- Bump `internal/protocol.Version` for negotiated wire layout changes (currently `66`). The preamble epoch (`wire.ProtocolEpoch`, QUIC ALPN `vev/1`) bumps only for an intentional clean break.
139			139
140		## Broker	140		## Broker
141			141

M docs/protocol.md +6 -5


@@ -133,8 +133,9 @@ plus 64 bytes) per data frame out of a 4 MiB window
133	133		(`brokerwire.StreamWindowBytes`) and waits when it has too little, and the
134	134		receiver returns the credit of frames its consumer has read in
135	135		`StreamWindowUpdate` batches. A slow consumer therefore slows its own stream
136		-	instead of losing it; data or credit beyond the window is a flow-control
137		-	violation that settles only that stream. The remote helpers `_broker-mux-stdio`,
	136	+	instead of losing it. Data or cumulative credit beyond the window is a
	137	+	flow-control violation that settles only that stream; a single grant of zero
	138	+	or above the window is a malformed frame and fails the connection. The remote helpers `_broker-mux-stdio`,
138	139		`_broker-mux-quic-bootstrap`, and `_broker-mux-quic-proxy` bridge a remote
139	140		daemonmux socket to stdio or to one freshly authenticated QUIC stream.
140	141

@@ -173,8 +174,8 @@ base. `ValidateOutput` enforces the epoch/base/new chain
173	174		(2 MiB). State frames are charged to their own state, side effects to the
174	175		next state, and a cumulative ACK releases what it covers. The next state
175	176		frame waits when either window is full, except that a frame may always be
176		-	sent while no state frame is in flight. Images share the remaining byte
177		-	budget and wait for a later frame when it is too small.
	177	+	sent while no state frame is in flight. A frame carrying a large image may
	178	+	exceed the byte window; it then holds the next frame until its ACK.
178	179		- The client coalesces ACKs (`client.go:cumulativeAckQueue`) and keeps
179	180		applying + acknowledging daemon frames even while a picker lease owns
180	181		the terminal (see below) — admission without physical write.

@@ -261,7 +262,7 @@ than `KillResult` and `CommandResult`.
261	262		\| Encoding buffer \| bounded per-connection framing buffers owned by `streamframe` \| `adapters/streamframe` \|
262	263		\| Output states in flight \| 8 (`MaxOutputWindow`) preferred for QUIC/IPC/SSH \| daemon, negotiated down \|
263	264		\| Unacknowledged Output bytes \| 2 MiB (`MaxOutputWindowBytes`); one frame always allowed when none is in flight \| daemon \|
264		-	\| Broker stream window \| 4 MiB per stream and direction (`brokerwire.StreamWindowBytes`), returned by `StreamWindowUpdate` \| `adapters/brokeripc` \|
	265	+	\| Broker stream window \| 4 MiB per stream and direction (`brokerwire.StreamWindowBytes`), returned by `StreamWindowUpdate` \| `adapters/brokerwire`, enforced in `adapters/brokeripc` \|
265	266		\| QUIC unauthenticated peers \| 4 connections, 3 s auth deadline, one bidirectional stream each; extra/unidirectional streams rejected \| `adapters/quic` \|
266	267		\| QUIC bootstrap \| 32-byte token + 16-byte nonce (base64), ≤4 KiB readiness (schema v1, host-independent port, SHA-256 pin, ≤15 s expiry), one bounded (512 B) auth record, atomic one-time consumption, token erasure on close \| `adapters/quic` bootstrap, `app/_broker-mux-quic-bootstrap` + `_broker-mux-quic-proxy` \|
267	268		\| Handshake, dial → first committed publication \| 15 s, single absolute deadline propagated unchanged, never reset \| both ends \|

 (`brokerwire.StreamWindowBytes`) and waits when it has too little, and the
 receiver returns the credit of frames its consumer has read in
 `StreamWindowUpdate` batches. A slow consumer therefore slows its own stream
-instead of losing it; data or credit beyond the window is a flow-control
+instead of losing it. Data or cumulative credit beyond the window is a
-violation that settles only that stream. The remote helpers `_broker-mux-stdio`,
+flow-control violation that settles only that stream; a single grant of zero
+or above the window is a malformed frame and fails the connection. The remote helpers `_broker-mux-stdio`,
 `_broker-mux-quic-bootstrap`, and `_broker-mux-quic-proxy` bridge a remote
 daemonmux socket to stdio or to one freshly authenticated QUIC stream.
   (2 MiB). State frames are charged to their own state, side effects to the
   next state, and a cumulative ACK releases what it covers. The next state
   frame waits when either window is full, except that a frame may always be
-  sent while no state frame is in flight. Images share the remaining byte
+  sent while no state frame is in flight. A frame carrying a large image may
-  budget and wait for a later frame when it is too small.
+  exceed the byte window; it then holds the next frame until its ACK.
 - The client coalesces ACKs (`client.go:cumulativeAckQueue`) and keeps
   applying + acknowledging daemon frames even while a picker lease owns
   the terminal (see below) — admission without physical write.
 | Encoding buffer | bounded per-connection framing buffers owned by `streamframe` | `adapters/streamframe` |
 | Output states in flight | 8 (`MaxOutputWindow`) preferred for QUIC/IPC/SSH | daemon, negotiated down |
 | Unacknowledged Output bytes | 2 MiB (`MaxOutputWindowBytes`); one frame always allowed when none is in flight | daemon |
-| Broker stream window | 4 MiB per stream and direction (`brokerwire.StreamWindowBytes`), returned by `StreamWindowUpdate` | `adapters/brokeripc` |
+| Broker stream window | 4 MiB per stream and direction (`brokerwire.StreamWindowBytes`), returned by `StreamWindowUpdate` | `adapters/brokerwire`, enforced in `adapters/brokeripc` |
 | QUIC unauthenticated peers | 4 connections, 3 s auth deadline, one bidirectional stream each; extra/unidirectional streams rejected | `adapters/quic` |
 | QUIC bootstrap | 32-byte token + 16-byte nonce (base64), ≤4 KiB readiness (schema v1, host-independent port, SHA-256 pin, ≤15 s expiry), one bounded (512 B) auth record, atomic one-time consumption, token erasure on close | `adapters/quic` bootstrap, `app/_broker-mux-quic-bootstrap` + `_broker-mux-quic-proxy` |
 | Handshake, dial → first committed publication | 15 s, single absolute deadline propagated unchanged, never reset | both ends |

M internal/adapters/brokeripc/client.go +23 -22


@@ -351,18 +351,9 @@ func (c *client) dispatch(message brokerwire.ServerMessage) error {
351	351		if !c.scopeMatches(m.Epoch, m.Connection) {
352	352		return errors.Join(ErrScopeMismatch, ErrProtocol)
353	353		}
354		-	disposition, err := c.conn.StreamData(m.Stream)
355		-	if err != nil {
356		-	return errors.Join(ErrProtocol, err)
357		-	}
358		-	if disposition == brokerwire.StreamDiscarded {
359		-	return nil
360		-	}
361		-	st := c.lookupStream(m.Stream)
362		-	if st == nil {
363		-	// The stream was retired locally between the tracker check and the
364		-	// lookup; its data is discarded, exactly like a retired stream.
365		-	return nil
	354	+	st, err := c.streamFor(m.Stream)
	355	+	if st == nil \|\| err != nil {
	356	+	return err
366	357		}
367	358		if err := st.deliver(m.Data); errors.Is(err, ErrStreamCredit) {
368	359		st.fail(ErrStreamCredit)

@@ -372,17 +363,12 @@ func (c *client) dispatch(message brokerwire.ServerMessage) error {
372	363		if !c.scopeMatches(m.Epoch, m.Connection) {
373	364		return errors.Join(ErrScopeMismatch, ErrProtocol)
374	365		}
375		-	disposition, err := c.conn.StreamWindowUpdate(m.Stream)
376		-	if err != nil {
377		-	return errors.Join(ErrProtocol, err)
378		-	}
379		-	if disposition == brokerwire.StreamDiscarded {
380		-	return nil
	366	+	st, err := c.streamFor(m.Stream)
	367	+	if st == nil \|\| err != nil {
	368	+	return err
381	369		}
382		-	if st := c.lookupStream(m.Stream); st != nil {
383		-	if err := st.pipe.granted(m.Credit); err != nil {
384		-	st.fail(err)
385		-	}
	370	+	if err := st.pipe.granted(m.Credit); err != nil {
	371	+	st.fail(err)
386	372		}
387	373		return nil
388	374		case brokerwire.StreamClosed:

@@ -941,6 +927,21 @@ func (c client) lookupStream(stream ports.BrokerStreamID) clientStream {
941	927		return c.streams[stream]
942	928		}
943	929
	930	+	// streamFor classifies one stream-scoped frame (data or a credit grant) and
	931	+	// returns its live stream. A frame for a retired stream, or one retired locally
	932	+	// between the tracker check and the lookup, yields no stream and no error; a
	933	+	// frame for a stream that was never opened is a peer protocol violation.
	934	+	func (c client) streamFor(id ports.BrokerStreamID) (clientStream, error) {
	935	+	disposition, err := c.conn.StreamData(id)
	936	+	if err != nil {
	937	+	return nil, errors.Join(ErrProtocol, err)
	938	+	}
	939	+	if disposition == brokerwire.StreamDiscarded {
	940	+	return nil, nil
	941	+	}
	942	+	return c.lookupStream(id), nil
	943	+	}
	944	+
944	945		// unregisterStream drops one local stream record.
945	946		func (c *client) unregisterStream(stream ports.BrokerStreamID) {
946	947		c.mu.Lock()

 		if !c.scopeMatches(m.Epoch, m.Connection) {
 			return errors.Join(ErrScopeMismatch, ErrProtocol)
 		}
-		disposition, err := c.conn.StreamData(m.Stream)
+		st, err := c.streamFor(m.Stream)
-		if err != nil {
+		if st == nil || err != nil {
-			return errors.Join(ErrProtocol, err)
+			return err
-		}
-		if disposition == brokerwire.StreamDiscarded {
-			return nil
-		}
-		st := c.lookupStream(m.Stream)
-		if st == nil {
-			// The stream was retired locally between the tracker check and the
-			// lookup; its data is discarded, exactly like a retired stream.
-			return nil
 		}
 		if err := st.deliver(m.Data); errors.Is(err, ErrStreamCredit) {
 			st.fail(ErrStreamCredit)
 		if !c.scopeMatches(m.Epoch, m.Connection) {
 			return errors.Join(ErrScopeMismatch, ErrProtocol)
 		}
-		disposition, err := c.conn.StreamWindowUpdate(m.Stream)
+		st, err := c.streamFor(m.Stream)
-		if err != nil {
+		if st == nil || err != nil {
-			return errors.Join(ErrProtocol, err)
+			return err
-		}
-		if disposition == brokerwire.StreamDiscarded {
-			return nil
 		}
-		if st := c.lookupStream(m.Stream); st != nil {
+		if err := st.pipe.granted(m.Credit); err != nil {
-			if err := st.pipe.granted(m.Credit); err != nil {
+			st.fail(err)
-				st.fail(err)
-			}
 		}
 		return nil
 	case brokerwire.StreamClosed:
 	return c.streams[stream]
 }
+// streamFor classifies one stream-scoped frame (data or a credit grant) and
+// returns its live stream. A frame for a retired stream, or one retired locally
+// between the tracker check and the lookup, yields no stream and no error; a
+// frame for a stream that was never opened is a peer protocol violation.
+func (c *client) streamFor(id ports.BrokerStreamID) (*clientStream, error) {
+	disposition, err := c.conn.StreamData(id)
+	if err != nil {
+		return nil, errors.Join(ErrProtocol, err)
+	}
+	if disposition == brokerwire.StreamDiscarded {
+		return nil, nil
+	}
+	return c.lookupStream(id), nil
+}
 // unregisterStream drops one local stream record.
 func (c *client) unregisterStream(stream ports.BrokerStreamID) {
 	c.mu.Lock()

M internal/adapters/brokeripc/limits_test.go +1 -1


@@ -177,7 +177,7 @@ func TestStreamPipeCloseReleasesParkedReader(t *testing.T) {
177	177		// silently dropping bytes.
178	178		func TestStreamPipeOutboundError(t *testing.T) {
179	179		failure := errors.New("outbound failed")
180		-	pipe := newStreamPipe(func([]byte) error { return failure }, func(uint64) error { return nil })
	180	+	pipe := newStreamPipe(int(brokerwire.MaxStreamChunkBytes), func([]byte) error { return failure }, func(uint64) error { return nil })
181	181		_, err := pipe.Write([]byte("payload"))
182	182		require.ErrorIs(t, err, failure)
183	183		require.NoError(t, pipe.Close())


@@ -177,7 +177,7 @@ func TestStreamPipeCloseReleasesParkedReader(t *testing.T) {
177		// silently dropping bytes.	177		// silently dropping bytes.
178		func TestStreamPipeOutboundError(t *testing.T) {	178		func TestStreamPipeOutboundError(t *testing.T) {
179		failure := errors.New("outbound failed")	179		failure := errors.New("outbound failed")
180	-	pipe := newStreamPipe(func([]byte) error { return failure }, func(uint64) error { return nil })	180	+	pipe := newStreamPipe(int(brokerwire.MaxStreamChunkBytes), func([]byte) error { return failure }, func(uint64) error { return nil })
181		_, err := pipe.Write([]byte("payload"))	181		_, err := pipe.Write([]byte("payload"))
182		require.ErrorIs(t, err, failure)	182		require.ErrorIs(t, err, failure)
183		require.NoError(t, pipe.Close())	183		require.NoError(t, pipe.Close())

M internal/adapters/brokeripc/pipe.go +29 -13


@@ -38,8 +38,9 @@ type streamPipe struct {
38	38		wake chan struct{}
39	39		closedCh chan struct{}
40	40
41		-	outbound func([]byte) error
42		-	grant func(uint64) error
	41	+	ceiling int
	42	+	sendFrame func([]byte) error
	43	+	grant func(uint64) error
43	44		}
44	45
45	46		// pipeChunk is one queued inbound frame and the credit it holds until it is

@@ -49,9 +50,11 @@ type pipeChunk struct {
49	50		cost uint64
50	51		}
51	52
52		-	// newStreamPipe builds one stream carriage. outbound writes one outbound byte
53		-	// slice; grant returns consumed inbound credit to the peer. Neither may be nil.
54		-	func newStreamPipe(outbound func([]byte) error, grant func(uint64) error) *streamPipe {
	53	+	// newStreamPipe builds one stream carriage. Write splits outbound bytes into
	54	+	// frames of at most ceiling bytes, the negotiated stream chunk limit, and hands
	55	+	// each to sendFrame once its credit is reserved; grant returns consumed inbound
	56	+	// credit to the peer. Neither hook may be nil.
	57	+	func newStreamPipe(ceiling int, sendFrame func([]byte) error, grant func(uint64) error) *streamPipe {
55	58		window := brokerwire.StreamWindowBytes
56	59		return &streamPipe{
57	60		returnAt: streamCreditReturnAt(window),

@@ -59,7 +62,8 @@ func newStreamPipe(outbound func([]byte) error, grant func(uint64) error) *strea
59	62		creditWake: make(chan struct{}),
60	63		wake: make(chan struct{}, 1),
61	64		closedCh: make(chan struct{}),
62		-	outbound: outbound,
	65	+	ceiling: ceiling,
	66	+	sendFrame: sendFrame,
63	67		grant: grant,
64	68		}
65	69		}

@@ -122,18 +126,29 @@ func (p *streamPipe) Read(b []byte) (int, error) {
122	126		}
123	127		}
124	128
125		-	// Write hands one byte slice to the outbound hook, which spends send credit
126		-	// per frame through reserve. It returns ErrStreamGone once the pipe reached its
127		-	// terminal outcome.
	129	+	// Write sends b in order as frames under the chunk ceiling, waiting for the
	130	+	// peer's credit before each one: a peer that is not reading slows this writer
	131	+	// instead of losing the stream. It returns ErrStreamGone once the pipe reached
	132	+	// its terminal outcome.
128	133		func (p *streamPipe) Write(b []byte) (int, error) {
	134	+	if p.ceiling <= 0 {
	135	+	return 0, ErrConfig
	136	+	}
129	137		p.mu.Lock()
130	138		closed := p.closed
131	139		p.mu.Unlock()
132	140		if closed {
133	141		return 0, ErrStreamGone
134	142		}
135		-	if err := p.outbound(b); err != nil {
136		-	return 0, err
	143	+	for data := b; len(data) > 0; {
	144	+	n := min(len(data), p.ceiling)
	145	+	if err := p.reserve(n); err != nil {
	146	+	return 0, err
	147	+	}
	148	+	if err := p.sendFrame(data[:n]); err != nil {
	149	+	return 0, err
	150	+	}
	151	+	data = data[n:]
137	152		}
138	153		return len(b), nil
139	154		}

@@ -183,7 +198,8 @@ func (p *streamPipe) granted(credit uint64) error {
183	198		// deliver accepts one inbound frame from the connection reader. It returns
184	199		// ErrStreamCredit when the peer sent beyond the credit it was granted and
185	200		// ErrStreamGone once the pipe is closed; the caller settles the affected stream
186		-	// in both cases.
	201	+	// in both cases. The pipe takes ownership of chunk: callers pass freshly
	202	+	// decoded bytes and never touch them again.
187	203		func (p *streamPipe) deliver(chunk []byte) error {
188	204		if len(chunk) == 0 {
189	205		return nil

@@ -198,7 +214,7 @@ func (p *streamPipe) deliver(chunk []byte) error {
198	214		p.mu.Unlock()
199	215		return ErrStreamCredit
200	216		}
201		-	p.queue = append(p.queue, pipeChunk{data: append([]byte(nil), chunk...), cost: cost})
	217	+	p.queue = append(p.queue, pipeChunk{data: chunk, cost: cost})
202	218		p.recvUsed += cost
203	219		p.mu.Unlock()
204	220		select {

 	wake     chan struct{}
 	closedCh chan struct{}
-	outbound func([]byte) error
+	ceiling   int
-	grant    func(uint64) error
+	sendFrame func([]byte) error
+	grant     func(uint64) error
 }
 // pipeChunk is one queued inbound frame and the credit it holds until it is
 	cost uint64
 }
-// newStreamPipe builds one stream carriage. outbound writes one outbound byte
+// newStreamPipe builds one stream carriage. Write splits outbound bytes into
-// slice; grant returns consumed inbound credit to the peer. Neither may be nil.
+// frames of at most ceiling bytes, the negotiated stream chunk limit, and hands
-func newStreamPipe(outbound func([]byte) error, grant func(uint64) error) *streamPipe {
+// each to sendFrame once its credit is reserved; grant returns consumed inbound
+// credit to the peer. Neither hook may be nil.
+func newStreamPipe(ceiling int, sendFrame func([]byte) error, grant func(uint64) error) *streamPipe {
 	window := brokerwire.StreamWindowBytes
 	return &streamPipe{
 		returnAt:   streamCreditReturnAt(window),
 		creditWake: make(chan struct{}),
 		wake:       make(chan struct{}, 1),
 		closedCh:   make(chan struct{}),
-		outbound:   outbound,
+		ceiling:    ceiling,
+		sendFrame:  sendFrame,
 		grant:      grant,
 	}
 }
 	}
 }
-// Write hands one byte slice to the outbound hook, which spends send credit
+// Write sends b in order as frames under the chunk ceiling, waiting for the
-// per frame through reserve. It returns ErrStreamGone once the pipe reached its
+// peer's credit before each one: a peer that is not reading slows this writer
-// terminal outcome.
+// instead of losing the stream. It returns ErrStreamGone once the pipe reached
+// its terminal outcome.
 func (p *streamPipe) Write(b []byte) (int, error) {
+	if p.ceiling <= 0 {
+		return 0, ErrConfig
+	}
 	p.mu.Lock()
 	closed := p.closed
 	p.mu.Unlock()
 	if closed {
 		return 0, ErrStreamGone
 	}
-	if err := p.outbound(b); err != nil {
+	for data := b; len(data) > 0; {
-		return 0, err
+		n := min(len(data), p.ceiling)
+		if err := p.reserve(n); err != nil {
+			return 0, err
+		}
+		if err := p.sendFrame(data[:n]); err != nil {
+			return 0, err
+		}
+		data = data[n:]
 	}
 	return len(b), nil
 }
 // deliver accepts one inbound frame from the connection reader. It returns
 // ErrStreamCredit when the peer sent beyond the credit it was granted and
 // ErrStreamGone once the pipe is closed; the caller settles the affected stream
-// in both cases.
+// in both cases. The pipe takes ownership of chunk: callers pass freshly
+// decoded bytes and never touch them again.
 func (p *streamPipe) deliver(chunk []byte) error {
 	if len(chunk) == 0 {
 		return nil
 		p.mu.Unlock()
 		return ErrStreamCredit
 	}
-	p.queue = append(p.queue, pipeChunk{data: append([]byte(nil), chunk...), cost: cost})
+	p.queue = append(p.queue, pipeChunk{data: chunk, cost: cost})
 	p.recvUsed += cost
 	p.mu.Unlock()
 	select {

M internal/adapters/brokeripc/pipe_test.go +8 -7


@@ -112,15 +112,16 @@ func TestStreamPipeCloseWithPreservesQueuedDataOnOrderlyClose(t *testing.T) {
112	112
113	113		// newTestPipe is a pipe whose outbound writes and credit grants go nowhere.
114	114		func newTestPipe() *streamPipe {
115		-	return newStreamPipe(func([]byte) error { return nil }, func(uint64) error { return nil })
	115	+	return newStreamPipe(int(brokerwire.MaxStreamChunkBytes), func([]byte) error { return nil }, func(uint64) error { return nil })
116	116		}
117	117
118		-	// TestStreamWindowCoversTheSessionOutputWindow pins the sizing invariant
	118	+	// TestStreamWindowCoversTheSessionOutputWindow pins the sizing relation
119	119		// between the two flow-control layers: one broker stream window holds the
120		-	// daemon's whole unacknowledged byte budget, so in normal operation the daemon
121		-	// is paced by its own ACK window and coalesces frames, rather than blocking a
122		-	// send on stream credit. A single Output larger than the window (a big image)
123		-	// still crosses, only paced by credit.
	120	+	// daemon's unacknowledged byte budget, so the daemon is mostly paced by its own
	121	+	// ACK window and coalesces frames. Credit may still pace it at the margin: the
	122	+	// receiver batches up to half a window before returning it, chunks cost 64
	123	+	// bytes of overhead, and one frame may overrun the byte budget (a big image).
	124	+	// Credit stays the hard bound either way.
124	125		func TestStreamWindowCoversTheSessionOutputWindow(t *testing.T) {
125	126		require.GreaterOrEqual(t, brokerwire.StreamWindowBytes, uint64(protocol.MaxOutputWindowBytes),
126	127		"one stream window must hold the daemon's whole unacknowledged byte budget")

@@ -145,7 +146,7 @@ func TestStreamPipeCredit(t *testing.T) {
145	146
146	147		t.Run("reads return credit in batches", func(t *testing.T) {
147	148		var grants []uint64
148		-	p := newStreamPipe(func([]byte) error { return nil }, func(c uint64) error { grants = append(grants, c); return nil })
	149	+	p := newStreamPipe(int(brokerwire.MaxStreamChunkBytes), func([]byte) error { return nil }, func(c uint64) error { grants = append(grants, c); return nil })
149	150		for range perWindow {
150	151		require.NoError(t, p.deliver(make([]byte, chunk)))
151	152		}

 // newTestPipe is a pipe whose outbound writes and credit grants go nowhere.
 func newTestPipe() *streamPipe {
-	return newStreamPipe(func([]byte) error { return nil }, func(uint64) error { return nil })
+	return newStreamPipe(int(brokerwire.MaxStreamChunkBytes), func([]byte) error { return nil }, func(uint64) error { return nil })
 }
-// TestStreamWindowCoversTheSessionOutputWindow pins the sizing invariant
+// TestStreamWindowCoversTheSessionOutputWindow pins the sizing relation
 // between the two flow-control layers: one broker stream window holds the
-// daemon's whole unacknowledged byte budget, so in normal operation the daemon
+// daemon's unacknowledged byte budget, so the daemon is mostly paced by its own
-// is paced by its own ACK window and coalesces frames, rather than blocking a
+// ACK window and coalesces frames. Credit may still pace it at the margin: the
-// send on stream credit. A single Output larger than the window (a big image)
+// receiver batches up to half a window before returning it, chunks cost 64
-// still crosses, only paced by credit.
+// bytes of overhead, and one frame may overrun the byte budget (a big image).
+// Credit stays the hard bound either way.
 func TestStreamWindowCoversTheSessionOutputWindow(t *testing.T) {
 	require.GreaterOrEqual(t, brokerwire.StreamWindowBytes, uint64(protocol.MaxOutputWindowBytes),
 		"one stream window must hold the daemon's whole unacknowledged byte budget")
 	t.Run("reads return credit in batches", func(t *testing.T) {
 		var grants []uint64
-		p := newStreamPipe(func([]byte) error { return nil }, func(c uint64) error { grants = append(grants, c); return nil })
+		p := newStreamPipe(int(brokerwire.MaxStreamChunkBytes), func([]byte) error { return nil }, func(c uint64) error { grants = append(grants, c); return nil })
 		for range perWindow {
 			require.NoError(t, p.deliver(make([]byte, chunk)))
 		}

M internal/adapters/brokeripc/session_test.go +18 -9

@@ -1,6 +1,7 @@
1	1		package brokeripc
2	2
3	3		import (
	4	+	"bytes"
4	5		"context"
5	6		"errors"
6	7		"io"

@@ -883,13 +884,9 @@ func TestStalledClientStreamWaitsForCredit(t *testing.T) {
883	884		}
884	885
885	886		// Well over one stream window of daemon output while nothing reads. Side
886		-	// effect Outputs need no view context; random-looking data defeats
887		-	// compression so every byte crosses the stream.
	887	+	// effect Outputs need no view context and are never compressed.
888	888		const messages = 24
889		-	data := make([]byte, 512<<10)
890		-	for i := range data {
891		-	data[i] = byte(i*131 + i/7)
892		-	}
	889	+	data := bytes.Repeat([]byte{'x'}, 512<<10)
893	890		go func() {
894	891		for i := range messages {
895	892		payload := append([]byte{byte(i)}, data...)

@@ -901,12 +898,24 @@ func TestStalledClientStreamWaitsForCredit(t *testing.T) {
901	898		}
902	899		}()
903	900
904		-	// The relay is blocked on credit, not failed: the stream stays open and the
905		-	// connection keeps serving other work.
	901	+	// The relay fills exactly one window of the client pipe and then waits on
	902	+	// credit instead of failing: the stream stays open and the connection keeps
	903	+	// serving other work.
	904	+	local, ok := stream.(*clientStream)
	905	+	require.True(t, ok)
	906	+	full := brokerwire.StreamWindowBytes - brokerwire.StreamChunkCredit(int(brokerwire.MaxStreamChunkBytes))
	907	+	held := func() uint64 {
	908	+	local.pipe.mu.Lock()
	909	+	defer local.pipe.mu.Unlock()
	910	+	return local.pipe.recvUsed + local.pipe.recvReturn
	911	+	}
	912	+	require.Eventually(t, func() bool { return held() > full }, 5*time.Second, time.Millisecond,
	913	+	"the relay must fill one stream window")
	914	+	require.LessOrEqual(t, held(), brokerwire.StreamWindowBytes, "the relay never exceeds its credit")
906	915		select {
907	916		case <-stream.Done():
908	917		t.Fatalf("a stalled consumer must not settle its stream: %v", stream.Err())
909		-	case <-time.After(200 * time.Millisecond):
	918	+	default:
910	919		}
911	920		requireConnectionDelegates(t, client)
912	921

 package brokeripc
 import (
+	"bytes"
 	"context"
 	"errors"
 	"io"
 	}
 	// Well over one stream window of daemon output while nothing reads. Side
-	// effect Outputs need no view context; random-looking data defeats
+	// effect Outputs need no view context and are never compressed.
-	// compression so every byte crosses the stream.
 	const messages = 24
-	data := make([]byte, 512<<10)
+	data := bytes.Repeat([]byte{'x'}, 512<<10)
-	for i := range data {
-		data[i] = byte(i*131 + i/7)
-	}
 	go func() {
 		for i := range messages {
 			payload := append([]byte{byte(i)}, data...)
 		}
 	}()
-	// The relay is blocked on credit, not failed: the stream stays open and the
+	// The relay fills exactly one window of the client pipe and then waits on
-	// connection keeps serving other work.
+	// credit instead of failing: the stream stays open and the connection keeps
+	// serving other work.
+	local, ok := stream.(*clientStream)
+	require.True(t, ok)
+	full := brokerwire.StreamWindowBytes - brokerwire.StreamChunkCredit(int(brokerwire.MaxStreamChunkBytes))
+	held := func() uint64 {
+		local.pipe.mu.Lock()
+		defer local.pipe.mu.Unlock()
+		return local.pipe.recvUsed + local.pipe.recvReturn
+	}
+	require.Eventually(t, func() bool { return held() > full }, 5*time.Second, time.Millisecond,
+		"the relay must fill one stream window")
+	require.LessOrEqual(t, held(), brokerwire.StreamWindowBytes, "the relay never exceeds its credit")
 	select {
 	case <-stream.Done():
 		t.Fatalf("a stalled consumer must not settle its stream: %v", stream.Err())
-	case <-time.After(200 * time.Millisecond):
+	default:
 	}
 	requireConnectionDelegates(t, client)

M internal/adapters/brokeripc/stream.go +15 -45


@@ -42,33 +42,18 @@ func newServerStream(s *serverSession, id ports.BrokerStreamID, core ports.Broke
42	42		return nil, ErrConfig
43	43		}
44	44		st := &serverStream{session: s, id: id, core: core, done: make(chan struct{})}
45		-	st.pipe = newStreamPipe(st.sendChunk, st.grantCredit)
	45	+	st.pipe = newStreamPipe(int(s.ceilings.StreamChunkLimit), st.sendFrame, st.grantCredit)
46	46		st.carriage = newCarriage(st.pipe)
47	47		return st, nil
48	48		}
49	49
50		-	// sendChunk splits one outbound byte slice under the negotiated stream chunk
51		-	// ceiling and writes each frame in order, waiting for the client's credit
52		-	// before each one: a client that is not reading slows this relay, and through
53		-	// the daemon link the daemon, instead of losing its stream.
54		-	func (st *serverStream) sendChunk(data []byte) error {
55		-	ceiling := int(st.session.ceilings.StreamChunkLimit)
56		-	if ceiling <= 0 {
57		-	return ErrConfig
58		-	}
59		-	for len(data) > 0 {
60		-	n := min(len(data), ceiling)
61		-	if err := st.pipe.reserve(n); err != nil {
62		-	return err
63		-	}
64		-	if err := st.session.send(brokerwire.ServerStreamData{
65		-	Epoch: st.session.epoch, Connection: st.session.scope.Connection, Stream: st.id, Data: data[:n],
66		-	}); err != nil {
67		-	return err
68		-	}
69		-	data = data[n:]
70		-	}
71		-	return nil
	50	+	// sendFrame writes one credited outbound frame to the client. A client that
	51	+	// is not reading slows this relay, and through the daemon link the daemon,
	52	+	// instead of losing its stream.
	53	+	func (st *serverStream) sendFrame(data []byte) error {
	54	+	return st.session.send(brokerwire.ServerStreamData{
	55	+	Epoch: st.session.epoch, Connection: st.session.scope.Connection, Stream: st.id, Data: data,
	56	+	})
72	57		}
73	58
74	59		// grantCredit returns consumed inbound credit to the client.

@@ -403,7 +388,7 @@ func (s serverSession) streamFor(id ports.BrokerStreamID) (serverStream, error
403	388		}
404	389
405	390		// clientStream is the client-side half of one logical stream: the sessionwire
406		-	// client connection over the stream's bounded pipe, plus the terminal outcome a
	391	+	// client connection over the stream's credit-controlled pipe, plus the terminal outcome a
407	392		// ports.BrokerLogicalConnection exposes independently of reads.
408	393		type clientStream struct {
409	394		ports.ClientConnection

@@ -430,7 +415,7 @@ func newClientStream(c client, id ports.BrokerStreamID) (clientStream, error)
430	415		opened: make(chan error, 1),
431	416		done: make(chan struct{}),
432	417		}
433		-	st.pipe = newStreamPipe(st.sendChunk, st.grantCredit)
	418	+	st.pipe = newStreamPipe(int(c.ceilings.StreamChunkLimit), st.sendFrame, st.grantCredit)
434	419		st.carriage = newCarriage(st.pipe)
435	420		st.ClientConnection = sessionwire.NewClientConnection(st.carriage)
436	421		if st.ClientConnection == nil {

@@ -439,26 +424,11 @@ func newClientStream(c client, id ports.BrokerStreamID) (clientStream, error)
439	424		return st, nil
440	425		}
441	426
442		-	// sendChunk splits one outbound client byte slice under the negotiated stream
443		-	// chunk ceiling, waiting for the broker's credit before each frame.
444		-	func (st *clientStream) sendChunk(data []byte) error {
445		-	ceiling := int(st.client.ceilings.StreamChunkLimit)
446		-	if ceiling <= 0 {
447		-	return ErrConfig
448		-	}
449		-	for len(data) > 0 {
450		-	n := min(len(data), ceiling)
451		-	if err := st.pipe.reserve(n); err != nil {
452		-	return err
453		-	}
454		-	if err := st.client.send(brokerwire.ClientStreamData{
455		-	Epoch: st.client.scope.Epoch, Connection: st.client.scope.Connection, Stream: st.id, Data: data[:n],
456		-	}); err != nil {
457		-	return err
458		-	}
459		-	data = data[n:]
460		-	}
461		-	return nil
	427	+	// sendFrame writes one credited outbound frame to the broker.
	428	+	func (st *clientStream) sendFrame(data []byte) error {
	429	+	return st.client.send(brokerwire.ClientStreamData{
	430	+	Epoch: st.client.scope.Epoch, Connection: st.client.scope.Connection, Stream: st.id, Data: data,
	431	+	})
462	432		}
463	433
464	434		// grantCredit returns consumed inbound credit to the broker. It is written

 		return nil, ErrConfig
 	}
 	st := &serverStream{session: s, id: id, core: core, done: make(chan struct{})}
-	st.pipe = newStreamPipe(st.sendChunk, st.grantCredit)
+	st.pipe = newStreamPipe(int(s.ceilings.StreamChunkLimit), st.sendFrame, st.grantCredit)
 	st.carriage = newCarriage(st.pipe)
 	return st, nil
 }
-// sendChunk splits one outbound byte slice under the negotiated stream chunk
+// sendFrame writes one credited outbound frame to the client. A client that
-// ceiling and writes each frame in order, waiting for the client's credit
+// is not reading slows this relay, and through the daemon link the daemon,
-// before each one: a client that is not reading slows this relay, and through
+// instead of losing its stream.
-// the daemon link the daemon, instead of losing its stream.
+func (st *serverStream) sendFrame(data []byte) error {
-func (st *serverStream) sendChunk(data []byte) error {
+	return st.session.send(brokerwire.ServerStreamData{
-	ceiling := int(st.session.ceilings.StreamChunkLimit)
+		Epoch: st.session.epoch, Connection: st.session.scope.Connection, Stream: st.id, Data: data,
-	if ceiling <= 0 {
+	})
-		return ErrConfig
-	}
-	for len(data) > 0 {
-		n := min(len(data), ceiling)
-		if err := st.pipe.reserve(n); err != nil {
-			return err
-		}
-		if err := st.session.send(brokerwire.ServerStreamData{
-			Epoch: st.session.epoch, Connection: st.session.scope.Connection, Stream: st.id, Data: data[:n],
-		}); err != nil {
-			return err
-		}
-		data = data[n:]
-	}
-	return nil
 }
 // grantCredit returns consumed inbound credit to the client.
 }
 // clientStream is the client-side half of one logical stream: the sessionwire
-// client connection over the stream's bounded pipe, plus the terminal outcome a
+// client connection over the stream's credit-controlled pipe, plus the terminal outcome a
 // ports.BrokerLogicalConnection exposes independently of reads.
 type clientStream struct {
 	ports.ClientConnection
 		opened: make(chan error, 1),
 		done:   make(chan struct{}),
 	}
-	st.pipe = newStreamPipe(st.sendChunk, st.grantCredit)
+	st.pipe = newStreamPipe(int(c.ceilings.StreamChunkLimit), st.sendFrame, st.grantCredit)
 	st.carriage = newCarriage(st.pipe)
 	st.ClientConnection = sessionwire.NewClientConnection(st.carriage)
 	if st.ClientConnection == nil {
 	return st, nil
 }
-// sendChunk splits one outbound client byte slice under the negotiated stream
+// sendFrame writes one credited outbound frame to the broker.
-// chunk ceiling, waiting for the broker's credit before each frame.
+func (st *clientStream) sendFrame(data []byte) error {
-func (st *clientStream) sendChunk(data []byte) error {
+	return st.client.send(brokerwire.ClientStreamData{
-	ceiling := int(st.client.ceilings.StreamChunkLimit)
+		Epoch: st.client.scope.Epoch, Connection: st.client.scope.Connection, Stream: st.id, Data: data,
-	if ceiling <= 0 {
+	})
-		return ErrConfig
-	}
-	for len(data) > 0 {
-		n := min(len(data), ceiling)
-		if err := st.pipe.reserve(n); err != nil {
-			return err
-		}
-		if err := st.client.send(brokerwire.ClientStreamData{
-			Epoch: st.client.scope.Epoch, Connection: st.client.scope.Connection, Stream: st.id, Data: data[:n],
-		}); err != nil {
-			return err
-		}
-		data = data[n:]
-	}
-	return nil
 }
 // grantCredit returns consumed inbound credit to the broker. It is written

M internal/adapters/brokerwire/codec_test.go +39 -1


@@ -360,8 +360,9 @@ func TestBrokerClientVariants(t *testing.T) {
360	360		{"update_host_policy", UpdateHostPolicy{Epoch: 7, Connection: connection, Operation: operation, Registration: testRegistration(), Policy: testPolicy()}, 111},
361	361		{"start_preview", testStartPreview(connection), 112},
362	362		{"cancel_preview", CancelPreview{Epoch: 7, Connection: connection, Generation: 3}, 113},
	363	+	{"stream_window_update", StreamWindowUpdate{Epoch: 7, Connection: connection, Stream: 3, Credit: 4096}, 114},
363	364		}
364		-	require.Len(t, messages, 13)
	365	+	require.Len(t, messages, 14)
365	366		for _, tc := range messages {
366	367		t.Run(tc.name, func(t *testing.T) {
367	368		raw := mustEncodeClient(t, tc.message)

@@ -653,6 +654,7 @@ func TestBrokerServerVariants(t *testing.T) {
653	654		{"shutdown", Shutdown{Epoch: 7, Connection: connection, Reason: ShutdownTerminating, Text: "stopping"}, 209},
654	655		{"preview_publication", testPreviewPublication(connection), 210},
655	656		{"preview_publication_error", PreviewPublication{Epoch: 7, Connection: connection, Generation: 5, Error: testErrorDetail(), HasError: true}, 210},
	657	+	{"stream_window_update", StreamWindowUpdate{Epoch: 7, Connection: connection, Stream: 3, Credit: StreamWindowBytes}, 211},
656	658		}
657	659		for _, tc := range messages {
658	660		t.Run(tc.name, func(t *testing.T) {

@@ -924,6 +926,40 @@ func TestBrokerValidationNegatives(t *testing.T) {
924	926		_, err := EncodeClient(CloseStream{Epoch: 7, Connection: connection}, testEnvelopeCeiling, testChunkCeiling)
925	927		require.ErrorIs(t, err, ErrInvalidMessage)
926	928		})
	929	+	t.Run("stream window update bounds", func(t *testing.T) {
	930	+	valid := StreamWindowUpdate{Epoch: 7, Connection: connection, Stream: 3, Credit: 1}
	931	+	cases := map[string]func(*StreamWindowUpdate){
	932	+	"zero credit": func(m *StreamWindowUpdate) { m.Credit = 0 },
	933	+	"credit above window": func(m *StreamWindowUpdate) { m.Credit = StreamWindowBytes + 1 },
	934	+	"zero epoch": func(m *StreamWindowUpdate) { m.Epoch = 0 },
	935	+	"zero stream": func(m *StreamWindowUpdate) { m.Stream = 0 },
	936	+	}
	937	+	for name, mutate := range cases {
	938	+	t.Run(name, func(t *testing.T) {
	939	+	bad := valid
	940	+	mutate(&bad)
	941	+	_, err := EncodeClient(bad, testEnvelopeCeiling, testChunkCeiling)
	942	+	require.ErrorIs(t, err, ErrInvalidMessage)
	943	+	_, err = EncodeServer(bad, testEnvelopeCeiling, testChunkCeiling)
	944	+	require.ErrorIs(t, err, ErrInvalidMessage)
	945	+	})
	946	+	}
	947	+	// The decoder refuses the same out-of-range credit a peer could put
	948	+	// on the wire directly.
	949	+	for name, credit := range map[string]uint64{"zero credit": 0, "credit above window": StreamWindowBytes + 1} {
	950	+	t.Run("decode "+name, func(t *testing.T) {
	951	+	message := &wire.StreamWindowUpdate{Ref: refToWire(7, connection, 3), Credit: credit}
	952	+	client, err := proto.Marshal(&wire.BrokerClientEnvelope{Payload: &wire.BrokerClientEnvelope_StreamWindowUpdate{StreamWindowUpdate: message}})
	953	+	require.NoError(t, err)
	954	+	_, err = DecodeClient(client, testEnvelopeCeiling, testChunkCeiling)
	955	+	require.ErrorIs(t, err, ErrInvalidMessage)
	956	+	server, err := proto.Marshal(&wire.BrokerServerEnvelope{Payload: &wire.BrokerServerEnvelope_StreamWindowUpdate{StreamWindowUpdate: message}})
	957	+	require.NoError(t, err)
	958	+	_, err = DecodeServer(server, testEnvelopeCeiling, testChunkCeiling)
	959	+	require.ErrorIs(t, err, ErrInvalidMessage)
	960	+	})
	961	+	}
	962	+	})
927	963		t.Run("error detail never travels without HasError", func(t *testing.T) {
928	964		detail := testErrorDetail()
929	965		// HasError=false is exactly the absent detail, whatever the outcome

@@ -1254,6 +1290,7 @@ func fuzzClientSeeds(t testing.TB) [][]byte {
1254	1290		OpenStream{Epoch: 7, Connection: testConnectionID(0x11), Stream: 3, Purpose: ports.BrokerStreamAttachment, Admission: ports.BrokerAdmissionExact, Endpoint: "dev@host:22", Registration: testRegistration(), Target: testTarget(), Policy: testPolicy(), StartMode: ports.BrokerDaemonStartIfNeeded},
1255	1291		ClientStreamData{Epoch: 7, Connection: testConnectionID(0x11), Stream: 3, Data: []byte("frame")},
1256	1292		CloseStream{Epoch: 7, Connection: testConnectionID(0x11), Stream: 3},
	1293	+	StreamWindowUpdate{Epoch: 7, Connection: testConnectionID(0x11), Stream: 3, Credit: 4096},
1257	1294		} {
1258	1295		raw, err := EncodeClient(message, testEnvelopeCeiling, testChunkCeiling)
1259	1296		require.NoError(t, err)

@@ -1272,6 +1309,7 @@ func fuzzServerSeeds(t testing.TB) [][]byte {
1272	1309		OperationResult{Epoch: 7, Connection: testConnectionID(0x11), Operation: testOperationID(0x14), Outcome: ports.BrokerOutcomeOK, Registration: testRegistration()},
1273	1310		BrokerErrorMessage{Epoch: 7, Connection: testConnectionID(0x11), Error: testErrorDetail()},
1274	1311		Shutdown{Epoch: 7, Connection: testConnectionID(0x11), Reason: ShutdownTerminating},
	1312	+	StreamWindowUpdate{Epoch: 7, Connection: testConnectionID(0x11), Stream: 3, Credit: 4096},
1275	1313		} {
1276	1314		raw, err := EncodeServer(message, testEnvelopeCeiling, testChunkCeiling)
1277	1315		require.NoError(t, err)


@@ -360,8 +360,9 @@ func TestBrokerClientVariants(t *testing.T) {
360		{"update_host_policy", UpdateHostPolicy{Epoch: 7, Connection: connection, Operation: operation, Registration: testRegistration(), Policy: testPolicy()}, 111},	360		{"update_host_policy", UpdateHostPolicy{Epoch: 7, Connection: connection, Operation: operation, Registration: testRegistration(), Policy: testPolicy()}, 111},
361		{"start_preview", testStartPreview(connection), 112},	361		{"start_preview", testStartPreview(connection), 112},
362		{"cancel_preview", CancelPreview{Epoch: 7, Connection: connection, Generation: 3}, 113},	362		{"cancel_preview", CancelPreview{Epoch: 7, Connection: connection, Generation: 3}, 113},
			363	+	{"stream_window_update", StreamWindowUpdate{Epoch: 7, Connection: connection, Stream: 3, Credit: 4096}, 114},
363		}	364		}
364	-	require.Len(t, messages, 13)	365	+	require.Len(t, messages, 14)
365		for _, tc := range messages {	366		for _, tc := range messages {
366		t.Run(tc.name, func(t *testing.T) {	367		t.Run(tc.name, func(t *testing.T) {
367		raw := mustEncodeClient(t, tc.message)	368		raw := mustEncodeClient(t, tc.message)

@@ -653,6 +654,7 @@ func TestBrokerServerVariants(t *testing.T) {
653		{"shutdown", Shutdown{Epoch: 7, Connection: connection, Reason: ShutdownTerminating, Text: "stopping"}, 209},	654		{"shutdown", Shutdown{Epoch: 7, Connection: connection, Reason: ShutdownTerminating, Text: "stopping"}, 209},
654		{"preview_publication", testPreviewPublication(connection), 210},	655		{"preview_publication", testPreviewPublication(connection), 210},
655		{"preview_publication_error", PreviewPublication{Epoch: 7, Connection: connection, Generation: 5, Error: testErrorDetail(), HasError: true}, 210},	656		{"preview_publication_error", PreviewPublication{Epoch: 7, Connection: connection, Generation: 5, Error: testErrorDetail(), HasError: true}, 210},
			657	+	{"stream_window_update", StreamWindowUpdate{Epoch: 7, Connection: connection, Stream: 3, Credit: StreamWindowBytes}, 211},
656		}	658		}
657		for _, tc := range messages {	659		for _, tc := range messages {
658		t.Run(tc.name, func(t *testing.T) {	660		t.Run(tc.name, func(t *testing.T) {

@@ -924,6 +926,40 @@ func TestBrokerValidationNegatives(t *testing.T) {
924		_, err := EncodeClient(CloseStream{Epoch: 7, Connection: connection}, testEnvelopeCeiling, testChunkCeiling)	926		_, err := EncodeClient(CloseStream{Epoch: 7, Connection: connection}, testEnvelopeCeiling, testChunkCeiling)
925		require.ErrorIs(t, err, ErrInvalidMessage)	927		require.ErrorIs(t, err, ErrInvalidMessage)
926		})	928		})
			929	+	t.Run("stream window update bounds", func(t *testing.T) {
			930	+	valid := StreamWindowUpdate{Epoch: 7, Connection: connection, Stream: 3, Credit: 1}
			931	+	cases := map[string]func(*StreamWindowUpdate){
			932	+	"zero credit": func(m *StreamWindowUpdate) { m.Credit = 0 },
			933	+	"credit above window": func(m *StreamWindowUpdate) { m.Credit = StreamWindowBytes + 1 },
			934	+	"zero epoch": func(m *StreamWindowUpdate) { m.Epoch = 0 },
			935	+	"zero stream": func(m *StreamWindowUpdate) { m.Stream = 0 },
			936	+	}
			937	+	for name, mutate := range cases {
			938	+	t.Run(name, func(t *testing.T) {
			939	+	bad := valid
			940	+	mutate(&bad)
			941	+	_, err := EncodeClient(bad, testEnvelopeCeiling, testChunkCeiling)
			942	+	require.ErrorIs(t, err, ErrInvalidMessage)
			943	+	_, err = EncodeServer(bad, testEnvelopeCeiling, testChunkCeiling)
			944	+	require.ErrorIs(t, err, ErrInvalidMessage)
			945	+	})
			946	+	}
			947	+	// The decoder refuses the same out-of-range credit a peer could put
			948	+	// on the wire directly.
			949	+	for name, credit := range map[string]uint64{"zero credit": 0, "credit above window": StreamWindowBytes + 1} {
			950	+	t.Run("decode "+name, func(t *testing.T) {
			951	+	message := &wire.StreamWindowUpdate{Ref: refToWire(7, connection, 3), Credit: credit}
			952	+	client, err := proto.Marshal(&wire.BrokerClientEnvelope{Payload: &wire.BrokerClientEnvelope_StreamWindowUpdate{StreamWindowUpdate: message}})
			953	+	require.NoError(t, err)
			954	+	_, err = DecodeClient(client, testEnvelopeCeiling, testChunkCeiling)
			955	+	require.ErrorIs(t, err, ErrInvalidMessage)
			956	+	server, err := proto.Marshal(&wire.BrokerServerEnvelope{Payload: &wire.BrokerServerEnvelope_StreamWindowUpdate{StreamWindowUpdate: message}})
			957	+	require.NoError(t, err)
			958	+	_, err = DecodeServer(server, testEnvelopeCeiling, testChunkCeiling)
			959	+	require.ErrorIs(t, err, ErrInvalidMessage)
			960	+	})
			961	+	}
			962	+	})
927		t.Run("error detail never travels without HasError", func(t *testing.T) {	963		t.Run("error detail never travels without HasError", func(t *testing.T) {
928		detail := testErrorDetail()	964		detail := testErrorDetail()
929		// HasError=false is exactly the absent detail, whatever the outcome	965		// HasError=false is exactly the absent detail, whatever the outcome

@@ -1254,6 +1290,7 @@ func fuzzClientSeeds(t testing.TB) [][]byte {
1254		OpenStream{Epoch: 7, Connection: testConnectionID(0x11), Stream: 3, Purpose: ports.BrokerStreamAttachment, Admission: ports.BrokerAdmissionExact, Endpoint: "dev@host:22", Registration: testRegistration(), Target: testTarget(), Policy: testPolicy(), StartMode: ports.BrokerDaemonStartIfNeeded},	1290		OpenStream{Epoch: 7, Connection: testConnectionID(0x11), Stream: 3, Purpose: ports.BrokerStreamAttachment, Admission: ports.BrokerAdmissionExact, Endpoint: "dev@host:22", Registration: testRegistration(), Target: testTarget(), Policy: testPolicy(), StartMode: ports.BrokerDaemonStartIfNeeded},
1255		ClientStreamData{Epoch: 7, Connection: testConnectionID(0x11), Stream: 3, Data: []byte("frame")},	1291		ClientStreamData{Epoch: 7, Connection: testConnectionID(0x11), Stream: 3, Data: []byte("frame")},
1256		CloseStream{Epoch: 7, Connection: testConnectionID(0x11), Stream: 3},	1292		CloseStream{Epoch: 7, Connection: testConnectionID(0x11), Stream: 3},
			1293	+	StreamWindowUpdate{Epoch: 7, Connection: testConnectionID(0x11), Stream: 3, Credit: 4096},
1257		} {	1294		} {
1258		raw, err := EncodeClient(message, testEnvelopeCeiling, testChunkCeiling)	1295		raw, err := EncodeClient(message, testEnvelopeCeiling, testChunkCeiling)
1259		require.NoError(t, err)	1296		require.NoError(t, err)

@@ -1272,6 +1309,7 @@ func fuzzServerSeeds(t testing.TB) [][]byte {
1272		OperationResult{Epoch: 7, Connection: testConnectionID(0x11), Operation: testOperationID(0x14), Outcome: ports.BrokerOutcomeOK, Registration: testRegistration()},	1309		OperationResult{Epoch: 7, Connection: testConnectionID(0x11), Operation: testOperationID(0x14), Outcome: ports.BrokerOutcomeOK, Registration: testRegistration()},
1273		BrokerErrorMessage{Epoch: 7, Connection: testConnectionID(0x11), Error: testErrorDetail()},	1310		BrokerErrorMessage{Epoch: 7, Connection: testConnectionID(0x11), Error: testErrorDetail()},
1274		Shutdown{Epoch: 7, Connection: testConnectionID(0x11), Reason: ShutdownTerminating},	1311		Shutdown{Epoch: 7, Connection: testConnectionID(0x11), Reason: ShutdownTerminating},
			1312	+	StreamWindowUpdate{Epoch: 7, Connection: testConnectionID(0x11), Stream: 3, Credit: 4096},
1275		} {	1313		} {
1276		raw, err := EncodeServer(message, testEnvelopeCeiling, testChunkCeiling)	1314		raw, err := EncodeServer(message, testEnvelopeCeiling, testChunkCeiling)
1277		require.NoError(t, err)	1315		require.NoError(t, err)

M internal/adapters/brokerwire/state.go +0 -7


@@ -486,13 +486,6 @@ func (c *Connection) StreamData(stream ports.BrokerStreamID) (StreamDisposition,
486	486		return c.streams.Data(c.scope, stream)
487	487		}
488	488
489		-	// StreamWindowUpdate classifies one credit grant for a stream exactly like
490		-	// stream data: legal while open, discarded once retired. The ready check and
491		-	// the tracker lookup share one connection lock.
492		-	func (c *Connection) StreamWindowUpdate(stream ports.BrokerStreamID) (StreamDisposition, error) {
493		-	return c.StreamData(stream)
494		-	}
495		-
496	489		// StreamProgress reports one phase of an in-flight stream operation. The
497	490		// ready check and the tracker mutation share one connection lock.
498	491		func (c *Connection) StreamProgress(stream ports.BrokerStreamID) (StreamDisposition, error) {


@@ -486,13 +486,6 @@ func (c *Connection) StreamData(stream ports.BrokerStreamID) (StreamDisposition,
486		return c.streams.Data(c.scope, stream)	486	return c.streams.Data(c.scope, stream)
487		}	487	}
488			488
489	-	// StreamWindowUpdate classifies one credit grant for a stream exactly like
490	-	// stream data: legal while open, discarded once retired. The ready check and
491	-	// the tracker lookup share one connection lock.
492	-	func (c *Connection) StreamWindowUpdate(stream ports.BrokerStreamID) (StreamDisposition, error) {
493	-	return c.StreamData(stream)
494	-	}
495	-
496		// StreamProgress reports one phase of an in-flight stream operation. The	489	// StreamProgress reports one phase of an in-flight stream operation. The
497		// ready check and the tracker mutation share one connection lock.	490	// ready check and the tracker mutation share one connection lock.
498		func (c *Connection) StreamProgress(stream ports.BrokerStreamID) (StreamDisposition, error) {	491	func (c *Connection) StreamProgress(stream ports.BrokerStreamID) (StreamDisposition, error) {

M internal/usecase/daemon/attachment_output.go +8 -20


@@ -33,8 +33,10 @@ type attachmentOutput struct {
33	33		maxOutstandingAtomic atomic.Uint64
34	34		// unackedBytes charges Output data sent and not yet acknowledged, keyed by
35	35		// the state an ACK must reach to release it. Side-effect output carries no
36		-	// state of its own, so it is charged to the next state. Guarded by the view
37		-	// lock; unackedBytesAtomic mirrors the total for lock-free capacity reads.
	36	+	// state of its own, so it is charged to the next state. Charges stay in
	37	+	// non-decreasing state order and merge per state, so the slice holds at most
	38	+	// maxOutstanding+1 entries. Guarded by the view lock; unackedBytesAtomic
	39	+	// mirrors the total for lock-free capacity reads.
38	40		unackedBytes []unackedOutputBytes
39	41		unackedBytesAtomic atomic.Uint64
40	42		forceSnapshot bool

@@ -220,10 +222,10 @@ func (o attachmentOutput) prepareFrame(d Daemon, state *capturedRenderState, f
220	222		ansi.viewRevision = state.view.revision
221	223		cursor := o.prepareCursorTail(desired, len(ansi.data) > 0)
222	224		graphicsReset := reset \|\| o.forceSnapshot \|\| !o.initialized
223		-	// Images are optional decoration: they share the frame's data limit and the
224		-	// attachment's unacknowledged byte window, and an upload that fits neither
225		-	// waits for a later frame instead of overrunning buffers toward the client.
226		-	graphicsBudget := max(min(protocol.MaxOutputDataLen, o.unackedBytesBudget())-len(ansi.data)-len(cursor.data), 0)
	225	+	// Images share the frame's data limit. The unacknowledged byte window gates
	226	+	// the next frame rather than this one: an upload must fit one frame, so
	227	+	// clamping it here would drop large images and their deletes outright.
	228	+	graphicsBudget := max(protocol.MaxOutputDataLen-len(ansi.data)-len(cursor.data), 0)
227	229		preparedGraphics, err := graphicsOutputDataWithDaemonLimit(d, state, o.attachment, graphicsReset, graphicsBudget)
228	230		if err != nil {
229	231		return nil, err

@@ -677,20 +679,6 @@ func (s *attachmentOutput) atCapacity() bool {
677	679		(outstanding > 0 && s.unackedBytesAtomic.Load() >= protocol.MaxOutputWindowBytes)
678	680		}
679	681
680		-	// unackedBytesBudget is the Output data that may still be sent before the byte
681		-	// window is full. It is advisory: a frame may exceed it when no state frame is
682		-	// in flight.
683		-	func (s *attachmentOutput) unackedBytesBudget() int {
684		-	if s == nil {
685		-	return protocol.MaxOutputDataLen
686		-	}
687		-	used := s.unackedBytesAtomic.Load()
688		-	if used >= protocol.MaxOutputWindowBytes {
689		-	return 0
690		-	}
691		-	return int(protocol.MaxOutputWindowBytes - used)
692		-	}
693		-
694	682		// committedState reports the latest committed output state number. It takes
695	683		// only the view lock, so barrier snapshots can read it without joining the
696	684		// sendMu-ordered paint path.

 	maxOutstandingAtomic atomic.Uint64
 	// unackedBytes charges Output data sent and not yet acknowledged, keyed by
 	// the state an ACK must reach to release it. Side-effect output carries no
-	// state of its own, so it is charged to the next state. Guarded by the view
+	// state of its own, so it is charged to the next state. Charges stay in
-	// lock; unackedBytesAtomic mirrors the total for lock-free capacity reads.
+	// non-decreasing state order and merge per state, so the slice holds at most
+	// maxOutstanding+1 entries. Guarded by the view lock; unackedBytesAtomic
+	// mirrors the total for lock-free capacity reads.
 	unackedBytes              []unackedOutputBytes
 	unackedBytesAtomic        atomic.Uint64
 	forceSnapshot             bool
 	ansi.viewRevision = state.view.revision
 	cursor := o.prepareCursorTail(desired, len(ansi.data) > 0)
 	graphicsReset := reset || o.forceSnapshot || !o.initialized
-	// Images are optional decoration: they share the frame's data limit and the
+	// Images share the frame's data limit. The unacknowledged byte window gates
-	// attachment's unacknowledged byte window, and an upload that fits neither
+	// the next frame rather than this one: an upload must fit one frame, so
-	// waits for a later frame instead of overrunning buffers toward the client.
+	// clamping it here would drop large images and their deletes outright.
-	graphicsBudget := max(min(protocol.MaxOutputDataLen, o.unackedBytesBudget())-len(ansi.data)-len(cursor.data), 0)
+	graphicsBudget := max(protocol.MaxOutputDataLen-len(ansi.data)-len(cursor.data), 0)
 	preparedGraphics, err := graphicsOutputDataWithDaemonLimit(d, state, o.attachment, graphicsReset, graphicsBudget)
 	if err != nil {
 		return nil, err
 		(outstanding > 0 && s.unackedBytesAtomic.Load() >= protocol.MaxOutputWindowBytes)
 }
-// unackedBytesBudget is the Output data that may still be sent before the byte
-// window is full. It is advisory: a frame may exceed it when no state frame is
-// in flight.
-func (s *attachmentOutput) unackedBytesBudget() int {
-	if s == nil {
-		return protocol.MaxOutputDataLen
-	}
-	used := s.unackedBytesAtomic.Load()
-	if used >= protocol.MaxOutputWindowBytes {
-		return 0
-	}
-	return int(protocol.MaxOutputWindowBytes - used)
-}
 // committedState reports the latest committed output state number. It takes
 // only the view lock, so barrier snapshots can read it without joining the
 // sendMu-ordered paint path.

M internal/usecase/daemon/attachment_output_test.go +35 -0


@@ -577,6 +577,41 @@ func TestAttachmentOutputByteWindow(t *testing.T) {
577	577		}
578	578		}
579	579
	580	+	// TestAttachmentOutputFullByteWindowKeepsImages proves the byte window gates
	581	+	// when a frame goes out, never what it carries: an upload must fit one frame,
	582	+	// so trimming images to the remaining byte budget would drop them for good.
	583	+	func TestAttachmentOutputFullByteWindowKeepsImages(t *testing.T) {
	584	+	scene := graphics.NewScene(graphics.Limits{})
	585	+	asset, err := scene.AddAsset(graphics.AssetBlob{Encoded: []byte("asset"), Width: 1, Height: 1})
	586	+	require.NoError(t, err)
	587	+	_, err = scene.PlaceAsset(asset, graphics.PixelRect{Width: 1, Height: 1})
	588	+	require.NoError(t, err)
	589	+	state := &capturedRenderState{
	590	+	route: protocol.CommittedRouteIdentity{Target: protocol.ExactSessionTarget{LifecycleID: domain.SessionLifecycleID{1}, SessionName: "work"}},
	591	+	view: attachmentView{tabID: "tab-1"},
	592	+	panes: []capturedPaneRenderState{{
	593	+	stableID: "pane-1", focused: true,
	594	+	graphics: scene.Snapshot(),
	595	+	graphicsGeometry: domain.Geometry{Size: domain.Size{Cols: 1, Rows: 1}, PixelWidth: 1, PixelHeight: 1},
	596	+	placement: layout.Placement{Content: domain.Rect{Width: 1, Height: 1}},
	597	+	}}}
	598	+	output := attachmentOutputWithGraphics(newGraphicsOutputState())
	599	+	ac := &attachedClient{output: output, terminalCapabilities: terminalcap.Capabilities{KittyGraphics: true}}
	600	+	output.attachment = ac
	601	+	// Side effects filled the byte window while no state frame is in flight.
	602	+	output.lockView()
	603	+	output.chargeBytesLocked(output.next+1, protocol.MaxOutputWindowBytes)
	604	+	output.unlockView()
	605	+	require.False(t, output.atCapacity(), "the only frame that can release side-effect bytes is allowed")
	606	+
	607	+	prepared, err := output.prepareFrame(nil, state, renderer.NewFrame(1, 1), []renderer.Damage{renderer.FullRedraw()}, true, cursorOut{})
	608	+	require.NoError(t, err)
	609	+	require.NotNil(t, prepared.graphics, "a full byte window must not drop the frame's images")
	610	+	require.NotEmpty(t, prepared.graphics.data)
	611	+	require.Equal(t, protocol.MaxOutputDataLen, prepared.graphics.maxBytes+len(prepared.data)-len(prepared.graphics.data),
	612	+	"images keep the whole frame data limit")
	613	+	}
	614	+
580	615		func TestAttachmentOutputSentFrameChargesItsBytes(t *testing.T) {
581	616		stream := newOutputStateStream()
582	617		frame := renderer.NewFrame(3, 1)


@@ -577,6 +577,41 @@ func TestAttachmentOutputByteWindow(t *testing.T) {
577	}	577		}
578	}	578		}
579		579
		580	+	// TestAttachmentOutputFullByteWindowKeepsImages proves the byte window gates
		581	+	// when a frame goes out, never what it carries: an upload must fit one frame,
		582	+	// so trimming images to the remaining byte budget would drop them for good.
		583	+	func TestAttachmentOutputFullByteWindowKeepsImages(t *testing.T) {
		584	+	scene := graphics.NewScene(graphics.Limits{})
		585	+	asset, err := scene.AddAsset(graphics.AssetBlob{Encoded: []byte("asset"), Width: 1, Height: 1})
		586	+	require.NoError(t, err)
		587	+	_, err = scene.PlaceAsset(asset, graphics.PixelRect{Width: 1, Height: 1})
		588	+	require.NoError(t, err)
		589	+	state := &capturedRenderState{
		590	+	route: protocol.CommittedRouteIdentity{Target: protocol.ExactSessionTarget{LifecycleID: domain.SessionLifecycleID{1}, SessionName: "work"}},
		591	+	view: attachmentView{tabID: "tab-1"},
		592	+	panes: []capturedPaneRenderState{{
		593	+	stableID: "pane-1", focused: true,
		594	+	graphics: scene.Snapshot(),
		595	+	graphicsGeometry: domain.Geometry{Size: domain.Size{Cols: 1, Rows: 1}, PixelWidth: 1, PixelHeight: 1},
		596	+	placement: layout.Placement{Content: domain.Rect{Width: 1, Height: 1}},
		597	+	}}}
		598	+	output := attachmentOutputWithGraphics(newGraphicsOutputState())
		599	+	ac := &attachedClient{output: output, terminalCapabilities: terminalcap.Capabilities{KittyGraphics: true}}
		600	+	output.attachment = ac
		601	+	// Side effects filled the byte window while no state frame is in flight.
		602	+	output.lockView()
		603	+	output.chargeBytesLocked(output.next+1, protocol.MaxOutputWindowBytes)
		604	+	output.unlockView()
		605	+	require.False(t, output.atCapacity(), "the only frame that can release side-effect bytes is allowed")
		606	+
		607	+	prepared, err := output.prepareFrame(nil, state, renderer.NewFrame(1, 1), []renderer.Damage{renderer.FullRedraw()}, true, cursorOut{})
		608	+	require.NoError(t, err)
		609	+	require.NotNil(t, prepared.graphics, "a full byte window must not drop the frame's images")
		610	+	require.NotEmpty(t, prepared.graphics.data)
		611	+	require.Equal(t, protocol.MaxOutputDataLen, prepared.graphics.maxBytes+len(prepared.data)-len(prepared.graphics.data),
		612	+	"images keep the whole frame data limit")
		613	+	}
		614	+
580	func TestAttachmentOutputSentFrameChargesItsBytes(t *testing.T) {	615		func TestAttachmentOutputSentFrameChargesItsBytes(t *testing.T) {
581	stream := newOutputStateStream()	616		stream := newOutputStateStream()
582	frame := renderer.NewFrame(3, 1)	617		frame := renderer.NewFrame(3, 1)