Plan — ToolDispatcher extraction

Extract _execute_tool and _normalize_tool_name from GeneratorAgent into a dedicated ToolDispatcher helper. State transitions stay in the generator.

Motivation

GeneratorAgent._execute_tool() owns bus I/O mechanics that have nothing to do with generation: opening a short-lived subscriber, publishing a tool request, polling for a matching correlation_id, handling timeout. This is reusable bus plumbing. _normalize_tool_name() is similarly a tool-registry concern.

Both belong in a dedicated helper, injected into the generator the same way CompactionService was.

What ToolDispatcher is

ToolDispatcher is a synchronous Participant. It has its own ZmqPublisher, CONFIG_NAME, and self.id. It is a bus participant in spirit — it owns a bus connection and has a single clear responsibility — but has no event-driven run loop. It is called synchronously by the generator mid-generation turn. This is consistent with the architectural invariant that tool calls are synchronous within a generation turn.

Responsibility Split

Concern	Owner	Why
Bus I/O: open subscriber, publish request, poll, close subscriber	ToolDispatcher	Generic bus plumbing, reusable, testable in isolation
Name normalization: map hallucinated names to registered schema names	ToolDispatcher	Tool-registry logic, not generation logic
State transitions: DISPATCH_TOOL, AWAIT_RESULT, TOOL_RESULT, TOOL_TIMEOUT	GeneratorAgent	Generator state machine; transitions must stay co-located with the machine they describe
Tool schema registry: maintain, register, expose	GeneratorAgent (unchanged)	Registry is used for status publishing and schema re-broadcast — broader than tool dispatch

ToolDispatcher API

File: src/local/tools/tool_dispatcher.py

class ToolDispatcher:
    """Bus I/O helper for synchronous tool call dispatch.

    Owned by GeneratorAgent. Uses the generator's publisher to send
    tool.request.* and opens a short-lived subscriber to receive
    tool.result.*. State transitions are the caller's responsibility.
    """

    CONFIG_NAME = "tool_dispatcher"

    def __init__(self, tool_timeout: float) -> None:
        self._tool_timeout = tool_timeout
        self._pub = ZmqPublisher(PROXY_FRONTEND_ADDR, bind=False)

    def execute(
        self,
        name: str,
        args: dict,
        correlation_id: str,
        schemas: list,
    ) -> tuple[str, bool]:
        """Dispatch a tool call and block for the result.

        Opens a ZmqSubscriber for tool.result. BEFORE publishing
        tool.request. to avoid a race between publish and subscribe.
        Polls until correlation_id matches or tool_timeout expires.

        Args:
            name: Tool function name; normalized via _normalize() first.
            args: Tool arguments dict from the model's tool call.
            correlation_id: Used to match the response envelope.
            schemas: Current tool schema list; used for name normalization.

        Returns:
            (result, timed_out): result is the tool output string;
            timed_out is True if no matching response arrived in time.
        """
        name = self._normalize(name, schemas)
        req_subject = f"tool.request.{name}"
        res_subject = f"tool.result.{name}"

        result_sub = ZmqSubscriber(PROXY_BACKEND_ADDR, subscriptions=[res_subject])
        try:
            self._pub.publish(MessageEnvelope.create(
                message_type="tool_request",
                subject=req_subject,
                sender_id=self.id,
                payload={"tool": name, "args": args},
                correlation_id=correlation_id,
            ))
            deadline = time.monotonic() + self._tool_timeout
            while time.monotonic() < deadline:
                remaining_ms = max(1, int((deadline - time.monotonic()) * 1000))
                msg = result_sub.receive_with_timeout(remaining_ms)
                if msg is None:
                    break
                if msg.correlation_id == correlation_id:
                    return msg.payload.get("result", ""), False
        finally:
            result_sub.close()

        return f"[tool timeout: {name!r} did not respond within {self._tool_timeout}s]", True

    def _normalize(self, name: str, schemas: list) -> str:
        registered = {s.get("function", {}).get("name") for s in schemas}
        if name in registered:
            return name
        name_lower = name.lower()
        for rname in registered:
            if rname in name_lower or name_lower in rname:
                logger.warning("ToolDispatcher: normalizing tool name %r → %r", name, rname)
                return rname
        return name

