fix: record gateway run_id on legacy approval event for response relay
The legacy /v1/chat/completions approval handler (added for #4549) rendered the approval card but never populated _STREAM_RUN_IDS, so /api/approval/respond could not relay the user's choice back to the gateway to resume the parked run — it fell through to the local approval path and returned {ok: false}. Record the run_id from the approval payload (no-op when absent), mirroring the runs-API path. Adds a behavioral regression test that fails on the pre-fix head. Codex ship-gate finding (CORE). Co-authored-by: rodboev <rodboev@users.noreply.github.com>
This commit is contained in:
nesquena-hermes
parent
172ca592b6
commit
5020f8de9a
2 changed files with 143 additions and 1 deletions
|
|
@ -667,6 +667,14 @@ def _run_gateway_chat_streaming(
|
|||
if _payload_event in {"hermes.approval.request", "approval.request"}:
|
||||
approval_data = _gateway_runs_approval_event(payload)
|
||||
if approval_data:
|
||||
# Record the gateway run_id so /api/approval/respond
|
||||
# can relay the choice back and resume the parked run
|
||||
# (legacy path never creates a local run; without this
|
||||
# the card renders but approve/deny returns ok:false).
|
||||
# No-op when the payload omits run_id.
|
||||
_approval_run_id = str(approval_data.get("run_id") or "").strip()
|
||||
if _approval_run_id:
|
||||
_STREAM_RUN_IDS[stream_id] = _approval_run_id
|
||||
put_gateway_event("approval", approval_data)
|
||||
try:
|
||||
from api.route_approvals import submit_gateway_pending_mirror
|
||||
|
|
|
|||
|
|
@ -81,7 +81,9 @@ def test_legacy_loop_resets_sse_event_after_approval():
|
|||
loop = _extract_legacy_sse_loop()
|
||||
approval_idx = loop.find('"hermes.approval.request"')
|
||||
assert approval_idx >= 0
|
||||
block_after = loop[approval_idx:approval_idx + 900]
|
||||
# Window sized to cover the approval handling block including the run_id
|
||||
# recording added in the #4549 follow-up (reset lands ~1360 chars in).
|
||||
block_after = loop[approval_idx:approval_idx + 1500]
|
||||
assert 'sse_event = "message"' in block_after, (
|
||||
"Must reset sse_event to 'message' after approval handling to prevent bleed"
|
||||
)
|
||||
|
|
@ -197,3 +199,135 @@ def test_legacy_sse_loop_relays_approval_event():
|
|||
assert approval_events[0][1]["command"] == "rm -rf /tmp/test"
|
||||
assert approval_events[0][1]["approval_id"] == "appr-legacy-1"
|
||||
assert approval_events[0][1]["description"] == "Delete temporary files"
|
||||
|
||||
|
||||
def test_legacy_approval_records_run_id_for_response_relay():
|
||||
"""#4549 follow-up: a legacy approval event carrying a run_id must populate
|
||||
_STREAM_RUN_IDS so /api/approval/respond can relay the choice back to the
|
||||
gateway and resume the parked run.
|
||||
|
||||
Without recording the run_id, the approval card renders but approve/deny
|
||||
falls through to the local path (no remote gateway agent to resume) and the
|
||||
response is {"ok": false}. This regression test fails on the pre-fix head
|
||||
(run_id never stored) and passes on the fixed head.
|
||||
"""
|
||||
import io
|
||||
from api.config import STREAMS, STREAMS_LOCK
|
||||
from api.gateway_chat import _STREAM_RUN_IDS, _run_gateway_chat_streaming
|
||||
|
||||
events = []
|
||||
# Capture _STREAM_RUN_IDS at the instant the approval event is emitted.
|
||||
# In production the legacy SSE connection stays open (blocked on the gateway
|
||||
# stream) while the run is parked for approval, so _STREAM_RUN_IDS is still
|
||||
# populated when the user responds. This test completes the stream
|
||||
# synchronously, after which the function's finally-block pops the mapping —
|
||||
# so we snapshot the live value mid-stream rather than after return.
|
||||
run_id_at_approval = {}
|
||||
|
||||
def _record(item):
|
||||
events.append(item)
|
||||
if isinstance(item, tuple) and item[0] == "approval":
|
||||
run_id_at_approval["value"] = _STREAM_RUN_IDS.get(stream_id)
|
||||
|
||||
q = MagicMock()
|
||||
q.put_nowait = _record
|
||||
|
||||
stream_id = "sid-legacy-runid"
|
||||
with STREAMS_LOCK:
|
||||
STREAMS[stream_id] = q
|
||||
_STREAM_RUN_IDS.pop(stream_id, None)
|
||||
|
||||
approval_payload = json.dumps({
|
||||
"command": "rm -rf /tmp/test",
|
||||
"description": "Delete temporary files",
|
||||
"pattern_key": "dangerous_command",
|
||||
"pattern_keys": ["dangerous_command"],
|
||||
"approval_id": "appr-legacy-runid",
|
||||
"run_id": "run-legacy-1",
|
||||
"choices": ["once", "session", "always", "deny"],
|
||||
})
|
||||
sse_body = (
|
||||
f"event: approval.request\ndata: {approval_payload}\n\n"
|
||||
'data: {"choices":[{"delta":{"content":"Done"}}]}\n\n'
|
||||
"data: [DONE]\n\n"
|
||||
).encode()
|
||||
|
||||
mock_session = MagicMock()
|
||||
mock_session.active_stream_id = stream_id
|
||||
mock_session.workspace = "/tmp"
|
||||
mock_session.model = "test"
|
||||
mock_session.model_provider = None
|
||||
mock_session.profile = None
|
||||
mock_session.context_messages = []
|
||||
mock_session.messages = []
|
||||
mock_session.pending_user_message = None
|
||||
mock_session.pending_attachments = None
|
||||
mock_session.pending_started_at = None
|
||||
|
||||
def fake_urlopen(req, *, timeout=None):
|
||||
resp = MagicMock()
|
||||
resp.__iter__ = lambda s: iter(sse_body.split(b"\n"))
|
||||
resp.__enter__ = lambda s: s
|
||||
resp.__exit__ = lambda s, *a: None
|
||||
return resp
|
||||
|
||||
try:
|
||||
with patch.dict("os.environ", {"HERMES_WEBUI_CHAT_BACKEND": "gateway"}):
|
||||
with patch("api.gateway_chat.gateway_supports_approval", return_value=False), \
|
||||
patch("urllib.request.urlopen", side_effect=fake_urlopen), \
|
||||
patch("api.gateway_chat.get_session", return_value=mock_session), \
|
||||
patch("api.gateway_chat._stream_writeback_is_current", return_value=True), \
|
||||
patch("api.gateway_chat.merge_session_messages_append_only", return_value=[]):
|
||||
_run_gateway_chat_streaming(
|
||||
session_id="sess-legacy-runid",
|
||||
msg_text="do something risky",
|
||||
model="test",
|
||||
workspace="/tmp",
|
||||
stream_id=stream_id,
|
||||
)
|
||||
|
||||
# The run_id from the approval payload must have been recorded at the
|
||||
# moment the approval event was emitted (before the synchronous stream
|
||||
# completed and the finally-block popped it).
|
||||
assert run_id_at_approval.get("value") == "run-legacy-1", (
|
||||
"Legacy approval event must record run_id in _STREAM_RUN_IDS so the "
|
||||
"approval response can relay to the gateway and resume the run. "
|
||||
f"Got {run_id_at_approval.get('value')!r}"
|
||||
)
|
||||
|
||||
# And /api/approval/respond must actually relay to the gateway runs API
|
||||
# when the mapping is live. Use a fresh session whose active_stream_id is
|
||||
# still set + re-seed _STREAM_RUN_IDS to model the production state
|
||||
# (connection still open, run parked) — this test's stream already ran to
|
||||
# completion, which cleared active_stream_id and popped the mapping.
|
||||
_STREAM_RUN_IDS[stream_id] = "run-legacy-1"
|
||||
relay_session = MagicMock()
|
||||
relay_session.active_stream_id = stream_id
|
||||
captured = {}
|
||||
|
||||
def fake_request_json(self, req):
|
||||
captured["url"] = req.full_url
|
||||
captured["body"] = json.loads(req.data)
|
||||
return {"ok": True}
|
||||
|
||||
handler = MagicMock()
|
||||
handler.wfile = io.BytesIO()
|
||||
body = {"session_id": "sess-legacy-runid", "choice": "once",
|
||||
"approval_id": "appr-legacy-runid"}
|
||||
|
||||
with patch("api.routes.get_session", return_value=relay_session), \
|
||||
patch("api.runner_client.HttpRunnerClient._request_json", new=fake_request_json), \
|
||||
patch("api.gateway_chat._gateway_base_url", return_value="http://gw:8642"), \
|
||||
patch("api.gateway_chat._gateway_api_key", return_value=""):
|
||||
from api.routes import _handle_approval_respond
|
||||
_handle_approval_respond(handler, body)
|
||||
|
||||
assert captured.get("url", "") == "http://gw:8642/v1/runs/run-legacy-1/approval", (
|
||||
f"approval respond must relay to the gateway run; got {captured.get('url')!r}"
|
||||
)
|
||||
assert captured["body"] == {"choice": "once", "approval_id": "appr-legacy-runid"}
|
||||
handler.send_response.assert_called_with(200)
|
||||
finally:
|
||||
with STREAMS_LOCK:
|
||||
STREAMS.pop(stream_id, None)
|
||||
_STREAM_RUN_IDS.pop(stream_id, None)
|
||||
|
|
|
|||
Loading…
Reference in a new issue