fix(gateway): resolve reattach URL and key as one pair; fail fast on 401/403
Drop the persisted base_url: pairing a saved URL with the profile's current key could poll the old gateway with a new credential. Both now come from the session profile's current config, and an auth rejection settles the turn at once with an explicit error instead of 150 retries.
This commit is contained in:
1 parent
65e336fb16
commit
94b07588dd
2 files changed
+20
-18
No files matched your search
+6
-8
@@ -895,6 +895,8 @@ def _await_gateway_run_result(
|
|||||||
except urllib.error.HTTPError as exc:
|
except urllib.error.HTTPError as exc:
|
||||||
if exc.code == 404:
|
if exc.code == 404:
|
||||||
raise RuntimeError("Gateway no longer has this run; its result could not be recovered after the WebUI restart.") from exc
|
raise RuntimeError("Gateway no longer has this run; its result could not be recovered after the WebUI restart.") from exc
|
||||||
|
if exc.code in (401, 403):
|
||||||
|
raise RuntimeError(f"Gateway rejected the WebUI credentials (HTTP {exc.code}) while reattaching to a run after the restart.") from exc
|
||||||
failures += 1
|
failures += 1
|
||||||
status = None
|
status = None
|
||||||
except (urllib.error.URLError, OSError, ValueError):
|
except (urllib.error.URLError, OSError, ValueError):
|
||||||
@@ -982,18 +984,15 @@ def _sidecars_with_active_stream(session_dir) -> list[str]:
|
|||||||
return ids
|
return ids
|
||||||
|
|
||||||
|
|
||||||
def _gateway_endpoint_for_reattach(session, run: dict) -> tuple[str, str]:
|
def _gateway_endpoint_for_reattach(session) -> tuple[str, str]:
|
||||||
"""Resolve the gateway that owns ``run`` using the session's profile, not the process default."""
|
"""Resolve gateway URL and key together from the session's profile, never the process default."""
|
||||||
from api import profiles as _profiles
|
from api import profiles as _profiles
|
||||||
from api.config import get_config
|
from api.config import get_config
|
||||||
|
|
||||||
with _profiles.profile_scope_for_detached_worker(
|
with _profiles.profile_scope_for_detached_worker(
|
||||||
getattr(session, "profile", None), "gateway reattach", logger_override=logger,
|
getattr(session, "profile", None), "gateway reattach", logger_override=logger,
|
||||||
):
|
):
|
||||||
base_url = _gateway_base_url(get_config())
|
return _gateway_base_url(get_config()), _gateway_api_key()
|
||||||
api_key = _gateway_api_key()
|
|
||||||
# The run lives on the gateway that accepted it, even if config changed since.
|
|
||||||
return str(run.get("base_url") or base_url).rstrip("/"), api_key
|
|
||||||
|
|
||||||
|
|
||||||
def _resume_gateway_run_for_session(session) -> bool:
|
def _resume_gateway_run_for_session(session) -> bool:
|
||||||
@@ -1005,7 +1004,7 @@ def _resume_gateway_run_for_session(session) -> bool:
|
|||||||
if not stream_id or not run_id or run.get("stream_id") != stream_id:
|
if not stream_id or not run_id or run.get("stream_id") != stream_id:
|
||||||
return False
|
return False
|
||||||
sid = session.session_id
|
sid = session.session_id
|
||||||
endpoint = _gateway_endpoint_for_reattach(session, run)
|
endpoint = _gateway_endpoint_for_reattach(session)
|
||||||
with STREAMS_LOCK:
|
with STREAMS_LOCK:
|
||||||
if stream_id in STREAMS:
|
if stream_id in STREAMS:
|
||||||
return False
|
return False
|
||||||
@@ -1318,7 +1317,6 @@ def _run_gateway_chat_streaming(
|
|||||||
active_provider=(model_provider or ""),
|
active_provider=(model_provider or ""),
|
||||||
on_run_id=lambda run_id: _record_gateway_run(
|
on_run_id=lambda run_id: _record_gateway_run(
|
||||||
session_id, stream_id, run_id,
|
session_id, stream_id, run_id,
|
||||||
base_url=base_url,
|
|
||||||
regeneration=bool(regeneration),
|
regeneration=bool(regeneration),
|
||||||
goal_related=bool(goal_related),
|
goal_related=bool(goal_related),
|
||||||
),
|
),
|
||||||
|
|||||||
@@ -113,7 +113,7 @@ def test_runs_api_start_sends_idempotency_key_and_persists_run_id(isolated_sessi
|
|||||||
assert captured["post_headers"]["Idempotency-key"] == f"webui-{stream_id}"
|
assert captured["post_headers"]["Idempotency-key"] == f"webui-{stream_id}"
|
||||||
assert captured["persisted_at_events"]["run_id"] == "run_live"
|
assert captured["persisted_at_events"]["run_id"] == "run_live"
|
||||||
assert captured["persisted_at_events"]["stream_id"] == stream_id
|
assert captured["persisted_at_events"]["stream_id"] == stream_id
|
||||||
assert captured["persisted_at_events"]["base_url"] == "http://gateway.local"
|
assert "api_key" not in captured["persisted_at_events"]
|
||||||
saved = json.loads((isolated_sessions / f"{s.session_id}.json").read_text())
|
saved = json.loads((isolated_sessions / f"{s.session_id}.json").read_text())
|
||||||
assert saved["gateway_run"] is None
|
assert saved["gateway_run"] is None
|
||||||
assert saved["active_stream_id"] is None
|
assert saved["active_stream_id"] is None
|
||||||
@@ -348,16 +348,20 @@ def test_reattach_resolves_gateway_from_the_session_profile(isolated_sessions, m
|
|||||||
assert saved["messages"][-1]["content"] == "answer"
|
assert saved["messages"][-1]["content"] == "answer"
|
||||||
|
|
||||||
|
|
||||||
def test_reattach_polls_the_gateway_that_accepted_the_run(isolated_sessions, monkeypatch):
|
@pytest.mark.parametrize("code", [401, 403])
|
||||||
sid, stream_id = _orphaned_gateway_turn()
|
def test_reattach_auth_rejection_fails_fast_with_an_auth_error(isolated_sessions, monkeypatch, code):
|
||||||
s = models.Session.load(sid)
|
sid, _stream_id = _orphaned_gateway_turn()
|
||||||
s.gateway_run = {**s.gateway_run, "base_url": "http://accepted-gateway:8642/"}
|
calls = []
|
||||||
s.save(touch_updated_at=False)
|
|
||||||
models.SESSIONS.clear()
|
|
||||||
seen = []
|
|
||||||
_poll_until_completed(monkeypatch, seen)
|
|
||||||
|
|
||||||
|
def rejected(base_url, api_key, run_id):
|
||||||
|
calls.append(run_id)
|
||||||
|
raise urllib.error.HTTPError("http://gateway.local/v1/runs/x", code, "denied", Message(), io.BytesIO(b""))
|
||||||
|
|
||||||
|
monkeypatch.setattr(gateway_chat, "_get_gateway_run_status", rejected)
|
||||||
gateway_chat.resume_gateway_runs_after_restart()
|
gateway_chat.resume_gateway_runs_after_restart()
|
||||||
_wait_for_reattach_threads()
|
_wait_for_reattach_threads()
|
||||||
|
|
||||||
assert {b for b, _k, _r in seen} == {"http://accepted-gateway:8642"}
|
assert len(calls) == 1
|
||||||
|
saved = json.loads((isolated_sessions / f"{sid}.json").read_text())
|
||||||
|
assert saved["active_stream_id"] is None and saved["gateway_run"] is None
|
||||||
|
assert f"HTTP {code}" in json.dumps(saved["messages"][-1])
|
||||||
Reference in new issue
Block a user