mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-04-26 01:01:40 +00:00
fix(honcho): isolate session routing for multi-user gateway (#1500)
Salvaged from PR #1470 by adavyas. Core fix: Honcho tool calls in a multi-session gateway could route to the wrong session because honcho_tools.py relied on process-global state. Now threads session context through the call chain: AIAgent._invoke_tool() → handle_function_call() → registry.dispatch() → handler **kw → _resolve_session_context() Changes: - Add _resolve_session_context() to prefer per-call context over globals - Plumb honcho_manager + honcho_session_key through handle_function_call - Add sync_honcho=False to run_conversation() for synthetic flush turns - Pass honcho_session_key through gateway memory flush lifecycle - Harden gateway PID detection when /proc cmdline is unreadable - Make interrupt test scripts import-safe for pytest-xdist - Wrap BibTeX examples in Jekyll raw blocks for docs build - Fix thread-order-dependent assertion in client lifecycle test - Expand Honcho docs: session isolation, lifecycle, routing internals Dropped from original PR: - Indentation change in _create_request_openai_client that would move client creation inside the lock (causes unnecessary contention) Co-authored-by: adavyas <adavyas@users.noreply.github.com>
This commit is contained in:
Teknium
GitHub
parent
eb4f0348e1
commit
dd7921d514
17 changed files with 522 additions and 252 deletions
|
|
@ -29,51 +29,6 @@ from unittest.mock import MagicMock, patch
|
|||
from run_agent import AIAgent, IterationBudget
|
||||
from tools.interrupt import set_interrupt, is_interrupted
|
||||
|
||||
set_interrupt(False)
|
||||
|
||||
# ─── Create parent agent ───
|
||||
parent = AIAgent.__new__(AIAgent)
|
||||
parent._interrupt_requested = False
|
||||
parent._interrupt_message = None
|
||||
parent._active_children = []
|
||||
parent.quiet_mode = True
|
||||
parent.model = "test/model"
|
||||
parent.base_url = "http://localhost:1"
|
||||
parent.api_key = "test"
|
||||
parent.provider = "test"
|
||||
parent.api_mode = "chat_completions"
|
||||
parent.platform = "cli"
|
||||
parent.enabled_toolsets = ["terminal", "file"]
|
||||
parent.providers_allowed = None
|
||||
parent.providers_ignored = None
|
||||
parent.providers_order = None
|
||||
parent.provider_sort = None
|
||||
parent.max_tokens = None
|
||||
parent.reasoning_config = None
|
||||
parent.prefill_messages = None
|
||||
parent._session_db = None
|
||||
parent._delegate_depth = 0
|
||||
parent._delegate_spinner = None
|
||||
parent.tool_progress_callback = None
|
||||
parent.iteration_budget = IterationBudget(max_total=100)
|
||||
parent._client_kwargs = {"api_key": "test", "base_url": "http://localhost:1"}
|
||||
|
||||
# Monkey-patch parent.interrupt to log
|
||||
_original_interrupt = AIAgent.interrupt
|
||||
def logged_interrupt(self, message=None):
|
||||
log.info(f"🔴 parent.interrupt() called with: {message!r}")
|
||||
log.info(f" _active_children count: {len(self._active_children)}")
|
||||
_original_interrupt(self, message)
|
||||
log.info(f" After interrupt: _interrupt_requested={self._interrupt_requested}")
|
||||
for i, c in enumerate(self._active_children):
|
||||
log.info(f" Child {i}._interrupt_requested={c._interrupt_requested}")
|
||||
parent.interrupt = lambda msg=None: logged_interrupt(parent, msg)
|
||||
|
||||
# ─── Simulate the exact CLI flow ───
|
||||
interrupt_queue = queue.Queue()
|
||||
child_running = threading.Event()
|
||||
agent_result = [None]
|
||||
|
||||
def make_slow_response(delay=2.0):
|
||||
"""API response that takes a while."""
|
||||
def create(**kwargs):
|
||||
|
|
@ -94,96 +49,154 @@ def make_slow_response(delay=2.0):
|
|||
return create
|
||||
|
||||
|
||||
def agent_thread_func():
|
||||
"""Simulates the agent_thread in cli.py's chat() method."""
|
||||
log.info("🟢 agent_thread starting")
|
||||
def main() -> int:
|
||||
set_interrupt(False)
|
||||
|
||||
with patch("run_agent.OpenAI") as MockOpenAI:
|
||||
mock_client = MagicMock()
|
||||
mock_client.chat.completions.create = make_slow_response(delay=3.0)
|
||||
mock_client.close = MagicMock()
|
||||
MockOpenAI.return_value = mock_client
|
||||
# ─── Create parent agent ───
|
||||
parent = AIAgent.__new__(AIAgent)
|
||||
parent._interrupt_requested = False
|
||||
parent._interrupt_message = None
|
||||
parent._active_children = []
|
||||
parent.quiet_mode = True
|
||||
parent.model = "test/model"
|
||||
parent.base_url = "http://localhost:1"
|
||||
parent.api_key = "test"
|
||||
parent.provider = "test"
|
||||
parent.api_mode = "chat_completions"
|
||||
parent.platform = "cli"
|
||||
parent.enabled_toolsets = ["terminal", "file"]
|
||||
parent.providers_allowed = None
|
||||
parent.providers_ignored = None
|
||||
parent.providers_order = None
|
||||
parent.provider_sort = None
|
||||
parent.max_tokens = None
|
||||
parent.reasoning_config = None
|
||||
parent.prefill_messages = None
|
||||
parent._session_db = None
|
||||
parent._delegate_depth = 0
|
||||
parent._delegate_spinner = None
|
||||
parent.tool_progress_callback = None
|
||||
parent.iteration_budget = IterationBudget(max_total=100)
|
||||
parent._client_kwargs = {"api_key": "test", "base_url": "http://localhost:1"}
|
||||
|
||||
from tools.delegate_tool import _run_single_child
|
||||
# Monkey-patch parent.interrupt to log
|
||||
_original_interrupt = AIAgent.interrupt
|
||||
|
||||
# Signal that child is about to start
|
||||
original_init = AIAgent.__init__
|
||||
def patched_init(self_agent, *a, **kw):
|
||||
log.info("🟡 Child AIAgent.__init__ called")
|
||||
original_init(self_agent, *a, **kw)
|
||||
child_running.set()
|
||||
log.info(f"🟡 Child started, parent._active_children = {len(parent._active_children)}")
|
||||
def logged_interrupt(self, message=None):
|
||||
log.info(f"🔴 parent.interrupt() called with: {message!r}")
|
||||
log.info(f" _active_children count: {len(self._active_children)}")
|
||||
_original_interrupt(self, message)
|
||||
log.info(f" After interrupt: _interrupt_requested={self._interrupt_requested}")
|
||||
for i, child in enumerate(self._active_children):
|
||||
log.info(f" Child {i}._interrupt_requested={child._interrupt_requested}")
|
||||
|
||||
with patch.object(AIAgent, "__init__", patched_init):
|
||||
result = _run_single_child(
|
||||
task_index=0,
|
||||
goal="Do a slow thing",
|
||||
context=None,
|
||||
toolsets=["terminal"],
|
||||
model="test/model",
|
||||
max_iterations=3,
|
||||
parent_agent=parent,
|
||||
task_count=1,
|
||||
override_provider="test",
|
||||
override_base_url="http://localhost:1",
|
||||
override_api_key="test",
|
||||
override_api_mode="chat_completions",
|
||||
)
|
||||
agent_result[0] = result
|
||||
log.info(f"🟢 agent_thread finished. Result status: {result.get('status')}")
|
||||
parent.interrupt = lambda msg=None: logged_interrupt(parent, msg)
|
||||
|
||||
# ─── Simulate the exact CLI flow ───
|
||||
interrupt_queue = queue.Queue()
|
||||
child_running = threading.Event()
|
||||
agent_result = [None]
|
||||
|
||||
# ─── Start agent thread (like chat() does) ───
|
||||
agent_thread = threading.Thread(target=agent_thread_func, name="agent_thread", daemon=True)
|
||||
agent_thread.start()
|
||||
def agent_thread_func():
|
||||
"""Simulates the agent_thread in cli.py's chat() method."""
|
||||
log.info("🟢 agent_thread starting")
|
||||
|
||||
# ─── Wait for child to start ───
|
||||
if not child_running.wait(timeout=10):
|
||||
print("FAIL: Child never started", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
with patch("run_agent.OpenAI") as MockOpenAI:
|
||||
mock_client = MagicMock()
|
||||
mock_client.chat.completions.create = make_slow_response(delay=3.0)
|
||||
mock_client.close = MagicMock()
|
||||
MockOpenAI.return_value = mock_client
|
||||
|
||||
# Give child time to enter its main loop and start API call
|
||||
time.sleep(1.0)
|
||||
from tools.delegate_tool import _run_single_child
|
||||
|
||||
# ─── Simulate user typing a message (like handle_enter does) ───
|
||||
log.info("📝 Simulating user typing 'Hey stop that'")
|
||||
interrupt_queue.put("Hey stop that")
|
||||
# Signal that child is about to start
|
||||
original_init = AIAgent.__init__
|
||||
|
||||
# ─── Simulate chat() polling loop (like the real chat() method) ───
|
||||
log.info("📡 Starting interrupt queue polling (like chat())")
|
||||
interrupt_msg = None
|
||||
poll_count = 0
|
||||
while agent_thread.is_alive():
|
||||
try:
|
||||
interrupt_msg = interrupt_queue.get(timeout=0.1)
|
||||
if interrupt_msg:
|
||||
log.info(f"📨 Got interrupt message from queue: {interrupt_msg!r}")
|
||||
log.info(f" Calling parent.interrupt()...")
|
||||
parent.interrupt(interrupt_msg)
|
||||
log.info(f" parent.interrupt() returned. Breaking poll loop.")
|
||||
break
|
||||
except queue.Empty:
|
||||
poll_count += 1
|
||||
if poll_count % 20 == 0: # Log every 2s
|
||||
log.info(f" Still polling ({poll_count} iterations)...")
|
||||
def patched_init(self_agent, *a, **kw):
|
||||
log.info("🟡 Child AIAgent.__init__ called")
|
||||
original_init(self_agent, *a, **kw)
|
||||
child_running.set()
|
||||
log.info(
|
||||
f"🟡 Child started, parent._active_children = {len(parent._active_children)}"
|
||||
)
|
||||
|
||||
# ─── Wait for agent to finish ───
|
||||
log.info("⏳ Waiting for agent_thread to join...")
|
||||
t0 = time.monotonic()
|
||||
agent_thread.join(timeout=10)
|
||||
elapsed = time.monotonic() - t0
|
||||
log.info(f"✅ agent_thread joined after {elapsed:.2f}s")
|
||||
with patch.object(AIAgent, "__init__", patched_init):
|
||||
result = _run_single_child(
|
||||
task_index=0,
|
||||
goal="Do a slow thing",
|
||||
context=None,
|
||||
toolsets=["terminal"],
|
||||
model="test/model",
|
||||
max_iterations=3,
|
||||
parent_agent=parent,
|
||||
task_count=1,
|
||||
override_provider="test",
|
||||
override_base_url="http://localhost:1",
|
||||
override_api_key="test",
|
||||
override_api_mode="chat_completions",
|
||||
)
|
||||
agent_result[0] = result
|
||||
log.info(f"🟢 agent_thread finished. Result status: {result.get('status')}")
|
||||
|
||||
# ─── Check results ───
|
||||
result = agent_result[0]
|
||||
if result:
|
||||
log.info(f"Result status: {result['status']}")
|
||||
log.info(f"Result duration: {result['duration_seconds']}s")
|
||||
if result["status"] == "interrupted" and elapsed < 2.0:
|
||||
print("✅ PASS: Interrupt worked correctly!", file=sys.stderr)
|
||||
else:
|
||||
# ─── Start agent thread (like chat() does) ───
|
||||
agent_thread = threading.Thread(target=agent_thread_func, name="agent_thread", daemon=True)
|
||||
agent_thread.start()
|
||||
|
||||
# ─── Wait for child to start ───
|
||||
if not child_running.wait(timeout=10):
|
||||
print("FAIL: Child never started", file=sys.stderr)
|
||||
set_interrupt(False)
|
||||
return 1
|
||||
|
||||
# Give child time to enter its main loop and start API call
|
||||
time.sleep(1.0)
|
||||
|
||||
# ─── Simulate user typing a message (like handle_enter does) ───
|
||||
log.info("📝 Simulating user typing 'Hey stop that'")
|
||||
interrupt_queue.put("Hey stop that")
|
||||
|
||||
# ─── Simulate chat() polling loop (like the real chat() method) ───
|
||||
log.info("📡 Starting interrupt queue polling (like chat())")
|
||||
interrupt_msg = None
|
||||
poll_count = 0
|
||||
while agent_thread.is_alive():
|
||||
try:
|
||||
interrupt_msg = interrupt_queue.get(timeout=0.1)
|
||||
if interrupt_msg:
|
||||
log.info(f"📨 Got interrupt message from queue: {interrupt_msg!r}")
|
||||
log.info(" Calling parent.interrupt()...")
|
||||
parent.interrupt(interrupt_msg)
|
||||
log.info(" parent.interrupt() returned. Breaking poll loop.")
|
||||
break
|
||||
except queue.Empty:
|
||||
poll_count += 1
|
||||
if poll_count % 20 == 0: # Log every 2s
|
||||
log.info(f" Still polling ({poll_count} iterations)...")
|
||||
|
||||
# ─── Wait for agent to finish ───
|
||||
log.info("⏳ Waiting for agent_thread to join...")
|
||||
t0 = time.monotonic()
|
||||
agent_thread.join(timeout=10)
|
||||
elapsed = time.monotonic() - t0
|
||||
log.info(f"✅ agent_thread joined after {elapsed:.2f}s")
|
||||
|
||||
# ─── Check results ───
|
||||
result = agent_result[0]
|
||||
if result:
|
||||
log.info(f"Result status: {result['status']}")
|
||||
log.info(f"Result duration: {result['duration_seconds']}s")
|
||||
if result["status"] == "interrupted" and elapsed < 2.0:
|
||||
print("✅ PASS: Interrupt worked correctly!", file=sys.stderr)
|
||||
set_interrupt(False)
|
||||
return 0
|
||||
print(f"❌ FAIL: status={result['status']}, elapsed={elapsed:.2f}s", file=sys.stderr)
|
||||
else:
|
||||
print("❌ FAIL: No result returned", file=sys.stderr)
|
||||
set_interrupt(False)
|
||||
return 1
|
||||
|
||||
set_interrupt(False)
|
||||
print("❌ FAIL: No result returned", file=sys.stderr)
|
||||
set_interrupt(False)
|
||||
return 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue