admin/Codex-Launcher---Any-AI-Porovider
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

v3.13.5: anti-loop resilience, auto 401 refresh, budget cap 150, cobra PR #17

Browse Source
This commit is contained in:
Roman | RyzenAdvanced
2026-05-27 15:32:55 +04:00
Unverified
parent f7baff7425
commit 5e1f4c6ca1
6 changed files with 635 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files

379
translate-proxy.py
View File

@@ -380,6 +380,14 @@ _conn_pool_lock = threading.Lock()
_conn_pool = {}
_STREAM_IDLE_TIMEOUT = 300
def _idle_timeout_for_model(model, default=300):
if not model:
return default
m = model.lower()
if "flash" in m or "mini" in m or "haiku" in m:
return 120
return default
_MAX_CONCURRENT_REQUESTS = 3
_request_semaphore = threading.Semaphore(_MAX_CONCURRENT_REQUESTS)
@@ -779,6 +787,20 @@ def _refresh_google_token(token_data, token_path):
print(f"[oauth] refresh failed: {e}", file=sys.stderr)
return token_data.get("access_token", "")
def _force_refresh_google_token():
token_path = os.path.join(os.path.expanduser("~"), ".cache", "codex-proxy",
"google-antigravity-oauth-token.json" if OAUTH_PROVIDER == "google-antigravity"
else "google-oauth-token.json")
try:
with open(token_path) as f:
token_data = json.load(f)
token_data["expires_at"] = 0
new_token = _refresh_google_token(token_data, token_path)
return bool(new_token)
except Exception as e:
print(f"[oauth] force refresh failed: {e}", file=sys.stderr)
return False
# ═══════════════════════════════════════════════════════════════════
# Gemini 3 thought signature preservation
# ═══════════════════════════════════════════════════════════════════
@@ -846,7 +868,12 @@ _GEMINI_AGENT_GUARDRAIL = (
_LOG_FILE_LOCK = threading.Lock()
_ANTIGRAVITY_LOOP_TRACKER = {}
_ANTIGRAVITY_LOOP_TRACKER_LOCK = threading.Lock()
def _antigravity_loop_key(session_id):
_ANTIGRAVITY_FILE_TRACKER = {}
_ANTIGRAVITY_MAX_TOOL_CALLS_PER_TASK = 150
_ANTIGRAVITY_WARN_TOOL_CALLS_PER_TASK = 80
def _antigravity_loop_key(session_id, user_request_hash=None):
if user_request_hash:
return f"ag:task:{user_request_hash}"
return f"ag:{session_id}"
def _validate_antigravity_version(version, access_token=None, project_id=None):
@@ -4925,7 +4952,7 @@ def _auto_continue_gemini(handler, flush_event, message_id, model, gen_config, g
cont_text = ""
cont_finish = ""
cont_buf = ""
for raw_line in _stream_with_idle_timeout(upstream):
for raw_line in _stream_with_idle_timeout(upstream, _idle_timeout_for_model(model)):
line = raw_line.decode(errors="replace")
if line.startswith("data: "):
cont_buf += line[6:]
@@ -5122,7 +5149,20 @@ def _antigravity_normalize_context(input_data, model=""):
compaction_summaries.append(msg_item)
if n_summarized > 0:
summary_text = f"[Tool history summary: {n_summarized} older tool outputs omitted. {n_tool_calls} prior function calls were made for file inspection/editing.]"
n_read_calls = sum(1 for it in input_data if isinstance(it, dict) and it.get("type") == "function_call"
and it.get("name", "") not in ("write", "apply_diff", "edit_file")
and "write" not in json.dumps(it.get("arguments", {})).lower())
n_write_calls = n_tool_calls - n_read_calls
if n_read_calls > 10 and n_write_calls == 0:
summary_text = (
f"[CONTEXT HISTORY: {n_summarized} prior tool calls compacted. "
f"YOU ALREADY READ THE TARGET FILE EXTENSIVELY. "
f"DO NOT READ ANY MORE FILES. "
f"YOU MUST NOW USE THE WRITE TOOL TO APPLY YOUR EDITS DIRECTLY. "
f"DO NOT call exec_command or read_files AGAIN.]"
)
else:
summary_text = f"[Tool history summary: {n_summarized} older tool outputs omitted. {n_tool_calls} prior function calls were made.]"
result.append({"type": "message", "role": "user", "content": [{"type": "input_text", "text": summary_text}]})
# CRITICAL: Add tool CALLS and their corresponding OUTPUTS in PAIRED ORDER
@@ -5744,10 +5784,12 @@ class Handler(http.server.BaseHTTPRequestHandler):
"latest_user_hash": None, "nudge_injected": False, "latest_user_appended": False,
"tool_calls_for_request": 0, "repeated_tool": False, "force_finalize": False,
"last_tool": None, "last_tool_count": 0,
"task_retry_count": 0, "total_tool_calls": 0, "first_seen": time.time(),
}
ag_state = _ANTIGRAVITY_LOOP_TRACKER[ag_key]
latest_user = ""
latest_user_hash = None
if isinstance(input_data, list):
for item in reversed(input_data):
if item.get("type") == "message" and item.get("role") == "user":
@@ -5760,17 +5802,91 @@ class Handler(http.server.BaseHTTPRequestHandler):
if latest_user:
latest_norm = " ".join(latest_user.strip().split())[:200]
latest_user_hash = hashlib.sha256(latest_norm.encode()).hexdigest()[:16]
if latest_user_hash != ag_state.get("latest_user_hash"):
ag_state["latest_user_hash"] = latest_user_hash
ag_state["nudge_injected"] = False
ag_state["latest_user_appended"] = False
ag_state["tool_calls_for_request"] = 0
ag_state["repeated_tool"] = False
ag_state["force_finalize"] = False
ag_state["last_tool"] = None
ag_state["last_tool_count"] = 0
# Cross-session key: stable across retries for same task
if latest_user_hash:
task_key = _antigravity_loop_key(self._session_id, latest_user_hash)
else:
task_key = ag_key
if task_key != ag_key:
with _ANTIGRAVITY_LOOP_TRACKER_LOCK:
if task_key not in _ANTIGRAVITY_LOOP_TRACKER:
_ANTIGRAVITY_LOOP_TRACKER[task_key] = dict(_ANTIGRAVITY_LOOP_TRACKER.get(ag_key, {
"latest_user_hash": None, "nudge_injected": False, "latest_user_appended": False,
"tool_calls_for_request": 0, "repeated_tool": False, "force_finalize": False,
"last_tool": None, "last_tool_count": 0,
"task_retry_count": 0, "total_tool_calls": 0, "first_seen": time.time(),
}))
ag_state = _ANTIGRAVITY_LOOP_TRACKER[task_key]
ag_key = task_key
with _ANTIGRAVITY_LOOP_TRACKER_LOCK:
if latest_user_hash and latest_user_hash != ag_state.get("latest_user_hash"):
ag_state["latest_user_hash"] = latest_user_hash
ag_state["nudge_injected"] = False
ag_state["latest_user_appended"] = False
ag_state["tool_calls_for_request"] = 0
ag_state["repeated_tool"] = False
ag_state["last_tool"] = None
ag_state["last_tool_count"] = 0
ag_state["task_retry_count"] = 1
ag_state["total_tool_calls"] = 0
ag_state["first_seen"] = time.time()
ag_state["force_finalize"] = False
else:
ag_state["task_retry_count"] = ag_state.get("task_retry_count", 0) + 1
# Cross-session retry cap — only fires when same task retried many times
if ag_state.get("task_retry_count", 0) >= 15:
ag_state["task_retry_count"] = 0
ag_state["force_finalize"] = False
return self._send_ag_finalize(
"Task retry limit reached. Breaking out of loop. "
"Try a more specific or smaller request if needed.",
stream=body.get("stream", False))
if ag_state.get("task_retry_count", 0) >= 8:
ag_state["force_finalize"] = True
if isinstance(input_data, list):
n_tool_calls = sum(1 for it in input_data if isinstance(it, dict) and it.get("type") == "function_call")
ag_state["tool_calls_for_request"] = n_tool_calls
cumulative_calls = ag_state.get("total_tool_calls", 0) + n_tool_calls
ag_state["total_tool_calls"] = cumulative_calls
if cumulative_calls > _ANTIGRAVITY_MAX_TOOL_CALLS_PER_TASK:
print(f"[{getattr(self, '_session_id', '?')}] [antigravity-budget] HARD CAP: {cumulative_calls} calls, injecting force-write directive", file=sys.stderr)
contents.append({"role": "user", "parts": [{"text":
f"CRITICAL BUDGET LIMIT: {cumulative_calls} tool calls made. "
f"YOU MUST STOP NOW. Do NOT call any more tools. "
f"Write your FINAL answer immediately using the information you already have. "
f"If you have file edits, apply them in this response using exec_command with a write command. "
f"DO NOT READ ANY MORE FILES."}]})
elif cumulative_calls > _ANTIGRAVITY_WARN_TOOL_CALLS_PER_TASK:
contents.append({"role": "user", "parts": [{"text":
f"WARNING: {cumulative_calls} tool calls made. "
f"{_ANTIGRAVITY_MAX_TOOL_CALLS_PER_TASK - cumulative_calls} remaining before forced stop. "
f"STOP READING FILES AND APPLY YOUR EDITS NOW."}]})
# CHANGE 2: File-path read-loop detection
if ag_key not in _ANTIGRAVITY_FILE_TRACKER:
_ANTIGRAVITY_FILE_TRACKER[ag_key] = {"last_path": None, "path_counts": {}, "total_reads": 0}
ft = _ANTIGRAVITY_FILE_TRACKER[ag_key]
for item in reversed(input_data):
if isinstance(item, dict) and item.get("type") == "function_call":
args_str = json.dumps(item.get("arguments", {}))
file_match = re.search(r'(/[\w/.-]+\.(?:html|py|js|ts|css|json|md|yaml|yml|xml|txt|sh))', args_str)
if file_match:
detected_path = file_match.group(1)
ft["total_reads"] += 1
ft["path_counts"][detected_path] = ft["path_counts"].get(detected_path, 0) + 1
ft["last_path"] = detected_path
if ft["path_counts"][detected_path] >= 5 or ft["total_reads"] > 30:
ag_state["force_finalize"] = True
print(f"[antigravity-loop] FILE READ LOOP: {detected_path} read "
f"{ft['path_counts'][detected_path]}x, total={ft['total_reads']}",
file=sys.stderr)
break
last_tool_key = None
for item in reversed(input_data):
if isinstance(item, dict) and item.get("type") == "function_call":
@@ -5893,6 +6009,23 @@ class Handler(http.server.BaseHTTPRequestHandler):
return self.send_json(e.code, {"error": {"type": "upstream_error", "message": _sanitize_err_body(err_body)}})
if err_class in ("auth_permanent", "forbidden", "account_banned", "validation_required"):
return self.send_json(e.code, {"error": {"type": "upstream_error", "message": _sanitize_err_body(err_body)}})
if err_class == "auth_transient":
print(f"[{self._session_id}] [antigravity-v2] 401 transient, force-refreshing token", file=sys.stderr)
try:
_force_refresh_google_token()
access_token = _refresh_oauth_token()
headers["Authorization"] = f"Bearer {access_token}"
new_body_b = json.dumps(wrapped).encode()
retry_req = urllib.request.Request(target, data=new_body_b, headers=headers)
upstream = urllib.request.urlopen(retry_req, timeout=_upstream_timeout(body, stream))
chosen_ep = ep
with _antigravity_endpoint_lock:
_antigravity_preferred_endpoint = ep
print(f"[{self._session_id}] [antigravity-v2] 401 retry succeeded", file=sys.stderr)
break
except Exception as retry_e:
print(f"[{self._session_id}] [antigravity-v2] 401 retry failed: {retry_e}", file=sys.stderr)
return self.send_json(e.code, {"error": {"type": "upstream_error", "message": _sanitize_err_body(err_body)}})
if err_class == "service_disabled":
_is_prod = "cloudcode-pa.googleapis.com" in ep and "sandbox" not in ep
if _is_prod:
@@ -6449,66 +6582,135 @@ class Handler(http.server.BaseHTTPRequestHandler):
if not is_latest_simple:
contents.insert(0, {"role": "user", "parts": [{"text": _GEMINI_AGENT_GUARDRAIL}]})
if OAUTH_PROVIDER == "google-antigravity":
import hashlib
ag_key = _antigravity_loop_key(self._session_id)
with _ANTIGRAVITY_LOOP_TRACKER_LOCK:
if ag_key not in _ANTIGRAVITY_LOOP_TRACKER:
_ANTIGRAVITY_LOOP_TRACKER[ag_key] = {
"latest_user_hash": None,
"nudge_injected": False,
"latest_user_appended": False,
"tool_calls_for_request": 0,
"repeated_tool": False,
"force_finalize": False,
"last_tool": None,
"last_tool_count": 0,
}
ag_state = _ANTIGRAVITY_LOOP_TRACKER[ag_key]
if OAUTH_PROVIDER == "google-antigravity":
import hashlib
ag_key = _antigravity_loop_key(self._session_id)
with _ANTIGRAVITY_LOOP_TRACKER_LOCK:
if ag_key not in _ANTIGRAVITY_LOOP_TRACKER:
_ANTIGRAVITY_LOOP_TRACKER[ag_key] = {
"latest_user_hash": None,
"nudge_injected": False,
"latest_user_appended": False,
"tool_calls_for_request": 0,
"repeated_tool": False,
"force_finalize": False,
"last_tool": None,
"last_tool_count": 0,
"task_retry_count": 0,
"total_tool_calls": 0,
"first_seen": time.time(),
}
ag_state = _ANTIGRAVITY_LOOP_TRACKER[ag_key]
latest_user = ""
latest_user_hash = None
if isinstance(input_data, list):
for item in reversed(input_data):
if item.get("type") == "message" and item.get("role") == "user":
c = item.get("content", "")
if isinstance(c, str):
latest_user = c
elif isinstance(c, list):
latest_user = "\n".join(p.get("text", p.get("input_text", "")) for p in c if isinstance(p, dict))
break
if latest_user:
latest_norm = " ".join(latest_user.strip().split())[:200]
latest_user_hash = hashlib.sha256(latest_norm.encode()).hexdigest()[:16]
if latest_user_hash != ag_state["latest_user_hash"]:
latest_user = ""
latest_user_hash = None
if isinstance(input_data, list):
for item in reversed(input_data):
if item.get("type") == "message" and item.get("role") == "user":
c = item.get("content", "")
if isinstance(c, str):
latest_user = c
elif isinstance(c, list):
latest_user = "\n".join(p.get("text", p.get("input_text", "")) for p in c if isinstance(p, dict))
break
if latest_user:
latest_norm = " ".join(latest_user.strip().split())[:200]
latest_user_hash = hashlib.sha256(latest_norm.encode()).hexdigest()[:16]
if latest_user_hash:
task_key = _antigravity_loop_key(self._session_id, latest_user_hash)
else:
task_key = ag_key
if task_key != ag_key:
with _ANTIGRAVITY_LOOP_TRACKER_LOCK:
if task_key not in _ANTIGRAVITY_LOOP_TRACKER:
_ANTIGRAVITY_LOOP_TRACKER[task_key] = dict(_ANTIGRAVITY_LOOP_TRACKER.get(ag_key, {
"latest_user_hash": None, "nudge_injected": False,
"latest_user_appended": False, "tool_calls_for_request": 0,
"repeated_tool": False, "force_finalize": False,
"last_tool": None, "last_tool_count": 0,
"task_retry_count": 0, "total_tool_calls": 0, "first_seen": time.time(),
}))
ag_state = _ANTIGRAVITY_LOOP_TRACKER[task_key]
ag_key = task_key
with _ANTIGRAVITY_LOOP_TRACKER_LOCK:
if latest_user_hash and latest_user_hash != ag_state.get("latest_user_hash"):
ag_state["latest_user_hash"] = latest_user_hash
ag_state["nudge_injected"] = False
ag_state["latest_user_appended"] = False
ag_state["tool_calls_for_request"] = 0
ag_state["repeated_tool"] = False
ag_state["force_finalize"] = False
ag_state["last_tool"] = None
ag_state["last_tool_count"] = 0
if isinstance(input_data, list):
n_tool_calls = sum(1 for it in input_data if isinstance(it, dict) and it.get("type") == "function_call")
ag_state["tool_calls_for_request"] = n_tool_calls
last_tool_key = None
for item in reversed(input_data):
if isinstance(item, dict) and item.get("type") == "function_call":
fname = item.get("name", "")
args_str = json.dumps(item.get("arguments", {}), sort_keys=True)[:100]
last_tool_key = f"{fname}:{args_str}"
break
if last_tool_key:
if last_tool_key == ag_state["last_tool"]:
ag_state["last_tool_count"] += 1
if ag_state["last_tool_count"] >= 5:
ag_state["repeated_tool"] = True
ag_state["force_finalize"] = True
ag_state["task_retry_count"] = 1
ag_state["total_tool_calls"] = 0
ag_state["first_seen"] = time.time()
ag_state["force_finalize"] = False
else:
ag_state["last_tool"] = last_tool_key
ag_state["last_tool_count"] = 1
ag_state["task_retry_count"] = ag_state.get("task_retry_count", 0) + 1
if ag_state.get("task_retry_count", 0) >= 15:
ag_state["task_retry_count"] = 0
ag_state["force_finalize"] = False
self._send_ag_finalize("Task retry limit reached. Breaking loop.",
stream=body.get("stream", False) if isinstance(body, dict) else False)
return
if ag_state.get("task_retry_count", 0) >= 8:
ag_state["force_finalize"] = True
if isinstance(input_data, list):
n_tool_calls = sum(1 for it in input_data if isinstance(it, dict) and it.get("type") == "function_call")
ag_state["tool_calls_for_request"] = n_tool_calls
cumulative_calls = ag_state.get("total_tool_calls", 0) + n_tool_calls
ag_state["total_tool_calls"] = cumulative_calls
if cumulative_calls > _ANTIGRAVITY_MAX_TOOL_CALLS_PER_TASK:
print(f"[antigravity-budget] HARD CAP: {cumulative_calls} calls, injecting force-write", file=sys.stderr)
contents.append({"role": "user", "parts": [{"text":
f"CRITICAL BUDGET LIMIT: {cumulative_calls} tool calls. "
f"STOP ALL TOOL CALLS. Write your FINAL answer now. "
f"Apply any edits using exec_command with a write command in this response."}]})
elif cumulative_calls > _ANTIGRAVITY_WARN_TOOL_CALLS_PER_TASK:
contents.append({"role": "user", "parts": [{"text":
f"WARNING: {cumulative_calls} tool calls. "
f"{_ANTIGRAVITY_MAX_TOOL_CALLS_PER_TASK - cumulative_calls} remaining. "
f"STOP READING AND WRITE NOW."}]})
if ag_key not in _ANTIGRAVITY_FILE_TRACKER:
_ANTIGRAVITY_FILE_TRACKER[ag_key] = {"last_path": None, "path_counts": {}, "total_reads": 0}
ft = _ANTIGRAVITY_FILE_TRACKER[ag_key]
for item in reversed(input_data):
if isinstance(item, dict) and item.get("type") == "function_call":
args_str = json.dumps(item.get("arguments", {}))
file_match = re.search(r'(/[\w/.-]+\.(?:html|py|js|ts|css|json|md|yaml|yml|xml|txt|sh))', args_str)
if file_match:
dp = file_match.group(1)
ft["total_reads"] += 1
ft["path_counts"][dp] = ft["path_counts"].get(dp, 0) + 1
ft["last_path"] = dp
if ft["path_counts"][dp] >= 5 or ft["total_reads"] > 30:
ag_state["force_finalize"] = True
print(f"[antigravity-loop] FILE READ LOOP: {dp} read "
f"{ft['path_counts'][dp]}x, total={ft['total_reads']}", file=sys.stderr)
break
last_tool_key = None
for item in reversed(input_data):
if isinstance(item, dict) and item.get("type") == "function_call":
fname = item.get("name", "")
args_str = json.dumps(item.get("arguments", {}), sort_keys=True)[:100]
last_tool_key = f"{fname}:{args_str}"
break
if last_tool_key:
if last_tool_key == ag_state["last_tool"]:
ag_state["last_tool_count"] += 1
if ag_state["last_tool_count"] >= 5:
ag_state["repeated_tool"] = True
ag_state["force_finalize"] = True
else:
ag_state["last_tool"] = last_tool_key
ag_state["last_tool_count"] = 1
_EDIT_WORDS = ("change", "fix", "update", "redesign", "rewrite", "modify", "improve", "replace", "edit", "make it", "add", "remove", "delete", "rename", "move", "convert")
latest_lower = ""
@@ -6671,6 +6873,23 @@ class Handler(http.server.BaseHTTPRequestHandler):
return self.send_json(e.code, {"error": {"type": "upstream_error", "message": _sanitize_err_body(err_body)}})
if err_class == "auth_permanent":
return self.send_json(e.code, {"error": {"type": "upstream_error", "message": _sanitize_err_body(err_body)}})
if err_class == "auth_transient":
print(f"[{self._session_id}] {ep.replace('https://','')} 401 transient, force-refreshing token and retrying", file=sys.stderr)
try:
_force_refresh_google_token()
access_token = _refresh_oauth_token()
headers["Authorization"] = f"Bearer {access_token}"
new_body_b = json.dumps(wrapped).encode()
retry_req = urllib.request.Request(target, data=new_body_b, headers=headers)
upstream = urllib.request.urlopen(retry_req, timeout=_upstream_timeout(body, stream))
chosen_ep = ep
with _antigravity_endpoint_lock:
_antigravity_preferred_endpoint = ep
print(f"[{self._session_id}] 401 retry succeeded after token refresh", file=sys.stderr)
break
except Exception as retry_e:
print(f"[{self._session_id}] 401 retry also failed: {retry_e}", file=sys.stderr)
return self.send_json(e.code, {"error": {"type": "upstream_error", "message": _sanitize_err_body(err_body)}})
if err_class in ("quota_exhausted", "rate_limited"):
reset_s = _parse_rate_limit_reset(err_body)
if ep == ordered[-1]:
@@ -6730,7 +6949,7 @@ class Handler(http.server.BaseHTTPRequestHandler):
buf = ""
stream_finished = False
for raw_line in _stream_with_idle_timeout(upstream):
for raw_line in _stream_with_idle_timeout(upstream, _idle_timeout_for_model(model)):
if tracker and tracker.cancelled.is_set():
print("[gemini-oauth] stream cancelled", file=sys.stderr)
break
@@ -8144,6 +8363,38 @@ class Handler(http.server.BaseHTTPRequestHandler):
except (BrokenPipeError, ConnectionResetError, ConnectionAbortedError):
pass
def _send_ag_finalize(self, text, stream=False, is_responses_api=True):
sid = getattr(self, '_session_id', 'fin')
print(f"[{sid}] [antigravity-finalize] Sending finalize response: {text[:80]}...", file=sys.stderr)
resp_id = f"resp_{uuid.uuid4().hex[:12]}"
msg_id = f"msg_{uuid.uuid4().hex[:12]}"
if is_responses_api:
output_obj = [{"type": "message", "id": msg_id, "role": "assistant",
"content": [{"type": "output_text", "text": text}]}]
if stream:
events = [
f"event: response.created\ndata: {json.dumps({'type':'response.created','response':{'id':resp_id,'object':'response','status':'in_progress'}})}\n\n",
f"event: response.output_item.added\ndata: {json.dumps({'type':'response.output_item.added','output_index':0,'item':{'type':'message','id':msg_id,'role':'assistant','content':[]}})}\n\n",
f"event: response.content_part.added\ndata: {json.dumps({'type':'response.content_part.added','output_index':0,'content_index':0,'part':{'type':'output_text','text':''}})}\n\n",
f"event: response.output_text.delta\ndata: {json.dumps({'type':'response.output_text.delta','output_index':0,'content_index':0,'delta':text})}\n\n",
f"event: response.output_text.done\ndata: {json.dumps({'type':'response.output_text.done','output_index':0,'content_index':0,'text':text})}\n\n",
f"event: response.content_part.done\ndata: {json.dumps({'type':'response.content_part.done','output_index':0,'content_index':0,'part':{'type':'output_text','text':text}})}\n\n",
f"event: response.output_item.done\ndata: {json.dumps({'type':'response.output_item.done','output_index':0,'item':{'type':'message','id':msg_id,'role':'assistant','content':[{'type':'output_text','text':text}]}})}\n\n",
f"event: response.completed\ndata: {json.dumps({'type':'response.completed','response':{'id':resp_id,'object':'response','status':'completed','output':output_obj}})}\n\n",
]
self.send_response(200)
self.send_header("Content-Type", "text/event-stream")
self.send_header("Cache-Control", "no-cache")
self.send_header("Connection", "keep-alive")
self.end_headers()
for evt in events:
self.wfile.write(evt.encode())
self.wfile.flush()
else:
self.send_json(200, {"id": resp_id, "object": "response", "status": "completed",
"output": output_obj, "model": "gemini-3-flash"})
return None
def stream_buffered_events(self, event_iter, flush_interval=0.03, max_bytes=4096, on_event=None):
buf = bytearray()
last_flush = time.monotonic()