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.

---

2. ExecutorProtocol

@runtime_checkable protocol defined in domain/protocols.py:

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.

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

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].

_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:

{"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

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

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:

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:

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:

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
• CCTP Lane Executors
• Observe Probes
• Domain Protocols (ExecutionContext, ExecutionResult, PreparedAction)