sender_id on the tool request: currently hardcoded as "generator" inside _execute_tool via self._make_envelope. The full implementation passes the generator's self.id to execute() as a parameter so ToolDispatcher doesn't reach into generator state.

GeneratorAgent changes

init

Accept injected ToolDispatcher (constructed in run.py with tool_timeout from config):

def __init__(
    self,
    model: str | None = None,
    temperature: float | None = None,
    conversation_service=None,
    compaction_service=None,
    tool_dispatcher=None,           # ← add
) -> None:
    ...
    self._tool_dispatcher = tool_dispatcher or ToolDispatcher(self._tool_timeout)

Fallback construction in __init__ keeps tests that instantiate GeneratorAgent directly working without wiring up run.py.

_generate() — state transitions move out of _execute_tool, stay in _generate()

# before (inside for tc in tool_calls loop):
self._do_transition(GeneratorAction.DISPATCH_TOOL)
result = self._execute_tool(name, args, correlation_id)

# after:
self._do_transition(GeneratorAction.DISPATCH_TOOL)
self._do_transition(GeneratorAction.AWAIT_RESULT)
result, timed_out = self._tool_dispatcher.execute(
    name, args, correlation_id, self._tool_schemas
)
if timed_out:
    self._do_transition(GeneratorAction.TOOL_TIMEOUT)
else:
    self._do_transition(GeneratorAction.TOOL_RESULT)

Methods removed

_execute_tool() — replaced by self._tool_dispatcher.execute()
_normalize_tool_name() — moved to ToolDispatcher._normalize()

Imports removed

PROXY_BACKEND_ADDR from bus_config (used only by _execute_tool)
ZmqSubscriber (used only by _execute_tool)

run.py change

from local.tools.tool_dispatcher import ToolDispatcher

_gen_cfg = get_config("generator") or {}
shared_tool_dispatcher = ToolDispatcher(
    tool_timeout=_gen_cfg.get("tool_timeout", 20),
)

# Pass into generator:
gen_thread = threading.Thread(
    target=_start_generator, args=(args.model,),
    kwargs={
        "conversation_service": shared_conv,
        "compaction_service": shared_compaction,
        "tool_dispatcher": shared_tool_dispatcher,
    },
    ...
)

Transition sequence — before and after

Before (all in _execute_tool): _generate() calls: DISPATCH_TOOL _execute_tool(): open subscriber AWAIT_RESULT publish tool.request.* poll ... TOOL_RESULT (or TOOL_TIMEOUT) close subscriber After (transitions in _generate, I/O in ToolDispatcher): _generate(): DISPATCH_TOOL AWAIT_RESULT tool_dispatcher.execute(): open subscriber publish tool.request.* poll ... close subscriber return (result, timed_out) _generate(): TOOL_RESULT (or TOOL_TIMEOUT)

The state machine sees the same transition sequence. The only change is that AWAIT_RESULT now fires before the subscriber opens rather than after. This is safe — AWAIT_RESULT is a visibility signal, not a coordination mechanism. The race-condition protection is subscribe-before-publish, which ToolDispatcher preserves internally.

File Map

File	Change
`src/local/tools/tool_dispatcher.py`	new — bus I/O + name normalization
`src/local/agents/generator_agent.py`	inject tool_dispatcher; remove _execute_tool + _normalize_tool_name; transitions moved inline in _generate()
`src/local/run.py`	construct ToolDispatcher; inject into generator
`config/tool_dispatcher.yaml`	id: tool_dispatcher
`src/local/defaults/tool_dispatcher.yaml`	id: tool_dispatcher

Acceptance Criteria

Generator state machine sees the same DISPATCH_TOOL → AWAIT_RESULT → TOOL_RESULT/TIMEOUT sequence
Tool calls still execute with correct correlation_id matching
Tool timeout still returns an error string and logs a warning
Hallucinated tool names are still normalized before dispatch
_execute_tool and _normalize_tool_name no longer exist in GeneratorAgent
All existing tests pass