fix(gateway): keep abandoned-turn reapers in the session profile

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()
 

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),

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]