Fix gateway approval settlement lifecycle
Settle removed approval producers before fallible projection work, and fail-close every producer owned by a terminal gateway run across completion, failure, cancellation, and teardown while preserving unrelated runs.
This commit is contained in:
1 parent
5dd9afd73f
commit
fdd72e815a
5 files changed
+437
-56
No files matched your search
+25
-9
@@ -741,8 +741,12 @@ def _run_gateway_runs_api_streaming(
|
||||
sse_event = "message"
|
||||
continue
|
||||
if payload_event == "run.completed":
|
||||
from api.route_approvals import retire_gateway_pending_mirror
|
||||
retire_gateway_pending_mirror(session_id, run_id=run_id)
|
||||
from api.route_approvals import settle_gateway_pending_run
|
||||
settle_gateway_pending_run(
|
||||
session_id,
|
||||
run_id,
|
||||
reason="Gateway run completed before approval resolution",
|
||||
)
|
||||
if payload.get("error"):
|
||||
raise RuntimeError(str(payload["error"]))
|
||||
output = str(payload.get("output") or "")
|
||||
@@ -754,12 +758,20 @@ def _run_gateway_runs_api_streaming(
|
||||
sse_event = "message"
|
||||
continue
|
||||
if payload_event == "run.failed":
|
||||
from api.route_approvals import retire_gateway_pending_mirror
|
||||
retire_gateway_pending_mirror(session_id, run_id=run_id)
|
||||
from api.route_approvals import settle_gateway_pending_run
|
||||
settle_gateway_pending_run(
|
||||
session_id,
|
||||
run_id,
|
||||
reason="Gateway run failed before approval resolution",
|
||||
)
|
||||
raise RuntimeError(str(payload.get("error") or "Gateway run failed"))
|
||||
if payload_event == "run.cancelled":
|
||||
from api.route_approvals import retire_gateway_pending_mirror
|
||||
retire_gateway_pending_mirror(session_id, run_id=run_id)
|
||||
from api.route_approvals import settle_gateway_pending_run
|
||||
settle_gateway_pending_run(
|
||||
session_id,
|
||||
run_id,
|
||||
reason="Gateway run was cancelled before approval resolution",
|
||||
)
|
||||
put_gateway_event("cancel", {"message": "Cancelled by gateway"})
|
||||
return None, usage
|
||||
reasoning_delta = _gateway_sse_reasoning_delta(payload)
|
||||
@@ -1467,10 +1479,14 @@ def _run_gateway_chat_streaming(
|
||||
mapped_run_id = str(_STREAM_RUN_IDS.get(stream_id) or "").strip()
|
||||
if mapped_run_id:
|
||||
try:
|
||||
from api.route_approvals import retire_gateway_pending_mirror
|
||||
retire_gateway_pending_mirror(session_id, run_id=mapped_run_id)
|
||||
from api.route_approvals import settle_gateway_pending_run
|
||||
settle_gateway_pending_run(
|
||||
session_id,
|
||||
mapped_run_id,
|
||||
reason="Gateway run ended during teardown before approval resolution",
|
||||
)
|
||||
except Exception:
|
||||
logger.debug("Failed to retire gateway pending mirrors during teardown", exc_info=True)
|
||||
logger.debug("Failed to settle gateway pending approvals during teardown", exc_info=True)
|
||||
if s is not None:
|
||||
try:
|
||||
with _get_session_agent_lock(session_id):
|
||||
|
||||
+76
-35
@@ -10,13 +10,16 @@ Gateway approvals carry two pieces of state with one lifecycle:
|
||||
* the **mirror** entry in ``_pending`` — this module's projection of that
|
||||
producer for the UI.
|
||||
|
||||
A producer is consumed only by an explicit resolution operation, which assigns
|
||||
its result and signals its waiter before reconciliation can observe it again.
|
||||
A producer is consumed only by an explicit settlement operation, which assigns
|
||||
its result and signals its waiter before fallible projection work begins.
|
||||
``retire_gateway_pending_mirror`` cleans up the UI projection; it must not
|
||||
silently delete a producer, because that would strand the agent thread parked
|
||||
on the producer's event. For a provider-qualified response,
|
||||
``resolve_gateway_pending_run`` matches the exact ``(run_id, approval_id)``
|
||||
pair, settles that producer, and preserves same-run siblings and other runs.
|
||||
When a run becomes terminal, ``settle_gateway_pending_run`` fail-closes every
|
||||
producer owned by that run with ``deny`` before retiring its mirrors; producers
|
||||
owned by another run remain live.
|
||||
"""
|
||||
import queue
|
||||
import threading
|
||||
@@ -520,19 +523,26 @@ def release_gateway_approval_relay_owner(session_key: str, run_id: str, approval
|
||||
_gateway_relay_owners.pop(key, None)
|
||||
|
||||
|
||||
def _settle_gateway_entry(entry, choice: str, reason: str | None = None) -> None:
|
||||
"""Complete a removed producer before any fallible projection work."""
|
||||
entry.result = choice
|
||||
if reason:
|
||||
entry.reason = reason
|
||||
entry.event.set()
|
||||
|
||||
|
||||
def retire_gateway_pending_mirror(
|
||||
session_key: str,
|
||||
approval_id: str = "",
|
||||
run_id: str = "",
|
||||
mirror_token: str = "",
|
||||
) -> bool:
|
||||
"""Retire one approval, or every mirror for a terminal run."""
|
||||
"""Retire one approval mirror, or every mirror for a terminal run."""
|
||||
with _lock:
|
||||
reconcile_gateway_pending_mirror_locked(session_key)
|
||||
queue = _pending.get(session_key)
|
||||
entries = queue if isinstance(queue, list) else [queue] if queue else []
|
||||
normalized_run_id = str(run_id or "").strip()
|
||||
gateway_queue_changed = False
|
||||
if approval_id:
|
||||
match = _gateway_pending_mirror_locked(
|
||||
session_key,
|
||||
@@ -555,20 +565,7 @@ def retire_gateway_pending_mirror(
|
||||
if _is_gateway_mirror_entry(entry)
|
||||
and not str(entry.get("run_id") or "").strip()
|
||||
]
|
||||
if normalized_run_id:
|
||||
gateway_queue = _gateway_queues.get(session_key) or []
|
||||
retained_gateway_queue = []
|
||||
for entry in gateway_queue:
|
||||
data = getattr(entry, "data", None) or {}
|
||||
if str(data.get("run_id") or "").strip() == normalized_run_id:
|
||||
continue
|
||||
retained_gateway_queue.append(entry)
|
||||
gateway_queue_changed = len(retained_gateway_queue) != len(gateway_queue)
|
||||
if retained_gateway_queue:
|
||||
_gateway_queues[session_key] = retained_gateway_queue
|
||||
else:
|
||||
_gateway_queues.pop(session_key, None)
|
||||
if not retired and not gateway_queue_changed:
|
||||
if not retired:
|
||||
head, total, changed = reconcile_gateway_pending_mirror_locked(session_key)
|
||||
_approval_sse_notify_locked(session_key, head, total)
|
||||
if changed:
|
||||
@@ -586,6 +583,60 @@ def retire_gateway_pending_mirror(
|
||||
return True
|
||||
|
||||
|
||||
def settle_gateway_pending_run(
|
||||
session_key: str,
|
||||
run_id: str,
|
||||
*,
|
||||
reason: str,
|
||||
) -> tuple[int, dict | None, int]:
|
||||
"""Fail-close every producer and mirror owned by a terminal gateway run."""
|
||||
run_id = str(run_id or "").strip()
|
||||
if not run_id:
|
||||
return 0, None, 0
|
||||
|
||||
targets = []
|
||||
with _lock:
|
||||
gateway_queue = _gateway_queues.get(session_key) or []
|
||||
retained_gateway_queue = []
|
||||
for entry in gateway_queue:
|
||||
data = getattr(entry, "data", None) or {}
|
||||
if str(data.get("run_id") or "").strip() == run_id:
|
||||
targets.append(entry)
|
||||
else:
|
||||
retained_gateway_queue.append(entry)
|
||||
if retained_gateway_queue:
|
||||
_gateway_queues[session_key] = retained_gateway_queue
|
||||
else:
|
||||
_gateway_queues.pop(session_key, None)
|
||||
|
||||
# Removal and waiter settlement are one semantic operation. In
|
||||
# particular, reconciliation and SSE notification below may raise.
|
||||
for entry in targets:
|
||||
_settle_gateway_entry(entry, "deny", reason)
|
||||
|
||||
queue = _pending.get(session_key)
|
||||
entries = queue if isinstance(queue, list) else [queue] if queue else []
|
||||
retained_pending = [
|
||||
entry
|
||||
for entry in entries
|
||||
if not (
|
||||
_is_gateway_mirror_entry(entry)
|
||||
and str(entry.get("run_id") or "").strip() == run_id
|
||||
)
|
||||
]
|
||||
mirrors_retired = len(retained_pending) != len(entries)
|
||||
if retained_pending:
|
||||
_pending[session_key] = retained_pending
|
||||
else:
|
||||
_pending.pop(session_key, None)
|
||||
|
||||
head, total, changed = reconcile_gateway_pending_mirror_locked(session_key)
|
||||
_approval_sse_notify_locked(session_key, head, total)
|
||||
if targets or mirrors_retired or changed:
|
||||
publish_session_list_changed("attention_resolved")
|
||||
return len(targets), head, total
|
||||
|
||||
|
||||
def _gateway_mirrored_pending_run_id(session_key: str, approval_id: str) -> str | None:
|
||||
"""Compatibility wrapper for exact run-backed lookup."""
|
||||
approval_id = str(approval_id or "").strip()
|
||||
@@ -817,14 +868,12 @@ def resolve_gateway_pending_local(
|
||||
_gateway_queues[session_key] = gateway_queue
|
||||
else:
|
||||
_gateway_queues.pop(session_key, None)
|
||||
if target is not None:
|
||||
_settle_gateway_entry(target, choice, reason)
|
||||
head, total, _changed = reconcile_gateway_pending_mirror_locked(session_key)
|
||||
_approval_sse_notify_locked(session_key, head, total)
|
||||
if target is None:
|
||||
return 0, head, total
|
||||
target.result = choice
|
||||
if reason:
|
||||
target.reason = reason
|
||||
target.event.set()
|
||||
publish_session_list_changed("attention_resolved")
|
||||
return 1, head, total
|
||||
|
||||
@@ -857,14 +906,12 @@ def resolve_gateway_pending_run(
|
||||
_gateway_queues[session_key] = retained_gateway_queue
|
||||
else:
|
||||
_gateway_queues.pop(session_key, None)
|
||||
if target is not None:
|
||||
_settle_gateway_entry(target, choice, reason)
|
||||
head, total, _changed = reconcile_gateway_pending_mirror_locked(session_key)
|
||||
_approval_sse_notify_locked(session_key, head, total)
|
||||
if target is None:
|
||||
return 0, head, total
|
||||
target.result = choice
|
||||
if reason:
|
||||
target.reason = reason
|
||||
target.event.set()
|
||||
publish_session_list_changed("attention_resolved")
|
||||
return 1, head, total
|
||||
|
||||
@@ -909,12 +956,9 @@ def resolve_gateway_pending_local_no_run_mirror(
|
||||
_pending[session_key] = entries
|
||||
else:
|
||||
_pending.pop(session_key, None)
|
||||
_settle_gateway_entry(target, choice, reason)
|
||||
head, total, _changed = reconcile_gateway_pending_mirror_locked(session_key)
|
||||
_approval_sse_notify_locked(session_key, head, total)
|
||||
target.result = choice
|
||||
if reason:
|
||||
target.reason = reason
|
||||
target.event.set()
|
||||
publish_session_list_changed("attention_resolved")
|
||||
return True, 1, head, total
|
||||
|
||||
@@ -957,14 +1001,11 @@ def resolve_gateway_pending_local_all(
|
||||
else:
|
||||
_pending.pop(session_key, None)
|
||||
|
||||
for entry in targets:
|
||||
_settle_gateway_entry(entry, choice, reason)
|
||||
head, total, _changed = reconcile_gateway_pending_mirror_locked(session_key)
|
||||
_approval_sse_notify_locked(session_key, head, total)
|
||||
|
||||
for entry in targets:
|
||||
entry.result = choice
|
||||
if reason:
|
||||
entry.reason = reason
|
||||
entry.event.set()
|
||||
if targets or removed_pending:
|
||||
publish_session_list_changed("attention_resolved")
|
||||
return len(targets), head, total
|
||||
|
||||
+6
-1
@@ -10771,6 +10771,7 @@ from api.route_approvals import ( # noqa: F401 — re-exports for backward comp
|
||||
gateway_pending_mirrors,
|
||||
release_gateway_approval_relay_owner,
|
||||
retire_gateway_pending_mirror,
|
||||
settle_gateway_pending_run,
|
||||
reconcile_gateway_pending_mirror_locked,
|
||||
resolve_gateway_pending_local,
|
||||
resolve_gateway_pending_run,
|
||||
@@ -14485,7 +14486,11 @@ def handle_get(handler, parsed) -> bool:
|
||||
if stop_gateway_run(run_id):
|
||||
owner_sid = stream_owner_session_id(stream_id)
|
||||
if owner_sid:
|
||||
retire_gateway_pending_mirror(owner_sid, run_id=run_id)
|
||||
settle_gateway_pending_run(
|
||||
owner_sid,
|
||||
run_id,
|
||||
reason="Gateway run was cancelled before approval resolution",
|
||||
)
|
||||
else:
|
||||
gateway_stop_blocked = True
|
||||
except Exception:
|
||||
|
||||
@@ -10,6 +10,8 @@ import threading
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from api.gateway_chat import _gateway_runs_approval_event
|
||||
|
||||
REPO_ROOT = Path(__file__).parent.parent
|
||||
@@ -1110,3 +1112,109 @@ def test_deny_settles_run_producer_so_reconciliation_cannot_resurrect_mirror():
|
||||
with ra._lock:
|
||||
ra._pending.pop(sid, None)
|
||||
ra._gateway_queues.pop(sid, None)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"failure_point",
|
||||
["reconcile_gateway_pending_mirror_locked", "_approval_sse_notify_locked"],
|
||||
)
|
||||
def test_exact_run_resolution_settles_before_projection_failure(failure_point):
|
||||
"""Projection failures must not strand an already-consumed producer."""
|
||||
from types import SimpleNamespace
|
||||
from api import route_approvals as ra
|
||||
|
||||
sid = f"sess-exact-projection-failure-{failure_point}"
|
||||
run_id = "run-exact-projection-failure"
|
||||
target = SimpleNamespace(
|
||||
data={"run_id": run_id, "approval_id": "approval-target"},
|
||||
event=threading.Event(),
|
||||
result=None,
|
||||
reason=None,
|
||||
)
|
||||
sibling = SimpleNamespace(
|
||||
data={"run_id": run_id, "approval_id": "approval-sibling"},
|
||||
event=threading.Event(),
|
||||
result=None,
|
||||
reason=None,
|
||||
)
|
||||
try:
|
||||
with ra._lock:
|
||||
ra._pending.pop(sid, None)
|
||||
ra._gateway_queues[sid] = [target, sibling]
|
||||
|
||||
with patch.object(ra, failure_point, side_effect=RuntimeError("projection failed")):
|
||||
with pytest.raises(RuntimeError, match="projection failed"):
|
||||
ra.resolve_gateway_pending_run(
|
||||
sid,
|
||||
approval_id="approval-target",
|
||||
run_id=run_id,
|
||||
choice="deny",
|
||||
reason="user denied",
|
||||
)
|
||||
|
||||
assert target.result == "deny"
|
||||
assert target.reason == "user denied"
|
||||
assert target.event.is_set(), "the consumed producer must be woken before projection work"
|
||||
assert sibling.result is None
|
||||
assert sibling.reason is None
|
||||
assert not sibling.event.is_set()
|
||||
with ra._lock:
|
||||
assert ra._gateway_queues[sid] == [sibling]
|
||||
finally:
|
||||
with ra._lock:
|
||||
ra._pending.pop(sid, None)
|
||||
ra._gateway_queues.pop(sid, None)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"terminal_reason",
|
||||
[
|
||||
"Gateway run completed before approval resolution",
|
||||
"Gateway run failed before approval resolution",
|
||||
"Gateway run was cancelled before approval resolution",
|
||||
"Gateway run ended during teardown before approval resolution",
|
||||
],
|
||||
)
|
||||
def test_terminal_run_settlement_denies_all_run_producers_and_preserves_other_runs(
|
||||
terminal_reason,
|
||||
):
|
||||
"""Every terminal exit fails closed without consuming another run's producer."""
|
||||
from types import SimpleNamespace
|
||||
from api import route_approvals as ra
|
||||
|
||||
sid = f"sess-terminal-run-{terminal_reason.split()[2]}"
|
||||
run_id = "run-terminal"
|
||||
|
||||
def producer(entry_run_id, approval_id):
|
||||
return SimpleNamespace(
|
||||
data={"run_id": entry_run_id, "approval_id": approval_id},
|
||||
event=threading.Event(),
|
||||
result=None,
|
||||
reason=None,
|
||||
)
|
||||
|
||||
targets = [producer(run_id, "approval-a"), producer(run_id, "approval-b")]
|
||||
survivor = producer("run-other", "approval-other")
|
||||
try:
|
||||
with ra._lock:
|
||||
ra._pending.pop(sid, None)
|
||||
ra._gateway_queues[sid] = [targets[0], survivor, targets[1]]
|
||||
|
||||
settled, _head, _total = ra.settle_gateway_pending_run(
|
||||
sid, run_id, reason=terminal_reason
|
||||
)
|
||||
|
||||
assert settled == 2
|
||||
for target in targets:
|
||||
assert target.result == "deny"
|
||||
assert target.reason == terminal_reason
|
||||
assert target.event.is_set()
|
||||
assert survivor.result is None
|
||||
assert survivor.reason is None
|
||||
assert not survivor.event.is_set()
|
||||
with ra._lock:
|
||||
assert ra._gateway_queues[sid] == [survivor]
|
||||
finally:
|
||||
with ra._lock:
|
||||
ra._pending.pop(sid, None)
|
||||
ra._gateway_queues.pop(sid, None)
|
||||
@@ -333,6 +333,178 @@ def test_gateway_runs_api_submission():
|
||||
assert captured["body_extras"]["reasoning_effort"] == "high"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("terminal_event", "terminal_reason"),
|
||||
[
|
||||
("run.completed", "Gateway run completed before approval resolution"),
|
||||
("run.failed", "Gateway run failed before approval resolution"),
|
||||
("run.cancelled", "Gateway run was cancelled before approval resolution"),
|
||||
],
|
||||
)
|
||||
def test_gateway_terminal_event_fail_closes_parked_run_producer(
|
||||
terminal_event,
|
||||
terminal_reason,
|
||||
):
|
||||
"""Each terminal gateway event denies and wakes the run's parked producers."""
|
||||
from api import gateway_chat
|
||||
from api import route_approvals as approvals
|
||||
|
||||
sid = f"sess-terminal-event-{terminal_event}"
|
||||
stream_id = f"stream-terminal-event-{terminal_event}"
|
||||
run_id = f"run-terminal-event-{terminal_event}"
|
||||
target = SimpleNamespace(
|
||||
data={"run_id": run_id, "approval_id": "approval-target"},
|
||||
event=threading.Event(),
|
||||
result=None,
|
||||
reason=None,
|
||||
)
|
||||
survivor = SimpleNamespace(
|
||||
data={"run_id": "run-other", "approval_id": "approval-other"},
|
||||
event=threading.Event(),
|
||||
result=None,
|
||||
reason=None,
|
||||
)
|
||||
|
||||
class JsonResponse:
|
||||
def __enter__(self):
|
||||
return self
|
||||
|
||||
def __exit__(self, *_args):
|
||||
return None
|
||||
|
||||
def read(self, _limit=None):
|
||||
return json.dumps({"run_id": run_id}).encode()
|
||||
|
||||
class SseResponse:
|
||||
def __enter__(self):
|
||||
return self
|
||||
|
||||
def __exit__(self, *_args):
|
||||
return None
|
||||
|
||||
def __iter__(self):
|
||||
payload = {"event": terminal_event}
|
||||
if terminal_event == "run.failed":
|
||||
payload["error"] = "terminal failure"
|
||||
return iter([f"data: {json.dumps(payload)}".encode(), b""])
|
||||
|
||||
def fake_urlopen(req, *, timeout=None):
|
||||
del timeout
|
||||
return JsonResponse() if req.full_url.endswith("/v1/runs") else SseResponse()
|
||||
|
||||
try:
|
||||
with approvals._lock:
|
||||
approvals._pending.pop(sid, None)
|
||||
approvals._gateway_queues[sid] = [target, survivor]
|
||||
with patch("urllib.request.urlopen", side_effect=fake_urlopen):
|
||||
if terminal_event == "run.failed":
|
||||
with pytest.raises(RuntimeError, match="terminal failure"):
|
||||
gateway_chat._run_gateway_runs_api_streaming(
|
||||
sid, "hi", "test", "/tmp", stream_id, "http://gw:8642", "", [], {},
|
||||
put_gateway_event=lambda *_args: None,
|
||||
cancel_event=threading.Event(),
|
||||
session=SimpleNamespace(context_messages=[]),
|
||||
)
|
||||
else:
|
||||
gateway_chat._run_gateway_runs_api_streaming(
|
||||
sid, "hi", "test", "/tmp", stream_id, "http://gw:8642", "", [], {},
|
||||
put_gateway_event=lambda *_args: None,
|
||||
cancel_event=threading.Event(),
|
||||
session=SimpleNamespace(context_messages=[]),
|
||||
)
|
||||
|
||||
assert target.result == "deny"
|
||||
assert target.reason == terminal_reason
|
||||
assert target.event.is_set()
|
||||
assert survivor.result is None
|
||||
assert not survivor.event.is_set()
|
||||
with approvals._lock:
|
||||
assert approvals._gateway_queues[sid] == [survivor]
|
||||
finally:
|
||||
gateway_chat._STREAM_RUN_IDS.pop(stream_id, None)
|
||||
with approvals._lock:
|
||||
approvals._pending.pop(sid, None)
|
||||
approvals._gateway_queues.pop(sid, None)
|
||||
|
||||
|
||||
def test_gateway_stream_teardown_fail_closes_parked_run_producer():
|
||||
"""A disconnect before a terminal event still settles the mapped run."""
|
||||
from api import gateway_chat
|
||||
from api import route_approvals as approvals
|
||||
from api.config import STREAMS, STREAMS_LOCK
|
||||
|
||||
sid = "sess-terminal-teardown"
|
||||
stream_id = "stream-terminal-teardown"
|
||||
run_id = "run-terminal-teardown"
|
||||
target = SimpleNamespace(
|
||||
data={"run_id": run_id, "approval_id": "approval-target"},
|
||||
event=threading.Event(),
|
||||
result=None,
|
||||
reason=None,
|
||||
)
|
||||
survivor = SimpleNamespace(
|
||||
data={"run_id": "run-other", "approval_id": "approval-other"},
|
||||
event=threading.Event(),
|
||||
result=None,
|
||||
reason=None,
|
||||
)
|
||||
events = []
|
||||
queue = MagicMock()
|
||||
queue.put_nowait = lambda item: events.append(item)
|
||||
with STREAMS_LOCK:
|
||||
STREAMS[stream_id] = queue
|
||||
|
||||
session = MagicMock()
|
||||
session.active_stream_id = stream_id
|
||||
session.workspace = "/tmp"
|
||||
session.model = "test"
|
||||
session.model_provider = None
|
||||
session.profile = None
|
||||
session.context_messages = []
|
||||
session.messages = []
|
||||
session.pending_user_message = None
|
||||
session.pending_attachments = None
|
||||
session.pending_started_at = None
|
||||
|
||||
def disconnect_after_run_id(*_args, **_kwargs):
|
||||
gateway_chat._publish_gateway_run_id(stream_id, run_id)
|
||||
raise RuntimeError("gateway disconnected")
|
||||
|
||||
try:
|
||||
with approvals._lock:
|
||||
approvals._pending.pop(sid, None)
|
||||
approvals._gateway_queues[sid] = [target, survivor]
|
||||
with patch.dict(
|
||||
"os.environ",
|
||||
{"HERMES_WEBUI_CHAT_BACKEND": "gateway", "HERMES_WEBUI_GATEWAY_USE_RUNS_API": "1"},
|
||||
), patch("api.gateway_chat.gateway_supports_approval", return_value=True), patch(
|
||||
"api.gateway_chat._run_gateway_runs_api_streaming",
|
||||
side_effect=disconnect_after_run_id,
|
||||
), patch("api.gateway_chat.get_session", return_value=session):
|
||||
gateway_chat._run_gateway_chat_streaming(
|
||||
session_id=sid,
|
||||
msg_text="hi",
|
||||
model="test-model",
|
||||
workspace="/tmp",
|
||||
stream_id=stream_id,
|
||||
)
|
||||
|
||||
assert target.result == "deny"
|
||||
assert target.reason == "Gateway run ended during teardown before approval resolution"
|
||||
assert target.event.is_set()
|
||||
assert survivor.result is None
|
||||
assert not survivor.event.is_set()
|
||||
with approvals._lock:
|
||||
assert approvals._gateway_queues[sid] == [survivor]
|
||||
finally:
|
||||
gateway_chat._STREAM_RUN_IDS.pop(stream_id, None)
|
||||
with STREAMS_LOCK:
|
||||
STREAMS.pop(stream_id, None)
|
||||
with approvals._lock:
|
||||
approvals._pending.pop(sid, None)
|
||||
approvals._gateway_queues.pop(sid, None)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 3. Approval event translation
|
||||
# ---------------------------------------------------------------------------
|
||||
@@ -1268,22 +1440,32 @@ def test_terminal_run_retirement_removes_all_same_run_mirrors():
|
||||
approvals._pending.pop(sid, None)
|
||||
|
||||
|
||||
def test_terminal_run_retirement_clears_same_run_gateway_queue_state():
|
||||
def test_terminal_run_settlement_clears_same_run_gateway_queue_state():
|
||||
import api.route_approvals as approvals
|
||||
|
||||
sid = "sess-terminal-run-queue"
|
||||
approvals._pending.pop(sid, None)
|
||||
approvals._gateway_queues.pop(sid, None)
|
||||
try:
|
||||
approvals._gateway_queues[sid] = [
|
||||
SimpleNamespace(data={"run_id": "run-a", "approval_id": "appr-a", "command": "a"}),
|
||||
SimpleNamespace(data={"command": "local-head"}),
|
||||
]
|
||||
target = SimpleNamespace(
|
||||
data={"run_id": "run-a", "approval_id": "appr-a", "command": "a"},
|
||||
event=threading.Event(),
|
||||
result=None,
|
||||
reason=None,
|
||||
)
|
||||
local_head = SimpleNamespace(data={"command": "local-head"})
|
||||
approvals._gateway_queues[sid] = [target, local_head]
|
||||
approvals.submit_gateway_pending_mirror(
|
||||
sid,
|
||||
{"run_id": "run-a", "approval_id": "appr-a", "command": "a"},
|
||||
)
|
||||
assert approvals.retire_gateway_pending_mirror(sid, run_id="run-a")
|
||||
settled, _head, _total = approvals.settle_gateway_pending_run(
|
||||
sid, "run-a", reason="terminal"
|
||||
)
|
||||
assert settled == 1
|
||||
assert target.result == "deny"
|
||||
assert target.reason == "terminal"
|
||||
assert target.event.is_set()
|
||||
assert approvals.gateway_pending_mirror(sid, run_id="run-a") is None
|
||||
assert not any(
|
||||
str((getattr(entry, "data", None) or {}).get("run_id") or "").strip() == "run-a"
|
||||
@@ -1294,18 +1476,30 @@ def test_terminal_run_retirement_clears_same_run_gateway_queue_state():
|
||||
approvals._gateway_queues.pop(sid, None)
|
||||
|
||||
|
||||
def test_terminal_run_retirement_clears_non_head_same_run_gateway_queue_state():
|
||||
def test_terminal_run_settlement_clears_non_head_same_run_gateway_queue_state():
|
||||
import api.route_approvals as approvals
|
||||
|
||||
sid = "sess-terminal-run-non-head-queue"
|
||||
approvals._pending.pop(sid, None)
|
||||
approvals._gateway_queues.pop(sid, None)
|
||||
try:
|
||||
target = SimpleNamespace(
|
||||
data={"run_id": "run-a", "approval_id": "appr-a", "command": "a"},
|
||||
event=threading.Event(),
|
||||
result=None,
|
||||
reason=None,
|
||||
)
|
||||
approvals._gateway_queues[sid] = [
|
||||
SimpleNamespace(data={"command": "local-head"}),
|
||||
SimpleNamespace(data={"run_id": "run-a", "approval_id": "appr-a", "command": "a"}),
|
||||
target,
|
||||
]
|
||||
assert approvals.retire_gateway_pending_mirror(sid, run_id="run-a")
|
||||
settled, _head, _total = approvals.settle_gateway_pending_run(
|
||||
sid, "run-a", reason="terminal"
|
||||
)
|
||||
assert settled == 1
|
||||
assert target.result == "deny"
|
||||
assert target.reason == "terminal"
|
||||
assert target.event.is_set()
|
||||
assert approvals.gateway_pending_mirror(sid, run_id="run-a") is None
|
||||
assert not any(
|
||||
str((getattr(entry, "data", None) or {}).get("run_id") or "").strip() == "run-a"
|
||||
@@ -1865,7 +2059,15 @@ def test_chat_cancel_waits_for_worker_published_run_id_before_settlement(
|
||||
STREAMS[stream_id] = SimpleNamespace(put_nowait=lambda *_args, **_kwargs: None)
|
||||
register_stream_owner(stream_id, sid)
|
||||
approvals._pending.pop(sid, None)
|
||||
approvals.submit_gateway_pending_mirror(sid, {"run_id": "run-stop/1", "command": "first"})
|
||||
approval_entry = SimpleNamespace(
|
||||
data={"run_id": "run-stop/1", "approval_id": "approval-stop"},
|
||||
event=threading.Event(),
|
||||
result=None,
|
||||
reason=None,
|
||||
)
|
||||
with approvals._lock:
|
||||
approvals._gateway_queues[sid] = [approval_entry]
|
||||
approvals.submit_gateway_pending_mirror(sid, approval_entry.data)
|
||||
session = SimpleNamespace(
|
||||
profile=None,
|
||||
workspace="/tmp",
|
||||
@@ -1945,8 +2147,15 @@ def test_chat_cancel_waits_for_worker_published_run_id_before_settlement(
|
||||
assert called["stop"] == "run-stop/1"
|
||||
assert called["cancel"] is expect_cancel
|
||||
if stop_result:
|
||||
assert approval_entry.result == "deny"
|
||||
assert approval_entry.reason == (
|
||||
"Gateway run was cancelled before approval resolution"
|
||||
)
|
||||
assert approval_entry.event.is_set()
|
||||
assert approvals.gateway_pending_mirror(sid, run_id="run-stop/1") is None
|
||||
else:
|
||||
assert approval_entry.result is None
|
||||
assert not approval_entry.event.is_set()
|
||||
assert stream_id in STREAMS
|
||||
assert stream_owner_session_id(stream_id) == sid
|
||||
assert session.active_stream_id == stream_id
|
||||
@@ -1957,7 +2166,9 @@ def test_chat_cancel_waits_for_worker_published_run_id_before_settlement(
|
||||
release_worker.set()
|
||||
request_thread.join(timeout=5)
|
||||
worker_thread.join(timeout=5)
|
||||
approvals._pending.pop(sid, None)
|
||||
with approvals._lock:
|
||||
approvals._pending.pop(sid, None)
|
||||
approvals._gateway_queues.pop(sid, None)
|
||||
STREAMS.pop(stream_id, None)
|
||||
ACTIVE_RUNS.pop(stream_id, None)
|
||||
unregister_stream_owner(stream_id)
|
||||
|
||||
Reference in new issue
Block a user