@@ -26,15 +26,19 @@ def _clean_reasoning_stream(full_so_far: str) -> str:
2626 if not full_so_far :
2727 return ""
2828 text = re .sub (r"\[END_OF_RESPONSE(?:_[A-Za-z0-9-]+)?\]" , "" , full_so_far , flags = re .IGNORECASE )
29- text = re .sub (r"\bKIM_[a-f0-9]{8}\b" , "" , text ).strip ()
29+ text = re .sub (r"\bKIM_[a-f0-9]{8}\b" , "" , text )
30+ text = re .sub (r"#\s*Codex Agent[\s\S]*?(?=(?:\b(?:Hi|Hello|I'll|I will|Let's|Looking|Checking|Running|Updating|Here|Done)\b|```|\Z))" , "" , text , flags = re .IGNORECASE )
31+ text = re .sub (r"This document defines exactly how the Codex agent[\s\S]*?(?=(?:\b(?:Hi|Hello|I'll|I will|Let's|Looking|Checking|Running|Updating|Here|Done)\b|```|\Z))" , "" , text , flags = re .IGNORECASE )
32+ text = re .sub (r"## Single Command Turn-by-Turn Protocol[\s\S]*?(?=(?:\b(?:Hi|Hello|I'll|I will|Let's|Looking|Checking|Running|Updating|Here|Done)\b|```|\Z))" , "" , text , flags = re .IGNORECASE )
33+ text = re .sub (r"in your response to cite this file[^\n]*\n?" , "" , text , flags = re .IGNORECASE )
34+ text = re .sub (r"Make sure to include\s+(?:filecite[^]+|turn\d+file\d+|\s*)*" , "" , text , flags = re .IGNORECASE )
35+ text = re .sub (r"filecite[^]+" , "" , text )
36+ text = re .sub (r"\bturn\d+file\d+\b" , "" , text , flags = re .IGNORECASE )
37+ text = re .sub (r"^\s*Make sure to include\s*$" , "" , text , flags = re .IGNORECASE | re .MULTILINE ).strip ()
3038 if text .startswith ("{" ):
3139 m = re .search (r'"text"\s*:\s*"((?:[^"\\]|\\.)*)' , text )
3240 if m :
3341 text = m .group (1 ).replace ('\\ "' , '"' ).replace ('\\ n' , '\n ' )
34- else :
35- return ""
36- if "```" in text :
37- text = text .split ("```" , 1 )[0 ]
3842 return text .strip ()
3943
4044
@@ -63,6 +67,7 @@ async def stream_responses_http(
6367
6468 resp_id = f"resp_{ uuid .uuid4 ().hex [:16 ]}
6569 reasoning_id = f"item_{ uuid .uuid4 ().hex [:12 ]}
70+ msg_item_id = f"item_{ uuid .uuid4 ().hex [:12 ]}
6671
6772 in_progress = {
6873 "id" : resp_id ,
@@ -77,27 +82,52 @@ async def stream_responses_http(
7782 f"data: { json .dumps ({'type' : 'response.created' , 'response' : in_progress })} \n \n " .encode ("utf-8" )
7883 )
7984
80- # 2. Send response.output_item.added for reasoning (index 0)
85+ # 2. Send response.output_item.added for reasoning (index 0) and message (index 1)
8186 reasoning_item = {
8287 "id" : reasoning_id ,
8388 "type" : "reasoning" ,
8489 "reasoning_text" : "" ,
85- "summary" : [],
90+ "summary" : [{ "type" : "summary_text" , "text" : "" } ],
8691 "status" : "in_progress" ,
8792 }
8893 await stream_res .write (
8994 f"data: { json .dumps ({'type' : 'response.output_item.added' , 'output_index' : 0 , 'item' : reasoning_item })} \n \n " .encode ("utf-8" )
9095 )
9196
97+ msg_item_added = {
98+ "id" : msg_item_id ,
99+ "type" : "message" ,
100+ "role" : "assistant" ,
101+ "content" : [],
102+ "status" : "in_progress" ,
103+ }
104+ await stream_res .write (
105+ f"data: { json .dumps ({'type' : 'response.output_item.added' , 'output_index' : 1 , 'item' : msg_item_added })} \n \n " .encode ("utf-8" )
106+ )
107+ await stream_res .write (
108+ f"data: { json .dumps ({'type' : 'response.content_part.added' , 'item_id' : msg_item_id , 'output_index' : 1 , 'content_index' : 0 , 'part' : {'type' : 'output_text' , 'text' : '' }})} \n \n " .encode ("utf-8" )
109+ )
110+
92111 loop = asyncio .get_running_loop ()
93112 queue : asyncio .Queue [Optional [str ]] = asyncio .Queue ()
94113 accumulated_reasoning : list [str ] = []
95114 last_emitted_len = 0
115+ is_streaming_done = False
96116
97117 def _on_delta (delta : str ) -> None :
98118 if delta :
99119 loop .call_soon_threadsafe (queue .put_nowait , delta )
100120
121+ async def _heartbeat_worker ():
122+ while not is_streaming_done :
123+ await asyncio .sleep (1.5 )
124+ if not is_streaming_done :
125+ try :
126+ await stream_res .write (b": ping\n \n " )
127+ await stream_res .drain ()
128+ except Exception :
129+ break
130+
101131 async def _stream_worker ():
102132 nonlocal last_emitted_len
103133 while True :
@@ -111,20 +141,38 @@ async def _stream_worker():
111141 if len (clean_full ) > last_emitted_len :
112142 new_delta = clean_full [last_emitted_len :]
113143 last_emitted_len = len (clean_full )
114- ev = {
144+ ev1 = {
145+ "type" : "response.reasoning_summary.text.delta" ,
146+ "item_id" : reasoning_id ,
147+ "output_index" : 0 ,
148+ "content_index" : 0 ,
149+ "delta" : new_delta ,
150+ }
151+ ev2 = {
115152 "type" : "response.reasoning.text.delta" ,
116153 "item_id" : reasoning_id ,
117154 "output_index" : 0 ,
118155 "content_index" : 0 ,
119156 "delta" : new_delta ,
120157 }
158+ ev3 = {
159+ "type" : "response.output_text.delta" ,
160+ "item_id" : msg_item_id ,
161+ "output_index" : 1 ,
162+ "content_index" : 0 ,
163+ "delta" : new_delta ,
164+ }
121165 try :
122- await stream_res .write (f"data: { json .dumps (ev )} \n \n " .encode ("utf-8" ))
166+ await stream_res .write (f"data: { json .dumps (ev1 )} \n \n " .encode ("utf-8" ))
167+ await stream_res .write (f"data: { json .dumps (ev2 )} \n \n " .encode ("utf-8" ))
168+ await stream_res .write (f"data: { json .dumps (ev3 )} \n \n " .encode ("utf-8" ))
169+ await stream_res .drain ()
123170 except Exception :
124171 pass
125172 queue .task_done ()
126173
127174 worker_task = asyncio .create_task (_stream_worker ())
175+ heartbeat_task = asyncio .create_task (_heartbeat_worker ())
128176
129177 # 3. Call provider complete with on_delta callback
130178 extra_kwargs = {"handoff" : handoff } if handoff else {}
@@ -148,15 +196,9 @@ async def _stream_worker():
148196 on_delta = _on_delta ,
149197 ** extra_kwargs ,
150198 )
151- except Exception as e :
152- logger .error (f"[relay #{ relay_num } { e } )
153- await queue .put (None )
154- await worker_task
155- error_payload = {"error" : {"message" : f"LLM call failed: { e } }}
156- await stream_res .write (f"data: { json .dumps (error_payload )} \n \n " .encode ("utf-8" ))
157- await stream_res .write (b"data: [DONE]\n \n " )
158- await stream_res .write_eof ()
159- return stream_res
199+ finally :
200+ is_streaming_done = True
201+ heartbeat_task .cancel ()
160202
161203 # Signal worker to finish draining deltas
162204 await queue .put (None )
@@ -254,36 +296,52 @@ async def _stream_worker():
254296 }
255297 await stream_res .write (f"data: { json .dumps (reasoning_item_completed )} \n \n " .encode ("utf-8" ))
256298
257- # 6. Stream remaining output items (message, function_call, custom_tool_call)
258- out_idx = 1
299+ final_text = curr_reasoning or "" .join (accumulated_reasoning )
300+ for item in responses_reply .get ("output" ) or []:
301+ if isinstance (item , dict ) and item .get ("type" ) == "message" :
302+ for block in (item .get ("content" ) or []):
303+ if isinstance (block , dict ) and block .get ("type" ) == "output_text" :
304+ t = str (block .get ("text" ) or "" )
305+ if t :
306+ final_text = t
307+ break
308+
309+ part_dict = {"type" : "output_text" , "text" : final_text }
310+ await stream_res .write (
311+ f"data: { json .dumps ({'type' : 'response.content_part.done' , 'item_id' : msg_item_id , 'output_index' : 1 , 'content_index' : 0 , 'part' : part_dict })} \n \n " .encode ("utf-8" )
312+ )
313+ await stream_res .write (
314+ f"data: { json .dumps ({'type' : 'response.output_item.done' , 'output_index' : 1 , 'item' : {'id' : msg_item_id , 'type' : 'message' , 'role' : 'assistant' , 'status' : 'completed' , 'content' : [part_dict ]}})} \n \n " .encode ("utf-8" )
315+ )
316+
317+ # Ensure responses_reply output contains message item
318+ has_msg = False
319+ for item in (responses_reply .get ("output" ) or []):
320+ if isinstance (item , dict ) and item .get ("type" ) == "message" :
321+ has_msg = True
322+ if not item .get ("content" ):
323+ item ["content" ] = [part_dict ]
324+ if not has_msg :
325+ responses_reply .setdefault ("output" , []).append ({
326+ "id" : msg_item_id ,
327+ "type" : "message" ,
328+ "role" : "assistant" ,
329+ "status" : "completed" ,
330+ "content" : [part_dict ],
331+ })
332+
333+ # 6. Stream remaining tool call output items
334+ out_idx = 2
259335 for item in responses_reply .get ("output" ) or []:
260336 if not isinstance (item , dict ):
261337 continue
262338 itype = item .get ("type" , "" )
263- if itype == "reasoning" :
339+ if itype in ( "reasoning" , "message" ) :
264340 continue
265341 item_id = str (item .get ("id" ) or f"item_{ uuid .uuid4 ().hex [:12 ]} )
266342 item_dict = {** item , "id" : item_id }
267343
268- if itype == "message" :
269- text = "" .join (
270- str (block .get ("text" ) or "" )
271- for block in (item .get ("content" ) or [])
272- if isinstance (block , dict ) and block .get ("type" ) == "output_text"
273- )
274- await stream_res .write (
275- f"data: { json .dumps ({'type' : 'response.output_item.added' , 'output_index' : out_idx , 'item' : {** item_dict , 'content' : [], 'status' : 'in_progress' }})} \n \n " .encode ("utf-8" )
276- )
277- await stream_res .write (
278- f"data: { json .dumps ({'type' : 'response.output_text.delta' , 'item_id' : item_id , 'output_index' : out_idx , 'content_index' : 0 , 'delta' : text })} \n \n " .encode ("utf-8" )
279- )
280- await stream_res .write (
281- f"data: { json .dumps ({'type' : 'response.output_text.done' , 'item_id' : item_id , 'output_index' : out_idx , 'content_index' : 0 , 'text' : text })} \n \n " .encode ("utf-8" )
282- )
283- await stream_res .write (
284- f"data: { json .dumps ({'type' : 'response.output_item.done' , 'output_index' : out_idx , 'item' : {** item_dict , 'status' : 'completed' }})} \n \n " .encode ("utf-8" )
285- )
286- elif itype in ("function_call" , "custom_tool_call" ):
344+ if itype in ("function_call" , "custom_tool_call" ):
287345 arguments = str (item .get ("arguments" ) or item .get ("input" ) or "{}" )
288346 await stream_res .write (
289347 f"data: { json .dumps ({'type' : 'response.output_item.added' , 'output_index' : out_idx , 'item' : {** item_dict , 'arguments' : '' , 'status' : 'in_progress' }})} \n \n " .encode ("utf-8" )
0 commit comments