> ## Documentation Index > Fetch the complete documentation index at: https://jephalabs.mintlify.site/llms.txt > Use this file to discover all available pages before exploring further. # Executors Overview > Architecture of the executor system — dispatch, observe, and recover # Executor Architecture Overview > **Source files** > > * `src/qtg/infrastructure/executors/__init__.py` > * `src/qtg/infrastructure/executors/registry.py` > * `src/qtg/infrastructure/executors/local_adapter.py` > * `src/qtg/infrastructure/executors/remote_proxy.py` > * `src/qtg/infrastructure/bootstrap.py` *** ## 1. Architecture Overview The Movement Guard (MG) v3 executor is a **node-level execution engine**. Each node in a movement plan is bound to one executor, and the executor advances node state through five lifecycle methods (`preflight` / `prepare` / `submit` / `observe` / `recover`). Depending on the executor deployment model, they are divided into two categories: | Form | Class | Description | | ---------- | ---------------------- | ----------------------------------------------------------- | | **Local** | `LocalExecutorAdapter` | Wraps a handler object in the same process | | **Remote** | `RemoteExecutorProxy` | Delegates to an external service over an HTTP JSON protocol | Both classes implement the same `ExecutorProtocol` interface, so the dispatcher/observer can use them without distinguishing between local and remote. ```mermaid theme={null} flowchart TB subgraph App["Dispatcher / Observer (application layer)"] Get["get_executor(node.executor_binding)"] Reg["Executor Registry
dict[str, executor]"] Local["Local
Adapter"] Remote["Remote
Proxy"] H1["handler.xxx()"] H2["HTTP POST /xxx"] Get --> Reg Reg --> Local Reg --> Remote Local --> H1 Remote --> H2 end ``` *** ## 2. ExecutorProtocol `@runtime_checkable` protocol defined in `domain/protocols.py`: ```python theme={null} class ExecutorProtocol(Protocol): executor_key: str async def preflight(self, context: ExecutionContext) -> ExecutionResult | None: ... async def prepare(self, context: ExecutionContext) -> ExecutionResult: ... async def submit( self, context: ExecutionContext, prepared_action: PreparedAction, **kwargs, ) -> ExecutionResult: ... async def observe(self, context: ExecutionContext) -> ExecutionResult: ... async def recover(self, context: ExecutionContext) -> ExecutionResult: ... async def health(self) -> dict[str, Any] | None: ... ``` Role of each method: | Method | When called | Return meaning | | ----------- | --------------------------------------------- | ---------------------------------------------------------------------- | | `preflight` | Immediately before dispatch | `None` = pass, `ExecutionResult` = blocked (error\_code + next\_state) | | `prepare` | After preflight passes | Returns `ExecutionResult` including `PreparedAction` | | `submit` | Right after prepare, once signing is complete | Performs the actual external call, returns `ExecutionResult` | | `observe` | Nodes in SUBMITTED/OBSERVING state | Poll result: COMPLETED/OBSERVING/FAILED | | `recover` | Nodes in UNKNOWN state | Recovery-attempt result | | `health` | Registry diagnostics | executor status dict | In addition, there are `cancel` and `close` methods: * `cancel`: supported only by some executors (for example, CEX withdrawal cancellation) * `close`: cleans up resources such as internal HTTP clients *** ## 3. LocalExecutorAdapter **File**: `infrastructure/executors/local_adapter.py` Wraps an in-process handler object with the `ExecutorProtocol` interface. ```python theme={null} class LocalExecutorAdapter: def __init__(self, *, executor_key: str, handler) -> None: self.executor_key = executor_key self._handler = handler ``` All methods delegate directly to `self._handler`. Core behavior: | Method | Behavior | | ----------- | ------------------------------------------------------------------ | | `preflight` | `handler.preflight(ctx)` | | `prepare` | `handler.prepare(ctx)` | | `submit` | `handler.submit(ctx, prepared_action, **kwargs)` | | `observe` | `handler.observe(ctx)` | | `recover` | `handler.recover(ctx)` | | `cancel` | raises `NotImplementedError` if the handler has no `cancel` method | | `health` | hard-coded to `{"ok": True, "executor_key": ..., "mode": "local"}` | | `close` | calls the handler if it has a `close` method | Note that the health check does not call the handler and always returns `ok: True`. *** ## 4. RemoteExecutorProxy **File**: `infrastructure/executors/remote_proxy.py` A proxy that delegates executor calls to an external HTTP service. ### 4.1 Constructor ```python theme={null} class RemoteExecutorProxy: def __init__( self, *, executor_key: str, base_url: str, auth_token: str | None = None, timeout: float = 30.0, client: httpx.AsyncClient | None = None, ) -> None: ``` | Parameter | Description | | -------------- | ----------------------------------------------------------------------------------- | | `executor_key` | Registry lookup key | | `base_url` | Remote service URL (trailing slash removed automatically) | | `auth_token` | Bearer token (optional) | | `timeout` | Defaults to 30 seconds | | `client` | May be injected externally; if `None`, created internally (and cleaned up on close) | ### 4.2 HTTP JSON Protocol All calls are JSON requests in the form `POST {base_url}/{method}`. | Method | HTTP request | Request Body | Response | | ----------- | ----------------- | ---------------------------------------------------- | --------------------------------------------- | | `preflight` | `POST /preflight` | `ExecutionContext (json)` | `204` = pass(None), `2xx` = `ExecutionResult` | | `prepare` | `POST /prepare` | `ExecutionContext (json)` | `ExecutionResult` | | `submit` | `POST /submit` | `{"context": ..., "prepared_action": ..., **kwargs}` | `ExecutionResult` | | `observe` | `POST /observe` | `ExecutionContext (json)` | `ExecutionResult` | | `recover` | `POST /recover` | `ExecutionContext (json)` | `ExecutionResult` | | `cancel` | `POST /cancel` | `ExecutionContext (json)` | `ExecutionResult` | | `health` | `GET /health` | (none) | `dict` | Note that the `submit` body shape differs from the other methods. `context` and `prepared_action` are sent as separate keys, and additional kwargs such as `sign_result` are merged at the top level. Auth header: if `auth_token` is configured, every request includes `Authorization: Bearer {token}`. ### 4.3 close behavior If `client` is injected externally, `_owns_client = False` and `aclose()` is not called on close. Cleanup happens only for internally created clients. *** ## 5. Executor Registry **File**: `infrastructure/executors/registry.py` Registers and retrieves executors from a module-level `dict[str, object]`. ```python theme={null} _EXECUTOR_REGISTRY: dict[str, object] = {} def register_executor(executor) -> None: _EXECUTOR_REGISTRY[executor.executor_key] = executor def get_executor(binding: str | Mapping[str, object]): if isinstance(binding, str): key = binding else: key = str(binding["executor_key"]) return _EXECUTOR_REGISTRY[key] def iter_executors(): return tuple(_EXECUTOR_REGISTRY.values()) def clear_executor_registry() -> None: _EXECUTOR_REGISTRY.clear() ``` ### 5.1 executor\_binding and registry lookup `MovementRequestNode.executor_binding` is stored in the DB as JSON and has the following shape: ```json theme={null} {"executor_key": "exec.cex.withdrawal_action"} ``` When `get_executor(binding)` receives this dict, it extracts `binding["executor_key"]` and looks it up in the registry. A plain string is also allowed. ### 5.2 Built-in executor key list | executor\_key | Category | Gating | Description | | ----------------------------------------- | -------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ---------------------------------------------------------- | | `exec.cex.withdrawal_action` | CEX | always | Withdrawal submission | | `exec.cex.withdrawal_observe` | CEX | always | Withdrawal status polling | | `exec.cex.deposit_observe` | CEX | always | Deposit status polling | | `exec.cctp.burn` | CCTP | EVM endpoints present | burn tx submission | | `exec.cctp.mint` | CCTP | EVM endpoints present | mint tx submission | | `exec.evm.erc20_transfer` | EVM | EVM endpoints present | direct ERC-20 transfer tx | | `exec.ccip.send` | CCIP | `ccip_enabled` | Chainlink CCIP send | | `exec.usdt0.send` | USDT0 | `usdt0_enabled` + EVM endpoints | USDT0 (LayerZero native-OFT) send | | `exec.gateway.approve` | Gateway | `gateway_api_base_url` + EVM endpoints | ERC-20 approve for Gateway | | `exec.gateway.deposit` | Gateway | `gateway_api_base_url` + EVM endpoints | Gateway deposit tx | | `exec.gateway.deposit_observe` | Gateway | `gateway_api_base_url` + EVM endpoints | Gateway deposit observation | | `exec.gateway.intent` | Gateway | `gateway_api_base_url` + EVM endpoints | Gateway burn-intent (EIP-712) submission | | `exec.gateway.mint` | Gateway | `gateway_api_base_url` + EVM endpoints | Gateway mint tx | | `exec.gateway.mint_observe` | Gateway | `gateway_api_base_url` + EVM endpoints | Gateway mint observation | | `exec.hyperliquid.topup` | Hyperliquid | `hl_master_address` + signer + EVM endpoints | Hyperliquid USDC top-up | | `exec.hyperliquid.withdraw` | Hyperliquid | `hl_master_address` + signer + EVM endpoints | Hyperliquid USDC withdrawal | | `exec.observe.destination_chain_receive` | Observe | always | Destination-chain receive confirmation | | `exec.observe.destination_chain_finality` | Observe | always | Destination-chain finality confirmation | | `exec.observe.evm_balance` | Observe | always | EVM balance threshold observation | | `exec.observe.evm_finality` | Observe | always | EVM tx finality observation | | `exec.observe.protocol` | Observe | always | Protocol-level proof confirmation (CCTP attestation, etc.) | | `exec.stargate.send` | Pro (Stargate) | Pro package/loader gate + `MG_STARGATE_ENABLED` + required EVM RPC endpoints | Stargate bridge send | | `exec.stargate.dst_observe` | Pro (Stargate) | Pro package/loader gate + `MG_STARGATE_ENABLED` + required EVM RPC endpoints | Stargate destination-chain observation | | `lighter_secure_withdraw` | Pro (Lighter) | Pro loader; `AwsKmsEvmSigner`; configured L1 owner, contract, and account index; successful KMS address resolution with owner == KMS address; L1 RPC via explicit Lighter endpoint or EVM endpoints | Lighter secure withdrawal | The Stargate executor ships in the Pro distribution and is registered through the package/loader gate — it is **excluded from the OSS distribution**. Lighter registration for `lighter_secure_withdraw` requires the Pro loader, an `AwsKmsEvmSigner`, configured L1 owner, contract, and account index, successful KMS address resolution with owner == KMS address, and L1 RPC via an explicit Lighter endpoint or EVM endpoints. *** ## 6. Health Check Protocol ### Local ```python theme={null} async def health(self) -> dict[str, object]: return {"ok": True, "executor_key": self.executor_key, "mode": "local"} ``` Always returns success. It does not inspect the handler's actual state. ### Remote ```python theme={null} async def health(self) -> dict: response = await self._client.get(f"{self.base_url}/health", headers=self._headers()) response.raise_for_status() return response.json() ``` Calls the remote service's `/health` endpoint and returns the response JSON as-is. If the connection fails, the exception propagates. ### Registry-level health check In `application/services/registry.py`, the system iterates over all executors via `iter_executors()` and calls `health()` on each one. It records the status in the DB as `ExecutorRegistryEntry` rows. *** ## 7. Bootstrap Process **File**: `infrastructure/bootstrap.py` When `bootstrap_runtime(settings=settings)` is called, executors and probes are registered in the following order: ``` 1. register_builtin_cex_executors() → exec.cex.withdrawal_action → exec.cex.withdrawal_observe → exec.cex.deposit_observe 2. register_builtin_observe_executors() → exec.observe.destination_chain_receive → exec.observe.destination_chain_finality → exec.observe.evm_balance → exec.observe.evm_finality → exec.observe.protocol 3. register_protocol_probe("cctp_iris", CctpAttestationProbe(...)) 4. if settings.ccip_enabled: register_builtin_ccip_executors(...) → exec.ccip.send 5. (when EVM endpoints are present) register_builtin_cctp_executors(...) → exec.cctp.burn, exec.cctp.mint register_builtin_evm_executors(...) → exec.evm.erc20_transfer register_chain_receive_probe("evm" / "receive.default", EvmChainReceiveProbe(...)) register_chain_finality_probe("evm" / "finality.default", EvmChainFinalityProbe(...)) 6. _register_hyperliquid_executors(...) # when hl_master_address + signer set → exec.hyperliquid.topup (+ exec.hyperliquid.withdraw) 7. pro_loader.register_pro_executors(...) # sanctioned Pro hook; no-op in OSS → exec.stargate.send, exec.stargate.dst_observe (Pro, when enabled) → lighter_secure_withdraw (Pro, when configured) 8. if settings.gateway_api_base_url: register_builtin_gateway_executors(...) → exec.gateway.{approve,deposit,deposit_observe,intent,mint,mint_observe} 9. if settings.usdt0_enabled: register_builtin_usdt0_executors(...) → exec.usdt0.send ``` CCTP / EVM / Gateway / USDT0 / Hyperliquid / Stargate executors and the EVM probes are **not** registered when `MG_EVM_RPC_ENDPOINTS_JSON` is empty (no EVM endpoints). CEX and observe executors are always registered. CCIP is gated by `ccip_enabled` and registered before the EVM-endpoint guard. Stargate additionally requires `MG_STARGATE_ENABLED` and the Pro package/loader gate; Lighter uses the same package/loader gate. Both are absent from the OSS distribution. *** ## 8. Guide to Adding a New Executor ### 8.1 Local Executor 1. Implement a handler class. At minimum, five methods are required: ```python theme={null} class MyCustomExecutor: async def preflight(self, ctx: ExecutionContext) -> ExecutionResult | None: # Pre-validation. Return None to pass, ExecutionResult to block. return None async def prepare(self, ctx: ExecutionContext) -> ExecutionResult: # Create PreparedAction return ExecutionResult( next_state="SUBMITTING", prepared_action=PreparedAction( prepared_action_id="my-action:123", action_type="my_custom_action", payload_format="json", payload="...", payload_hash="sha256:...", signing_required=False, ), ) async def submit(self, ctx, prepared_action, **kwargs) -> ExecutionResult: # Perform the external call return ExecutionResult(next_state="COMPLETED", provider_refs={...}) async def observe(self, ctx: ExecutionContext) -> ExecutionResult: # Polling logic return ExecutionResult(next_state="COMPLETED") async def recover(self, ctx: ExecutionContext) -> ExecutionResult: # UNKNOWN recovery return ExecutionResult(next_state="COMPLETED") ``` 2. Wrap it with `LocalExecutorAdapter` and register it in the registry: ```python theme={null} from qtg.infrastructure.executors import LocalExecutorAdapter, register_executor register_executor( LocalExecutorAdapter( executor_key="exec.custom.my_action", handler=MyCustomExecutor(), ) ) ``` 3. Add the registration call to `bootstrap_runtime()` in `bootstrap.py`. ### 8.2 Remote Executor 1. Implement six endpoints in the remote service: * `POST /preflight`, `POST /prepare`, `POST /submit`, `POST /observe`, `POST /recover`, `GET /health` 2. Register it in the registry with `RemoteExecutorProxy`: ```python theme={null} from qtg.infrastructure.executors import RemoteExecutorProxy, register_executor register_executor( RemoteExecutorProxy( executor_key="exec.remote.my_action", base_url="https://my-executor.internal:8200", auth_token="secret-token", timeout=15.0, ) ) ``` *** ## Related Documents * [CEX Lane Executors](/reference/executors/cex-lane) * [CCTP Lane Executors](/reference/executors/bridges/cctp-lane) * [Observe Probes](/reference/executors/observe-probes) * [Domain Protocols](/reference/domain/executor-signer-protocols) (ExecutionContext, ExecutionResult, PreparedAction)