fix: salvage gateway dedup and executor cleanup from PR #993

Salvages the two still-relevant fixes from PR #993 onto current main:
- use a 3-tuple LOCAL delivery key so explicit/local-origin targets are not duplicated
- shut down the previous agent-loop ThreadPoolExecutor when resizing the global pool

Adds regression tests for both behaviors.
This commit is contained in:
Himess 2026-03-14 11:03:20 -07:00 committed by teknium1
parent 429c44e377
commit e5dc569daa
4 changed files with 32 additions and 2 deletions

View file

@ -39,7 +39,9 @@ def resize_tool_pool(max_workers: int):
Safe to call before any tasks are submitted.
"""
global _tool_executor
old_executor = _tool_executor
_tool_executor = concurrent.futures.ThreadPoolExecutor(max_workers=max_workers)
old_executor.shutdown(wait=False)
logger.info("Tool thread pool resized to %d workers", max_workers)
logger = logging.getLogger(__name__)

View file

@ -161,7 +161,7 @@ class DeliveryRouter:
# Always include local if configured
if self.config.always_log_local:
local_key = (Platform.LOCAL, None)
local_key = (Platform.LOCAL, None, None)
if local_key not in seen_platforms:
targets.append(DeliveryTarget(platform=Platform.LOCAL))

View file

@ -1,7 +1,7 @@
"""Tests for the delivery routing module."""
from gateway.config import Platform, GatewayConfig, PlatformConfig, HomeChannel
from gateway.delivery import DeliveryTarget, parse_deliver_spec
from gateway.delivery import DeliveryRouter, DeliveryTarget, parse_deliver_spec
from gateway.session import SessionSource
@ -85,3 +85,12 @@ class TestTargetToStringRoundtrip:
reparsed = DeliveryTarget.parse(s)
assert reparsed.platform == Platform.TELEGRAM
assert reparsed.chat_id == "999"
class TestDeliveryRouter:
def test_resolve_targets_does_not_duplicate_local_when_explicit(self):
router = DeliveryRouter(GatewayConfig(always_log_local=True))
targets = router.resolve_targets(["local"])
assert [target.platform for target in targets] == [Platform.LOCAL]

View file

@ -484,3 +484,22 @@ class TestResizeToolPool:
"""resize_tool_pool should not raise."""
resize_tool_pool(16) # Small pool for testing
resize_tool_pool(128) # Restore default
def test_resize_shuts_down_previous_executor(self, monkeypatch):
"""Replacing the global tool executor should shut down the old pool."""
import environments.agent_loop as agent_loop_module
old_executor = MagicMock()
new_executor = MagicMock()
monkeypatch.setattr(agent_loop_module, "_tool_executor", old_executor)
monkeypatch.setattr(
agent_loop_module.concurrent.futures,
"ThreadPoolExecutor",
MagicMock(return_value=new_executor),
)
resize_tool_pool(16)
old_executor.shutdown.assert_called_once_with(wait=False)
assert agent_loop_module._tool_executor is new_executor