diff --git a/gateway/platforms/api_server.py b/gateway/platforms/api_server.py index dc97d7f791278..02a1bd0081b7e 100644 --- a/gateway/platforms/api_server.py +++ b/gateway/platforms/api_server.py @@ -14,7 +14,7 @@ import itertools import json from contextlib import contextmanager, nullcontext, suppress -from contextvars import ContextVar +from contextvars import ContextVar, copy_context from functools import wraps import logging import os @@ -580,7 +580,7 @@ def _epoch_still_current(_task_id=process_task_id, _epoch=epoch): is_still_current = _epoch_still_current from gateway.run import _reap_gateway_turn_processes threading.Thread( - target=_reap_gateway_turn_processes, args=(process_task_id, process_baseline), + target=copy_context().run, args=(_reap_gateway_turn_processes, process_task_id, process_baseline), kwargs={"source": source, "is_still_current": is_still_current}, name=f"api-turn-reaper-{process_task_id[:12]}", daemon=True).start() diff --git a/gateway/run_agent_cache.py b/gateway/run_agent_cache.py index 3fe4809e43f2a..1197ab4a330a2 100644 --- a/gateway/run_agent_cache.py +++ b/gateway/run_agent_cache.py @@ -428,6 +428,7 @@ def _interrupt_running_turn(self, session_key: str, *, interrupt_reason: str, in """Sync core shared by /stop, /new and eviction: request a hard interrupt on the in-flight agent, invalidate its run generation, and reap the tool processes that turn spawned. Returns the post-bump generation.""" + from contextvars import copy_context from gateway.run import _AGENT_PENDING_SENTINEL, _reap_gateway_turn_processes, request_hard_interrupt state = self._peek_session_state(session_key) running_agent = state.turn.agent if state else None @@ -447,8 +448,8 @@ def _interrupt_running_turn(self, session_key: str, *, interrupt_reason: str, in _generation_at_interrupt = self._invalidate_session_run_generation(session_key, reason=invalidation_reason) if _process_task_id and _process_baseline is not None: threading.Thread( - target=_reap_gateway_turn_processes, - args=(_process_task_id, _process_baseline), + target=copy_context().run, + args=(_reap_gateway_turn_processes, _process_task_id, _process_baseline), kwargs={ "source": "gateway_turn_interrupt", "is_still_current": lambda: self._is_session_run_current(session_key, _generation_at_interrupt), diff --git a/tests/gateway/test_reaper_profile_scope.py b/tests/gateway/test_reaper_profile_scope.py new file mode 100644 index 0000000000000..49f984470a240 --- /dev/null +++ b/tests/gateway/test_reaper_profile_scope.py @@ -0,0 +1,69 @@ +"""Regression coverage for profile scope on abandoned-turn reaper threads.""" + +import threading +from types import SimpleNamespace + +from gateway.run_agent_cache import GatewayAgentCacheMixin +from hermes_constants import ( + get_hermes_home, + reset_hermes_home_override, + set_hermes_home_override, +) +from tools.process_registry import process_registry + + +class _RunningAgent: + _gateway_turn_process_task_id = "served-session" + _gateway_turn_process_baseline = frozenset({"proc_existing"}) + + def interrupt(self, *_args, **_kwargs): + return None + + +class _Runner: + def __init__(self): + self._state = SimpleNamespace(turn=SimpleNamespace(agent=_RunningAgent())) + + def _peek_session_state(self, _session_key): + return self._state + + def _invalidate_session_run_generation(self, _session_key, *, reason): + assert reason == "test" + return 7 + + def _is_session_run_current(self, _session_key, generation): + return generation == 7 + + +def test_interrupt_reaper_keeps_served_profile_home(tmp_path, monkeypatch): + """A /stop-style reaper must write/check process state in the session's profile, not launch home.""" + launch_home = tmp_path / "launch" + served_home = tmp_path / "served" + launch_home.mkdir() + served_home.mkdir() + monkeypatch.setenv("HERMES_HOME", str(launch_home)) + + seen_homes = [] + reaped = threading.Event() + + def _record_reap(_task_id, _baseline, *, source): + assert source == "gateway_turn_interrupt" + seen_homes.append(get_hermes_home()) + reaped.set() + return 0 + + monkeypatch.setattr(process_registry, "kill_started_since", _record_reap) + + token = set_hermes_home_override(served_home) + try: + GatewayAgentCacheMixin._interrupt_running_turn( + _Runner(), + "served-session", + interrupt_reason="test interrupt", + invalidation_reason="test", + ) + assert reaped.wait(timeout=1.0), "reaper thread did not run" + finally: + reset_hermes_home_override(token) + + assert seen_homes == [served_home]