fix(gateway): preserve approval identity provenance
This commit is contained in:
1 parent
f30178e18a
commit
a91bd88c70
5 files changed
+38
-1
No files matched your search
@@ -505,6 +505,8 @@ def _run_gateway_runs_api_streaming(
|
|||||||
approval_data = _gateway_runs_approval_event(payload)
|
approval_data = _gateway_runs_approval_event(payload)
|
||||||
if approval_data:
|
if approval_data:
|
||||||
approval_data["run_id"] = run_id
|
approval_data["run_id"] = run_id
|
||||||
|
from api.config import gateway_supports_approval_identity_v1
|
||||||
|
approval_data["_gateway_agent_identity_v1"] = gateway_supports_approval_identity_v1(base_url, api_key)
|
||||||
from api.route_approvals import submit_gateway_pending_mirror
|
from api.route_approvals import submit_gateway_pending_mirror
|
||||||
head, total = submit_gateway_pending_mirror(session_id, approval_data)
|
head, total = submit_gateway_pending_mirror(session_id, approval_data)
|
||||||
put_gateway_event("approval", {**(head or approval_data), "pending_count": total})
|
put_gateway_event("approval", {**(head or approval_data), "pending_count": total})
|
||||||
|
|||||||
@@ -47,6 +47,7 @@ _approval_sse_subscribers: dict[str, list[queue.Queue]] = {}
|
|||||||
_GATEWAY_MIRROR_FLAG = "_gateway_mirror"
|
_GATEWAY_MIRROR_FLAG = "_gateway_mirror"
|
||||||
_GATEWAY_MIRROR_TOKEN = "_gateway_mirror_token"
|
_GATEWAY_MIRROR_TOKEN = "_gateway_mirror_token"
|
||||||
_GATEWAY_ENTRY_DATA_TOKEN_KEY = "_webui_mirror_token"
|
_GATEWAY_ENTRY_DATA_TOKEN_KEY = "_webui_mirror_token"
|
||||||
|
_GATEWAY_AGENT_IDENTITY_V1 = "_gateway_agent_identity_v1"
|
||||||
|
|
||||||
|
|
||||||
def _approval_sse_subscribe(session_id: str) -> queue.Queue:
|
def _approval_sse_subscribe(session_id: str) -> queue.Queue:
|
||||||
@@ -323,6 +324,10 @@ def retire_gateway_pending_mirror(session_key: str, approval_id: str = "", run_i
|
|||||||
gateway_queue_changed = False
|
gateway_queue_changed = False
|
||||||
if approval_id:
|
if approval_id:
|
||||||
match = _gateway_pending_mirror_locked(session_key, approval_id, run_id)
|
match = _gateway_pending_mirror_locked(session_key, approval_id, run_id)
|
||||||
|
if match is None and not normalized_run_id:
|
||||||
|
match = next((entry for entry in entries if _is_gateway_mirror_entry(entry)
|
||||||
|
and not str(entry.get("run_id") or "").strip()
|
||||||
|
and str(entry.get("approval_id") or "").strip() == approval_id), None)
|
||||||
retired = [match] if match else []
|
retired = [match] if match else []
|
||||||
else:
|
else:
|
||||||
retired = [
|
retired = [
|
||||||
|
|||||||
+3
-1
@@ -23999,6 +23999,7 @@ def _handle_approval_respond(handler, body):
|
|||||||
webui_gateway_chat_enabled,
|
webui_gateway_chat_enabled,
|
||||||
)
|
)
|
||||||
from api.config import get_config as _get_config, gateway_supports_approval_identity_v1
|
from api.config import get_config as _get_config, gateway_supports_approval_identity_v1
|
||||||
|
from api.route_approvals import _GATEWAY_AGENT_IDENTITY_V1
|
||||||
s = get_session(sid)
|
s = get_session(sid)
|
||||||
_candidate_run_id = None
|
_candidate_run_id = None
|
||||||
if s is not None:
|
if s is not None:
|
||||||
@@ -24105,7 +24106,7 @@ def _handle_approval_respond(handler, body):
|
|||||||
_base = _gateway_base_url(_cfg)
|
_base = _gateway_base_url(_cfg)
|
||||||
_key = _gateway_api_key()
|
_key = _gateway_api_key()
|
||||||
try:
|
try:
|
||||||
identity_v1 = gateway_supports_approval_identity_v1(_base, _key)
|
identity_v1 = bool(matched_mirror.get(_GATEWAY_AGENT_IDENTITY_V1)) and gateway_supports_approval_identity_v1(_base, _key)
|
||||||
HttpRunnerClient(base_url=_base, api_key=_key).respond_approval(
|
HttpRunnerClient(base_url=_base, api_key=_key).respond_approval(
|
||||||
_run_id, matched_mirror["approval_id"] if identity_v1 else "", choice
|
_run_id, matched_mirror["approval_id"] if identity_v1 else "", choice
|
||||||
)
|
)
|
||||||
@@ -24127,6 +24128,7 @@ def _handle_approval_respond(handler, body):
|
|||||||
if webui_gateway_chat_enabled(_get_config()) and _gateway_pending_approval_without_run_id(
|
if webui_gateway_chat_enabled(_get_config()) and _gateway_pending_approval_without_run_id(
|
||||||
sid, approval_id
|
sid, approval_id
|
||||||
):
|
):
|
||||||
|
resolve_gateway_pending_local(sid, approval_id, choice)
|
||||||
retire_gateway_pending_mirror(sid, approval_id=approval_id)
|
retire_gateway_pending_mirror(sid, approval_id=approval_id)
|
||||||
return j(handler, {"ok": True, "choice": choice, "local_retired": True})
|
return j(handler, {"ok": True, "choice": choice, "local_retired": True})
|
||||||
except Exception:
|
except Exception:
|
||||||
|
|||||||
@@ -2135,6 +2135,29 @@ def test_gateway_approval_response_relay():
|
|||||||
approvals._pending.pop("sess-relay", None)
|
approvals._pending.pop("sess-relay", None)
|
||||||
|
|
||||||
|
|
||||||
|
def test_synthetic_gateway_identity_stays_fifo_after_capability_upgrade():
|
||||||
|
"""A local gwrun ID never becomes remote authority after a capability refresh."""
|
||||||
|
from api import routes
|
||||||
|
from api.gateway_chat import _STREAM_RUN_IDS
|
||||||
|
import api.route_approvals as approvals
|
||||||
|
|
||||||
|
_STREAM_RUN_IDS["sid-upgrade"] = "run-upgrade"
|
||||||
|
approvals.submit_gateway_pending_mirror("sess-upgrade", {
|
||||||
|
"run_id": "run-upgrade", "approval_id": "gwrun:run-upgrade:local", "command": "echo x"
|
||||||
|
})
|
||||||
|
handler = MagicMock()
|
||||||
|
handler.wfile = io.BytesIO()
|
||||||
|
with patch("api.routes.get_session", return_value=SimpleNamespace(active_stream_id="sid-upgrade")), \
|
||||||
|
patch("api.config.gateway_supports_approval_identity_v1", return_value=True), \
|
||||||
|
patch("api.runner_client.HttpRunnerClient.respond_approval") as respond:
|
||||||
|
routes._handle_approval_respond(handler, {
|
||||||
|
"session_id": "sess-upgrade", "choice": "once", "approval_id": "gwrun:run-upgrade:local",
|
||||||
|
})
|
||||||
|
respond.assert_called_once_with("run-upgrade", "", "once")
|
||||||
|
_STREAM_RUN_IDS.pop("sid-upgrade", None)
|
||||||
|
approvals._pending.pop("sess-upgrade", None)
|
||||||
|
|
||||||
|
|
||||||
def test_gateway_approval_response_without_approval_id_409s():
|
def test_gateway_approval_response_without_approval_id_409s():
|
||||||
"""Gateway relay requires the emitted per-approval id, not bare run state."""
|
"""Gateway relay requires the emitted per-approval id, not bare run state."""
|
||||||
from api.gateway_chat import _STREAM_RUN_IDS
|
from api.gateway_chat import _STREAM_RUN_IDS
|
||||||
|
|||||||
@@ -396,6 +396,11 @@ def test_gateway_mirrored_approval_without_run_retires_locally():
|
|||||||
resp = handler.json()
|
resp = handler.json()
|
||||||
assert handler.status == 200, f"expected 200, got {handler.status}: {resp}"
|
assert handler.status == 200, f"expected 200, got {handler.status}: {resp}"
|
||||||
assert resp == {"ok": True, "choice": "once", "local_retired": True}
|
assert resp == {"ok": True, "choice": "once", "local_retired": True}
|
||||||
|
assert entry.event.is_set()
|
||||||
|
assert entry.result == "once"
|
||||||
|
with ta._lock:
|
||||||
|
assert sid not in ta._pending
|
||||||
|
assert sid not in ta._gateway_queues
|
||||||
finally:
|
finally:
|
||||||
_cleanup(sid)
|
_cleanup(sid)
|
||||||
|
|
||||||
|
|||||||
Reference in new issue
Block a user