fix(gateway): scope approval mirror cleanup to live heads (#6008)
This commit is contained in:
parent
39fc169834
commit
489728d71f
2 changed files with 36 additions and 6 deletions
|
|
@ -148,6 +148,8 @@ def reconcile_gateway_pending_mirror_locked(session_key: str) -> tuple[dict | No
|
|||
live_head_entry = live_gateway_queue[0] if live_gateway_queue else None
|
||||
live_head_data = getattr(live_head_entry, "data", None) or {}
|
||||
live_token = _gateway_mirror_entry_token(live_head_entry) if live_head_entry and live_head_data else None
|
||||
live_run_id = str(live_head_data.get("run_id") or "").strip()
|
||||
live_approval_id = str(live_head_data.get("approval_id") or "").strip()
|
||||
|
||||
rebuilt: list[dict] = []
|
||||
live_mirror_present = False
|
||||
|
|
@ -155,13 +157,40 @@ def reconcile_gateway_pending_mirror_locked(session_key: str) -> tuple[dict | No
|
|||
if not _is_gateway_mirror_entry(entry):
|
||||
rebuilt.append(entry)
|
||||
continue
|
||||
if str(entry.get("run_id") or "").strip():
|
||||
rebuilt.append(entry)
|
||||
entry_run_id = str(entry.get("run_id") or "").strip()
|
||||
entry_approval_id = str(entry.get("approval_id") or "").strip()
|
||||
entry_token = str(entry.get(_GATEWAY_MIRROR_TOKEN) or "").strip()
|
||||
matches_live_head = False
|
||||
if live_token:
|
||||
if entry_token and entry_token == live_token:
|
||||
matches_live_head = True
|
||||
elif live_run_id and entry_run_id == live_run_id:
|
||||
matches_live_head = True
|
||||
elif live_approval_id and entry_approval_id == live_approval_id:
|
||||
matches_live_head = True
|
||||
|
||||
if entry_run_id:
|
||||
if matches_live_head and not live_mirror_present:
|
||||
if entry_token != live_token:
|
||||
entry[_GATEWAY_MIRROR_TOKEN] = live_token
|
||||
changed = True
|
||||
rebuilt.append(entry)
|
||||
live_mirror_present = True
|
||||
continue
|
||||
if not entry_token:
|
||||
rebuilt.append(entry)
|
||||
continue
|
||||
changed = True
|
||||
continue
|
||||
if live_token and entry.get(_GATEWAY_MIRROR_TOKEN) == live_token and not live_mirror_present:
|
||||
|
||||
if matches_live_head and not live_mirror_present:
|
||||
if entry_token != live_token:
|
||||
entry[_GATEWAY_MIRROR_TOKEN] = live_token
|
||||
changed = True
|
||||
rebuilt.append(entry)
|
||||
live_mirror_present = True
|
||||
continue
|
||||
|
||||
changed = True
|
||||
|
||||
if live_token and not live_mirror_present:
|
||||
|
|
|
|||
|
|
@ -426,13 +426,14 @@ def test_empty_id_runs_approval_reaches_real_response_lifecycle():
|
|||
import api.routes as routes
|
||||
|
||||
for choice in ("once", "deny"):
|
||||
choice_bytes = choice.encode()
|
||||
sid = f"sess-real-{choice}"
|
||||
stream_id = f"stream-real-{choice}"
|
||||
events = []
|
||||
|
||||
class _JsonResponse:
|
||||
def read(self, _limit=None):
|
||||
return b'{"run_id":"run-real-' + choice.encode() + b'"}'
|
||||
def read(self, _limit=None, _choice_bytes=choice_bytes):
|
||||
return b'{"run_id":"run-real-' + _choice_bytes + b'"}'
|
||||
def __enter__(self):
|
||||
return self
|
||||
def __exit__(self, *args):
|
||||
|
|
@ -462,7 +463,7 @@ def test_empty_id_runs_approval_reaches_real_response_lifecycle():
|
|||
patch("api.runner_client.HttpRunnerClient._request_json", return_value={"ok": True}):
|
||||
_run_gateway_runs_api_streaming(
|
||||
sid, "hi", "test", "/tmp", stream_id, "http://gw:8642", "", [], {},
|
||||
put_gateway_event=lambda event, data: events.append((event, data)),
|
||||
put_gateway_event=lambda event, data, _events=events: _events.append((event, data)),
|
||||
cancel_event=threading.Event(),
|
||||
)
|
||||
approval_id = events[0][1]["approval_id"]
|
||||
|
|
|
|||
Loading…
Reference in a new issue