@@ -95,6 +95,79 @@ export function buildFailedTailPayload(err: unknown): string {
9595 } ) ;
9696}
9797
98+ export type SseTerminalOutputBoundary = {
99+ feed ( chunk : Uint8Array ) : Uint8Array ;
100+ finish ( ) : Uint8Array ;
101+ terminalSeen ( ) : boolean ;
102+ doneSeen ( ) : boolean ;
103+ dispose ( ) : void ;
104+ } ;
105+
106+ /**
107+ * Frame-aware client output boundary shared by both native Responses relays.
108+ * It buffers only the current incomplete SSE block, forwards complete blocks
109+ * through the first Responses terminal, and drops every later block/byte.
110+ */
111+ export function createSseTerminalOutputBoundary ( ) : SseTerminalOutputBoundary {
112+ let decoder : TextDecoder | null = new TextDecoder ( ) ;
113+ const encoder = new TextEncoder ( ) ;
114+ let buffer = "" ;
115+ let terminal = false ;
116+ let done = false ;
117+ let disposed = false ;
118+
119+ const process = ( flush : boolean ) : Uint8Array => {
120+ if ( disposed || terminal ) return new Uint8Array ( 0 ) ;
121+ let output = "" ;
122+ let responsesTerminal = false ;
123+ for ( ; ; ) {
124+ const next = nextSseBlock ( buffer ) ;
125+ if ( ! next ) break ;
126+ buffer = next . rest ;
127+ const payload = sseDataPayload ( next . block ) ;
128+ if ( ! responsesTerminal ) output += next . block + next . delimiter ;
129+ if ( payload === "[DONE]" ) {
130+ done = true ;
131+ if ( responsesTerminal ) output += next . block + next . delimiter ;
132+ continue ;
133+ }
134+ if ( ! responsesTerminal && payload && terminalStatusFromSsePayload ( payload ) ) {
135+ responsesTerminal = true ;
136+ }
137+ }
138+ if ( responsesTerminal ) {
139+ terminal = true ;
140+ buffer = "" ;
141+ }
142+ if ( flush && ! terminal && buffer . length > 0 ) {
143+ output += buffer ;
144+ buffer = "" ;
145+ }
146+ return encoder . encode ( output ) ;
147+ } ;
148+
149+ return {
150+ feed ( chunk ) {
151+ if ( disposed || terminal ) return new Uint8Array ( 0 ) ;
152+ buffer += decoder ! . decode ( chunk , { stream : true } ) ;
153+ return process ( false ) ;
154+ } ,
155+ finish ( ) {
156+ if ( disposed || terminal ) return new Uint8Array ( 0 ) ;
157+ buffer += decoder ! . decode ( ) ;
158+ return process ( true ) ;
159+ } ,
160+ terminalSeen : ( ) => terminal ,
161+ doneSeen : ( ) => done ,
162+ dispose ( ) {
163+ if ( disposed ) return ;
164+ disposed = true ;
165+ decoder = null ;
166+ buffer = "" ;
167+ } ,
168+ } ;
169+ }
170+
98171/**
99172 * Relay a passthrough SSE body like relayWithAbort, but convert a MID-STREAM failure (upstream
100173 * reset after headers) into a clean terminal: any partial block is closed off, then a synthetic
@@ -110,18 +183,57 @@ export function relaySseWithFailedTail(
110183) : ReadableStream < Uint8Array > {
111184 const reader = body . getReader ( ) ;
112185 const encoder = new TextEncoder ( ) ;
186+ const terminalBoundary = createSseTerminalOutputBoundary ( ) ;
187+ let closed = false ;
188+ const relayChunk = (
189+ controller : ReadableStreamDefaultController < Uint8Array > ,
190+ value : Uint8Array ,
191+ ) : "terminal" | "output" | "buffered" => {
192+ const outbound = terminalBoundary . feed ( value ) ;
193+ if ( outbound . byteLength > 0 ) controller . enqueue ( outbound ) ;
194+ if ( ! terminalBoundary . terminalSeen ( ) ) return outbound . byteLength > 0 ? "output" : "buffered" ;
195+
196+ // A Responses terminal frame is the protocol boundary. Some compatible
197+ // gateways leave the HTTP connection open after response.completed, which
198+ // otherwise leaves Codex waiting forever even though the model turn is done.
199+ // Preserve through the terminal block only, add the conventional sentinel
200+ // when there was no real [DONE] data event, then stop reading upstream.
201+ if ( ! terminalBoundary . doneSeen ( ) ) {
202+ controller . enqueue ( encoder . encode ( "data: [DONE]\n\n" ) ) ;
203+ }
204+ closed = true ;
205+ controller . close ( ) ;
206+ const reason = "Responses terminal event received" ;
207+ // Notify the tee inspection branch as well. It has already received the
208+ // same terminal-bearing upstream chunk, so its bounded drain records the
209+ // real terminal and then releases the turn/upstream keep-alive connection.
210+ onClientGone ?.( reason ) ;
211+ reader . cancel ( reason ) . catch ( ( ) => { } ) ;
212+ terminalBoundary . dispose ( ) ;
213+ return "terminal" ;
214+ } ;
113215 return new ReadableStream < Uint8Array > ( {
114216 async pull ( controller ) {
115217 try {
116- const { done, value } = await reader . read ( ) ;
117- if ( done ) {
118- controller . close ( ) ;
119- return ;
218+ for ( ; ; ) {
219+ const { done, value } = await reader . read ( ) ;
220+ if ( done ) {
221+ const tail = terminalBoundary . finish ( ) ;
222+ if ( tail . byteLength > 0 ) controller . enqueue ( tail ) ;
223+ terminalBoundary . dispose ( ) ;
224+ controller . close ( ) ;
225+ return ;
226+ }
227+ const result = relayChunk ( controller , value ) ;
228+ if ( result !== "buffered" ) return ;
120229 }
121- controller . enqueue ( value ) ;
122230 } catch ( err ) {
231+ const partial = terminalBoundary . finish ( ) ;
232+ terminalBoundary . dispose ( ) ;
233+ if ( closed ) return ;
123234 const payload = buildFailedTailPayload ( err ) ;
124235 try {
236+ if ( partial . byteLength > 0 ) controller . enqueue ( partial ) ;
125237 // Leading blank line terminates a partial SSE block so the failed frame parses cleanly.
126238 controller . enqueue ( encoder . encode ( `\n\nevent: response.failed\ndata: ${ payload } ) ) ;
127239 controller . close ( ) ;
@@ -130,18 +242,20 @@ export function relaySseWithFailedTail(
130242 }
131243 } ,
132244 cancel ( reason ) {
245+ terminalBoundary . dispose ( ) ;
133246 if ( onClientGone ) onClientGone ( reason ) ;
134247 else upstream . abort ( reason ) ;
135248 reader . cancel ( reason ) . catch ( ( ) => { } ) ;
136249 } ,
137250 } ) ;
138251}
139252
140- export function nextSseBlock ( buffer : string ) : { block : string ; rest : string } | null {
253+ export function nextSseBlock ( buffer : string ) : { block : string ; delimiter : string ; rest : string } | null {
141254 const match = buffer . match ( / \r ? \n \r ? \n / ) ;
142255 if ( ! match || match . index === undefined ) return null ;
143256 return {
144257 block : buffer . slice ( 0 , match . index ) ,
258+ delimiter : match [ 0 ] ,
145259 rest : buffer . slice ( match . index + match [ 0 ] . length ) ,
146260 } ;
147261}
0 commit comments