Executor

Responsible for actually running the chosen handler(s) after the pipeline has selected candidates: single/sequential/parallel execution, retry with backoff, streaming, and tool chaining. All execution paths call Route.execute/Route.execute_async so retry/timing logic stays in one place.

_with_retry (module-level helper)

def _with_retry(fn: Callable[[], HandlerResponse], max_tries: int, delay: float, on_exceptions: List[str]) -> HandlerResponse

Runs fn() up to max_tries times, sleeping delay seconds between attempts.

Algorithm:

1. Call fn(). If response.success, return it immediately — no retry needed.
2. Otherwise store it as last_response.
3. Selective retry: if on_exceptions is non-empty and response.error is set, only continue retrying if any string in on_exceptions appears as a substring of response.error; if none match, break out of the retry loop immediately (don't waste further attempts on an error type that isn't in the retry list).
4. If not the last attempt, time.sleep(delay).
5. After the loop, return last_response.

Parameters:

Returns: The last HandlerResponse obtained (which may be a failure).

[!NOTE] Retry is response-based, not exception-based, despite the name Handlers are expected to catch their own exceptions and return HandlerResponse.fail(...) (which Route.execute/execute_async already do automatically by catching handler exceptions). _with_retry never itself catches a raised exception from fn() — if fn() raises rather than returning a failed HandlerResponse, that exception propagates out of _with_retry uncaught. The on_exceptions list (default ["RuntimeError", "TimeoutError"]) is matched against substrings of the response's error message string, not against actual Python exception types — this only works cleanly if your HandlerResponse.fail(...) error messages happen to contain those class-name substrings (which is exactly what Route.execute's own exception-wrapping produces: f"Handler raised {type(exc).__name__}: {exc}").

Blocking behavior: Uses time.sleep, which blocks the calling thread — in _execute_parallel, each retry sequence for a given candidate runs inside its own ThreadPoolExecutor worker thread, so sleeping one candidate's retry doesn't block others; there is no async-native backoff (asyncio.sleep) used anywhere — _execute_single_async/_execute_sequential_async/_execute_parallel_async do not call _with_retry at all (see below).

Related APIs: ExecutionConfig, Route.execute

ExecutionEngine

class ExecutionEngine:
    def __init__(self, cfg: ExecutionConfig, route_map: dict, aggregator: Optional[BaseAggregator] = None): ...

Constructor

Parameters:

Synchronous execution

execute(self, request: HandlerRequest, candidates: List[RouteCandidate], trace: RoutingTrace) -> RoutingResult

Top-level dispatcher. Returns an immediate failed RoutingResult if candidates is empty. Otherwise dispatches on self._cfg.mode:

Raises: ValueError(f"Unknown execution mode: {mode}") — defensive branch, unreachable via normal ExecutionMode enum values.

_execute_single(self, request, candidate, trace) -> RoutingResult

Looks up the route by candidate.route_name in route_map; if missing, returns a failed result ("Route '{name}' not found in registry.") without touching the trace's timing. Otherwise runs route.execute(request, candidate.combined_score) wrapped in _with_retry, times the whole retry sequence into trace.execution_time_ms, and returns RoutingResult(response, trace, matched=response.success).

_execute_sequential(self, request, candidates, trace) -> RoutingResult

Iterates candidates in ranked order. For each: skip if the route isn't in route_map; otherwise run with retry. On first success, stamp trace.selected_route/selected_group, add a note, and return immediately (does not try remaining candidates). On failure, add a note and continue to the next candidate. If every candidate fails, return a failed result ("All sequential candidates failed.").

Complexity: O(k) route executions in the worst case (all candidates fail), where k = len(candidates) (bounded by ScoringConfig.top_k).

_execute_parallel(self, request, candidates, trace) -> MultiRoutingResult

Submits one _with_retry(...) call per valid candidate to a ThreadPoolExecutor(max_workers=len(candidates)), then collects results via as_completed(futures, timeout=self._cfg.parallel_timeout).

Timeout behavior: If the overall as_completed iteration exceeds parallel_timeout, a concurrent.futures.TimeoutError is caught, a trace note is added ("Parallel execution timed out after {timeout}s."), and whatever results had already completed before the timeout are kept — the method does not raise, it proceeds with a partial results list (possibly empty).

Aggregation: If self._aggregator is set and there are results, delegate to self._aggregator.aggregate(results). Otherwise (no aggregator configured), fall back to a first-success scan, and if nothing succeeded, use results[0].response (even if it's a failure) as the aggregated response — this fallback path always produces some aggregated response as long as results is non-empty, but it may be a failed one.

Thread safety of the shared trace object: Multiple worker threads execute concurrently, but only the main thread mutates trace (workers only compute HandlerResponses; trace notes/timing are set back on the calling thread after as_completed resolves) — so RoutingTrace mutation itself is not concurrent in this method, though the underlying handlers' own thread safety is the caller's responsibility.

Async execution

execute_async(self, request, candidates, trace) -> RoutingResult (async)

Mirrors execute()'s dispatch logic for SINGLE/SEQUENTIAL/PARALLEL, calling the _async variants.

_execute_single_async / _execute_sequential_async

Same control flow as their sync counterparts, but call await route.execute_async(request, candidate.combined_score) directly.

[!WARNING] Async execution paths do not use _with_retry Unlike the synchronous _execute_single/_execute_sequential, their async counterparts call route.execute_async(...) directly, with no retry wrapper. ExecutionConfig.max_retries/retry_delay/retry_on_exceptions have no effect on async execution as implemented in the analyzed source — a single async attempt is made per candidate, full stop. If your application relies on retry behavior, be aware it differs between route_query/route() (sync, retried) and route_async() (async, not retried).

_execute_parallel_async(self, request, candidates, trace) -> MultiRoutingResult (async)

Builds one route.execute_async(...) coroutine per valid candidate (no retry wrapper — see warning above), gathers them with asyncio.wait_for(asyncio.gather(*tasks, return_exceptions=True), timeout=self._cfg.parallel_timeout).

Exception handling: return_exceptions=True means a handler coroutine raising an exception does not abort the whole gather — instead, the exception object itself appears in the responses list at that position, and is converted to HandlerResponse.fail(str(exc)) before being wrapped in a RoutingResult.

Timeout behavior: On asyncio.TimeoutError, responses is set to an empty list (not partial results — this differs from the sync _execute_parallel's behavior of keeping whatever completed before the timeout) and a trace note is added. Since responses is empty, the subsequent zip(valid_candidates, responses) produces zero results, so results ends up empty and aggregated stays None.

Aggregation: This method does not consult self._aggregator at all (unlike sync _execute_parallel) — it always uses an inline first-success-else-first-response fallback directly.

Streaming

stream(self, request: HandlerRequest, candidates: List[RouteCandidate]) -> Iterator[str]

Yields "[ERROR] No route candidates found." if candidates is empty, or "[ERROR] Route not found." if the top candidate's route isn't in route_map. Otherwise delegates entirely to route.handler.stream(request) — only the single best-ranked candidate is ever considered; there is no fallback to the next candidate on a streaming failure.

stream_async(self, request, candidates) -> AsyncIterator[str]

Async counterpart; delegates to route.handler.stream_async(request).

[!NOTE] Error signaling via yielded strings, not exceptions Both streaming methods signal "no candidates" / "route not found" by yielding an error string (prefixed [ERROR]) rather than raising an exception. Consumers of stream()/stream_async() that treat every yielded chunk as legitimate content should check for this prefix if they need to distinguish real output from these two specific failure conditions.

Tool chaining

chain(self, request: HandlerRequest, route_names: List[str]) -> HandlerResponse

Executes a fixed sequence of routes, threading state through a shared, mutable context dict copied from request.context.

Algorithm:

1. context = dict(request.context) (shallow copy).
2. For each name in route_names, in order:
   • Look up the route; if missing, return HandlerResponse.fail(f"Chain broken: route '{name}' not found.") immediately (chain aborts).
   • Build a fresh HandlerRequest reusing request.query/request_id/metadata but with the accumulated context.
   • Execute the route synchronously (route.execute(chained_request) — no retry wrapper, no async variant of chain exists in the analyzed source).
   • If the step fails, return HandlerResponse.fail(f"Chain broken at '{name}': {response.error}") immediately (chain aborts).
   • On success, set context[f"step_{name}"] = response.content and context["last_result"] = response.content for the next step to consume, and remember it as last_response.
3. Return last_response, or HandlerResponse.fail("Empty chain.") if route_names was empty (loop never executed).

Side effects: None beyond the local context dict (does not mutate request.context itself, since it's shallow-copied at the start — though note a shallow copy means nested mutable objects inside the original context are still shared).

Examples

1. Basic — single mode (default):

from zerda.router.execution.executor import ExecutionEngine, FirstWinsAggregator
from zerda.router import ExecutionConfig

engine = ExecutionEngine(ExecutionConfig(), route_map={"docs_qa": docs_route}, aggregator=FirstWinsAggregator())
result = engine.execute(request, candidates, trace)

2. Sequential fallback:

from zerda.router import ExecutionConfig, ExecutionMode
engine = ExecutionEngine(ExecutionConfig(mode=ExecutionMode.SEQUENTIAL), route_map=route_map)
result = engine.execute(request, candidates, trace)  # tries candidates[0], then [1], ... until one succeeds

3. Parallel with merge aggregation:

from zerda.router.execution.executor import MergeAggregator
engine = ExecutionEngine(
    ExecutionConfig(mode=ExecutionMode.PARALLEL, parallel_timeout=5.0),
    route_map=route_map,
    aggregator=MergeAggregator(),
)
result = engine.execute(request, candidates, trace)
print(result.response.content)  # a list of every successful candidate's content

4. Error handling — chain aborting partway:

response = engine.chain(request, ["step_a", "missing_step", "step_c"])
print(response.success)  # False
print(response.error)    # "Chain broken: route 'missing_step' not found."

5. Production — tool chaining with per-step context accumulation:

response = engine.chain(request, ["extract_intent", "retrieve_docs", "generate_answer"])
# Inside "generate_answer"'s handler, request.context["last_result"] holds
# retrieve_docs's output, and request.context["step_retrieve_docs"] holds
# the same value under a step-specific key.

Best practices

• If you need retries on the async path, implement them at the handler level (inside handle_async) rather than relying on ExecutionConfig — the async execution methods do not apply _with_retry.
• Prefer SEQUENTIAL mode over relying on chain() for fallback-style "try route A, then B" behavior; chain() is for pipelines where every step must succeed, not for trying alternatives.
• When using PARALLEL mode, remember async and sync parallel execution have different timeout-partial-results semantics (sync keeps completed results, async discards all results on timeout) and different aggregation logic (sync respects self._aggregator, async ignores it) — don't assume behavioral parity between route_query and route_async under PARALLEL mode.

Related APIs: RoutingResult, MultiRoutingResult, BaseAggregator

BaseAggregator

class BaseAggregator:
    def aggregate(self, results: List[RoutingResult]) -> HandlerResponse:
        raise NotImplementedError

Not an ABC (no abstractmethod decorator in the analyzed source) — aggregate simply raises NotImplementedError if not overridden; Python will not prevent instantiating BaseAggregator directly, but calling .aggregate(...) on it will fail.

FirstWinsAggregator

class FirstWinsAggregator(BaseAggregator):
    def aggregate(self, results: List[RoutingResult]) -> HandlerResponse: ...

Returns the response of the first RoutingResult in results where r.success is True. If none succeeded, returns HandlerResponse.fail("No successful responses from any candidate."). This is the default aggregator HierarchicalRouter constructs when ExecutionConfig.aggregation is anything other than "merge" (see Configuration → ExecutionConfig).

MergeAggregator

class MergeAggregator(BaseAggregator):
    def aggregate(self, results: List[RoutingResult]) -> HandlerResponse: ...

Collects r.content from every successful result into a list: merged = [r.content for r in results if r.success]. Returns HandlerResponse.fail("No successful responses to merge.") if the list is empty, otherwise HandlerResponse.ok(merged) — note the merged payload is a flat list of contents, not a dict keyed by route name; callers need results/trace (available separately on MultiRoutingResult) to know which content came from which route.

Example: custom aggregator

from zerda.router.execution.executor import BaseAggregator
from zerda.router import HandlerResponse

class NamedMergeAggregator(BaseAggregator):
    """Merge results into a dict keyed by route name instead of a flat list."""
    def aggregate(self, results):
        merged = {
            r.route_name: r.content
            for r in results
            if r.success
        }
        if not merged:
            return HandlerResponse.fail("No successful responses to merge.")
        return HandlerResponse.ok(merged)

Related APIs: ExecutionEngine, MultiRoutingResult