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

# Observe Probes

> Observer executor probes — how node status is polled and terminal states detected

# Observe Executors & Probes

> **Source files**
>
> * `src/qtg/infrastructure/executors/observe/__init__.py`
> * `src/qtg/infrastructure/executors/observe/registry.py`
> * `src/qtg/infrastructure/executors/observe/common.py`
> * `src/qtg/infrastructure/executors/observe/destination_chain_receive.py`
> * `src/qtg/infrastructure/executors/observe/destination_chain_finality.py`
> * `src/qtg/infrastructure/executors/observe/protocol.py`
> * `src/qtg/infrastructure/executors/observe/evm.py`
> * `src/qtg/infrastructure/executors/observe/cctp.py`
> * `src/qtg/application/services/observe.py`
> * `src/qtg/infrastructure/bootstrap.py`

***

## 1. Architecture overview

The Observe system has a two-layer **executor + probe** structure.

* **Observe Executor**: generic lifecycle management (preflight, prepare, submit, observe, recover)
* **Probe**: the actual external system polling logic (chain RPC, protocol API, etc.)

Because the executor looks up the probe at runtime via `node_config.probe_key`, the same executor can handle a wide range of chains/protocols.

```mermaid theme={null}
flowchart TB
    subgraph Worker["Observer Worker (observe_active_nodes)"]
        Get["get_executor(binding)"]
        subgraph Exec["Observe Executor (generic)"]
            Lookup["probe_key lookup"]
            Probe["Probe (specific)"]
            Lookup --> Probe
        end
        Get --> Exec
    end
```

### Three Observe Executor families

| Family             | Executor                                  | Probe Protocol       | Decision criterion                                 |
| ------------------ | ----------------------------------------- | -------------------- | -------------------------------------------------- |
| **Chain Receive**  | `DestinationChainReceiveObserveExecutor`  | `ChainReceiveProbe`  | match a receive on the destination chain           |
| **Chain Finality** | `DestinationChainFinalityObserveExecutor` | `ChainFinalityProbe` | destination chain confirmation reached             |
| **Protocol**       | `ProtocolObserveExecutor`                 | `ProtocolProbe`      | protocol-level proof completed (attestation, etc.) |

### Registered Executor Keys

| executor\_key                             | Class                                     |
| ----------------------------------------- | ----------------------------------------- |
| `exec.observe.destination_chain_receive`  | `DestinationChainReceiveObserveExecutor`  |
| `exec.observe.destination_chain_finality` | `DestinationChainFinalityObserveExecutor` |
| `exec.observe.evm_balance`                | `EvmBalanceObserveExecutor`               |
| `exec.observe.evm_finality`               | `EvmFinalityObserveExecutor`              |
| `exec.observe.protocol`                   | `ProtocolObserveExecutor`                 |

`EvmBalanceObserveExecutor` / `EvmFinalityObserveExecutor` are registered unconditionally by `register_builtin_observe_executors()` (alongside the destination-chain and protocol executors). They are EVM-specific observe lanes that read balances / tx finality directly via the runtime RPC client rather than going through the generic chain-probe registry.

***

## 2. Probe Registry

**File**: `observe/registry.py`

Manages three independent registries, each shaped as `dict[str, Probe]`.

```python theme={null}
_CHAIN_RECEIVE_PROBES: dict[str, ChainReceiveProbe] = {}
_CHAIN_FINALITY_PROBES: dict[str, ChainFinalityProbe] = {}
_PROTOCOL_PROBES: dict[str, ProtocolProbe] = {}
```

### Registration functions

| Function                                    | Registration target |
| ------------------------------------------- | ------------------- |
| `register_chain_receive_probe(key, probe)`  | ChainReceiveProbe   |
| `register_chain_finality_probe(key, probe)` | ChainFinalityProbe  |
| `register_protocol_probe(key, probe)`       | ProtocolProbe       |

### Lookup functions

| Function                        | Return type          | When not registered |
| ------------------------------- | -------------------- | ------------------- |
| `get_chain_receive_probe(key)`  | `ChainReceiveProbe`  | `KeyError`          |
| `get_chain_finality_probe(key)` | `ChainFinalityProbe` | `KeyError`          |
| `get_protocol_probe(key)`       | `ProtocolProbe`      | `KeyError`          |

### Initialization functions

| Function                                | Role       |
| --------------------------------------- | ---------- |
| `clear_chain_receive_probe_registry()`  | Test reset |
| `clear_chain_finality_probe_registry()` | Test reset |
| `clear_protocol_probe_registry()`       | Test reset |

### Bootstrap registration (current)

Probes registered in `infrastructure/bootstrap.py`:

```python theme={null}
# always registered
register_protocol_probe("cctp_iris", CctpAttestationProbe(base_url=..., iris_client=...))

# when EVM endpoints are present — each EVM probe is registered under TWO keys
register_chain_receive_probe("evm", receive_probe)
register_chain_receive_probe("receive.default", receive_probe)
register_chain_finality_probe("evm", finality_probe)
register_chain_finality_probe("finality.default", finality_probe)
```

The same `EvmChainReceiveProbe` / `EvmChainFinalityProbe` instance is registered under both the `"evm"` key and a `*.default` alias, so a `node_config.probe_key` of either `"evm"` or `"receive.default"` / `"finality.default"` resolves to the same probe.

Currently registered probes:

| probe\_key           | Protocol/class                  | Registry        |
| -------------------- | ------------------------------- | --------------- |
| `"evm"`              | `EvmChainReceiveProbe`          | chain\_receive  |
| `"receive.default"`  | `EvmChainReceiveProbe` (alias)  | chain\_receive  |
| `"evm"`              | `EvmChainFinalityProbe`         | chain\_finality |
| `"finality.default"` | `EvmChainFinalityProbe` (alias) | chain\_finality |
| `"cctp_iris"`        | `CctpAttestationProbe`          | protocol        |

***

## 3. Probe Protocol Definitions

**File**: `domain/protocols.py`

### 3.1 ChainReceiveProbe

```python theme={null}
@runtime_checkable
class ChainReceiveProbe(Protocol):
    async def check_receive(
        self, context: ExecutionContext, *, match_mode: str, targets: dict[str, Any],
    ) -> ChainReceiveProbeResult: ...
```

```python theme={null}
class ChainReceiveProbeResult(BaseModel):
    matched: bool = False
    provider_state: str | None = None
    provider_refs: dict[str, Any] = Field(default_factory=dict)
    proof: dict[str, Any] = Field(default_factory=dict)
    retry_after_seconds: int | None = None
```

### 3.2 ChainFinalityProbe

```python theme={null}
@runtime_checkable
class ChainFinalityProbe(Protocol):
    async def check_finality(
        self, context: ExecutionContext, *, txid: str, confirmations_required: int,
    ) -> ChainFinalityProbeResult: ...
```

```python theme={null}
class ChainFinalityProbeResult(BaseModel):
    finalized: bool = False
    confirmations: int = 0
    provider_state: str | None = None
    provider_refs: dict[str, Any] = Field(default_factory=dict)
    proof: dict[str, Any] = Field(default_factory=dict)
    retry_after_seconds: int | None = None
```

### 3.3 ProtocolProbe

```python theme={null}
@runtime_checkable
class ProtocolProbe(Protocol):
    async def check_proof(
        self, context: ExecutionContext, *, protocol_ref: str, proof_mode: str | None = None,
    ) -> ProtocolProbeResult: ...
```

```python theme={null}
class ProtocolProbeResult(BaseModel):
    completed: bool = False
    provider_state: str | None = None
    provider_refs: dict[str, Any] = Field(default_factory=dict)
    proof: dict[str, Any] = Field(default_factory=dict)
    generated_artifacts: list[GeneratedArtifact] = Field(default_factory=list)
    retry_after_seconds: int | None = None
```

A ProtocolProbe can return `generated_artifacts`. Used for data that must be persisted, such as attestations.

***

## 4. DestinationChainReceiveObserveExecutor

**File**: `observe/destination_chain_receive.py`

The executor that confirms asset receipt on the destination chain.

### 4.1 preflight

1. Check `match_mode` is present (from node\_config)
2. Extract targets (`resolve_destination_targets`)
3. Required-field validation per match\_mode:
   * `txid` -> txid required
   * `address` -> address required
   * `address_memo` -> address + memo required
4. Look up the probe (`probe_key`) + call `validate_context` (if present)

**Error codes**:

| Error Code                    | Condition                            |
| ----------------------------- | ------------------------------------ |
| `MISSING_MATCH_MODE`          | match\_mode missing                  |
| `MISSING_TXID`                | txid match mode but no txid          |
| `MISSING_DESTINATION_ADDRESS` | address match mode but no address    |
| `MISSING_DESTINATION_MEMO`    | address\_memo match mode but no memo |
| `PROBE_NOT_REGISTERED`        | no probe registered for probe\_key   |

### 4.2 prepare

```python theme={null}
build_observe_action(
    action_type="destination_chain_receive_observe",
    payload={"match_mode": match_mode, "targets": targets},
)
```

### 4.3 submit

Returns `SUBMITTED` immediately. Actual polling happens in observe.

### 4.4 observe

```python theme={null}
probe = get_chain_receive_probe(probe_key(ctx))
result = await probe.check_receive(ctx, match_mode=match_mode, targets=targets)
```

| Probe result                      | next\_state | Additional data                      |
| --------------------------------- | ----------- | ------------------------------------ |
| `result.matched == True`          | `COMPLETED` | provider\_refs, proof, artifacts=\[] |
| `result.matched == False`         | `OBSERVING` | retry\_after\_seconds                |
| `TemporaryMovementError`          | `OBSERVING` | rpc\_retry provider\_state           |
| `FatalMovementError` / `KeyError` | `FAILED`    | error\_code                          |

### 4.5 recover

Re-invokes `observe` directly.

***

## 5. DestinationChainFinalityObserveExecutor

**File**: `observe/destination_chain_finality.py`

The executor that checks whether a transaction's block has reached the confirmation threshold.

### 5.1 preflight

1. Check `txid` is present (from provider\_context or input\_params)
2. Check `confirmations_required` is present (from node\_config)
3. Look up the probe + `validate_context`

**Error codes**:

| Error Code                       | Condition                       |
| -------------------------------- | ------------------------------- |
| `MISSING_TXID`                   | txid missing                    |
| `MISSING_CONFIRMATIONS_REQUIRED` | confirmations\_required missing |
| `PROBE_NOT_REGISTERED`           | probe\_key not registered       |

### 5.2 prepare

```python theme={null}
build_observe_action(
    action_type="destination_chain_finality_observe",
    payload={"txid": txid, "confirmations_required": confirmations_required},
)
```

### 5.3 submit

Returns `SUBMITTED` immediately.

### 5.4 observe

```python theme={null}
probe = get_chain_finality_probe(probe_key(ctx))
result = await probe.check_finality(ctx, txid=txid, confirmations_required=confirmations_required)
```

| Probe result                      | next\_state | Additional data                 |
| --------------------------------- | ----------- | ------------------------------- |
| `result.finalized == True`        | `COMPLETED` | proof (confirmations, required) |
| `result.finalized == False`       | `OBSERVING` | retry\_after\_seconds           |
| `TemporaryMovementError`          | `OBSERVING` | rpc\_retry                      |
| `FatalMovementError` / `KeyError` | `FAILED`    | error\_code                     |

### 5.5 recover

Re-invokes `observe` directly.

***

## 6. ProtocolObserveExecutor

**File**: `observe/protocol.py`

A generic executor that checks protocol-level proofs (attestation, delivery proof, etc.).

### 6.1 preflight

1. Check `protocol_ref` is present (from provider\_context or input\_params)
2. Look up the probe (`probe_key`) + `validate_context`

**Error codes**:

| Error Code             | Condition                 |
| ---------------------- | ------------------------- |
| `MISSING_PROTOCOL_REF` | protocol\_ref missing     |
| `PROBE_NOT_REGISTERED` | probe\_key not registered |

### 6.2 prepare

```python theme={null}
build_observe_action(
    action_type="protocol_observe",
    payload={"protocol_ref": protocol_ref, "proof_mode": proof_mode},
)
```

### 6.3 submit

Returns `SUBMITTED` immediately.

### 6.4 observe

```python theme={null}
probe = get_protocol_probe(probe_key(ctx))
result = await probe.check_proof(ctx, protocol_ref=protocol_ref, proof_mode=proof_mode)
```

| Probe result                      | next\_state | Additional data                             |
| --------------------------------- | ----------- | ------------------------------------------- |
| `result.completed == True`        | `COMPLETED` | provider\_refs, proof, generated\_artifacts |
| `result.completed == False`       | `OBSERVING` | retry\_after\_seconds                       |
| `TemporaryMovementError`          | `OBSERVING` | rpc\_retry                                  |
| `FatalMovementError` / `KeyError` | `FAILED`    | error\_code                                 |

ProtocolObserveExecutor includes `generated_artifacts` in the `ExecutionResult`. The observer worker persists them to the DB.

### 6.5 recover

Re-invokes `observe` directly.

***

## 7. EVM Probes

**File**: `observe/evm.py`

### 7.1 EvmJsonRpcClient

The JSON-RPC client shared by every EVM probe:

```python theme={null}
class EvmJsonRpcClient:
    def __init__(self, *, timeout_seconds: int = 10, client: httpx.AsyncClient | None = None) -> None:
    async def call(self, endpoint: str, method: str, params: list[Any]) -> Any:
```

JSON-RPC 2.0 protocol. Sends `{"jsonrpc": "2.0", "id": 1, "method": ..., "params": ...}` via `POST endpoint`.

**Error classification**:

| Condition                               | Exception                                        | Description             |
| --------------------------------------- | ------------------------------------------------ | ----------------------- |
| timeout / connect error                 | `TemporaryMovementError(RPC_TEMPORARY)`          | retryable               |
| HTTP 429 / 5xx                          | `TemporaryMovementError(RPC_TEMPORARY)`          | retryable               |
| HTTP 4xx                                | `FatalMovementError(RPC_CLIENT_ERROR)`           | fatal                   |
| RPC error code=3 / "execution reverted" | `FatalMovementError(EVM_CALL_REVERTED)`          | contract revert         |
| "nonce too low" / "already known"       | `TemporaryMovementError(TX_BROADCAST_AMBIGUOUS)` | possible TX duplication |
| RPC error -32099 \~ -32000              | `TemporaryMovementError(RPC_TEMPORARY)`          | server error            |
| Other RPC error                         | `FatalMovementError(RPC_CLIENT_ERROR)`           | fatal                   |
| "result" missing                        | `FatalMovementError(RPC_INVALID_RESPONSE)`       | malformed response      |

Default retry wait: `DEFAULT_RETRY_AFTER_SECONDS = 15`

### 7.2 EvmChainReceiveProbe

```python theme={null}
@dataclass(slots=True)
class EvmChainReceiveProbe:
    endpoints: dict[str, str]   # chain_id -> RPC endpoint URL
    rpc_client: EvmJsonRpcClient
```

The probe that confirms transaction receipt on an EVM chain.

#### validate\_context

Required validation:

* `chain_id` -> RPC endpoint present
* `match_mode` = `"txid"` only
* `txid` present
* `destination address` present
* `amount_match` (optional): `"exact"` or `"at_least"` allowed
* `amount_floor_deduction_raw` (optional): non-negative integer; positive values require `amount_match="at_least"`
* For ERC-20 tokens: `token_decimals` required

#### check\_receive

1. Call `eth_getTransactionReceipt(txid)`
   * receipt missing -> `matched=False, provider_state="pending"`
   * `status == 0x0` -> `FatalMovementError(RECEIPT_REVERTED)`

2. Transaction matching:
   * **Native transfer** (`token_contract` unset): check `to` address + `value` via `eth_getTransactionByHash`
   * **ERC-20 transfer** (`token_contract` set): match the `Transfer(from, to, value)` event from receipt logs
     * Transfer topic: `0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef`
     * `topics[2]` (to) = destination address check
     * `log.data` = transfer amount

3. Amount validation (optional):
   * `amount_match = "exact"`: observed == expected
   * `amount_match = "at_least"`: observed >= expected

#### amount\_floor\_deduction\_raw

`amount_floor_deduction_raw` is an at-least-only raw-unit floor deduction. The expected amount is first scaled from the movement amount, then the probe requires:

```
floor = scaled(amount) - amount_floor_deduction_raw
observed >= floor
```

This is used when a protocol can deduct a bounded fee before destination receipt. The key is valid only with `amount_match: "at_least"`; it cannot make an exact-match lane permissive.

| Error code                                 | Condition                                                                           |
| ------------------------------------------ | ----------------------------------------------------------------------------------- |
| `INVALID_AMOUNT_FLOOR_DEDUCTION`           | deduction is bool, non-integer, negative, or greater than or equal to scaled amount |
| `AMOUNT_FLOOR_DEDUCTION_REQUIRES_AT_LEAST` | positive deduction without `amount_match: "at_least"`                               |
| `AMOUNT_MISMATCH`                          | observed amount is below the resulting floor                                        |

#### node\_config requirements

| Key                          | Type        | Required    | Description                                                                                  |
| ---------------------------- | ----------- | ----------- | -------------------------------------------------------------------------------------------- |
| `probe_key`                  | str         | required    | `"evm"`                                                                                      |
| `chain_id`                   | str         | required    | chain ID                                                                                     |
| `match_mode`                 | str         | required    | `"txid"` (currently the only supported value)                                                |
| `token_contract`             | str (0x...) | optional    | ERC-20 token address (omit for native)                                                       |
| `token_decimals`             | int         | conditional | required when token\_contract is set                                                         |
| `amount_match`               | str         | optional    | `"exact"` or `"at_least"`                                                                    |
| `amount_floor_deduction_raw` | int         | optional    | non-negative raw-unit deduction from scaled expected amount; positive only with `"at_least"` |

### 7.3 EvmChainFinalityProbe

```python theme={null}
@dataclass(slots=True)
class EvmChainFinalityProbe:
    endpoints: dict[str, str]   # chain_id -> RPC endpoint URL
    rpc_client: EvmJsonRpcClient
```

The probe that checks the number of block confirmations on an EVM chain.

#### validate\_context

Required validation:

* `chain_id` -> RPC endpoint present
* `txid` present
* `confirmations_required` > 0

#### check\_finality

1. Call `eth_getTransactionReceipt(txid)`
   * receipt missing -> `finalized=False, confirmations=0, provider_state="pending"`
   * `status == 0x0` -> `FatalMovementError(RECEIPT_REVERTED)`

2. Compute confirmations:
   ```
   block_number = receipt.blockNumber
   head = eth_blockNumber()
   confirmations = max(head - block_number + 1, 0)
   finalized = confirmations >= confirmations_required
   ```

3. Return:
   ```python theme={null}
   ChainFinalityProbeResult(
       finalized=True/False,
       confirmations=confirmations,
       provider_state="finalized" | "confirming",
       provider_refs={"txid": txid},
       proof={
           "proof_source": "destination_chain",
           "observed_at": "2026-03-19T...",
           "confirmations": 65,
           "required_confirmations": 65,
       },
   )
   ```

#### node\_config requirements

| Key                      | Type | Required | Description                          |
| ------------------------ | ---- | -------- | ------------------------------------ |
| `probe_key`              | str  | required | `"evm"`                              |
| `chain_id`               | str  | required | chain ID (key of the endpoints dict) |
| `confirmations_required` | int  | required | required confirmation count (> 0)    |

***

## 8. CCTP Attestation Probe

**File**: `observe/cctp.py`

The dedicated probe for CCTP attestation. For details, see [cctp-lane.md](/reference/executors/bridges/cctp-lane#4-cctp-attestation-node-2).

#### node\_config requirements

| Key                | Type | Required | Description                            |
| ------------------ | ---- | -------- | -------------------------------------- |
| `probe_key`        | str  | required | `"cctp_iris"`                          |
| `source_domain_id` | int  | required | CCTP source domain                     |
| `message_index`    | int  | optional | index when there are multiple messages |

***

## 9. Common Helpers

**File**: `observe/common.py`

### 9.1 build\_observe\_action

```python theme={null}
def build_observe_action(*, action_type: str, payload: dict[str, Any]) -> PreparedAction:
```

Creates a `PreparedAction` for observe. Canonical JSON serialization + SHA-256 hash.

* `payload_format = "provider_request"`
* `signing_required = False`
* `prepared_action_id = "{action_type}:{sha256_prefix_12}"`

### 9.2 probe\_key

```python theme={null}
def probe_key(ctx: ExecutionContext) -> str:
    # ctx.node_config["probe_key"], missing -> FatalMovementError(MISSING_PROBE_KEY)
```

### 9.3 match\_mode

```python theme={null}
def match_mode(ctx: ExecutionContext) -> str | None:
    # ctx.node_config["match_mode"], missing -> None
```

### 9.4 confirmations\_required

```python theme={null}
def confirmations_required(ctx: ExecutionContext) -> int | None:
    # ctx.node_config["confirmations_required"], integer parse failure -> None
```

### 9.5 proof\_mode

```python theme={null}
def proof_mode(ctx: ExecutionContext) -> str | None:
    # ctx.node_config["proof_mode"], missing -> None
```

### 9.6 resolve\_destination\_targets

```python theme={null}
def resolve_destination_targets(ctx: ExecutionContext, *, match_mode: str) -> dict[str, str]:
```

Extracts target information per match\_mode:

| match\_mode      | Returns                         | Extraction path (priority)                                                                    |
| ---------------- | ------------------------------- | --------------------------------------------------------------------------------------------- |
| `"txid"`         | `{"txid": ...}`                 | provider\_context.txid -> input\_params.txid                                                  |
| `"address"`      | `{"address": ...}`              | intent.destination.address -> input\_params.address -> provider\_context.destination\_address |
| `"address_memo"` | `{"address": ..., "memo": ...}` | same as above + memo added                                                                    |

### 9.7 resolve\_txid / resolve\_protocol\_ref

```python theme={null}
def resolve_txid(ctx) -> str | None:
    # provider_context.txid -> input_params.txid

def resolve_protocol_ref(ctx) -> str | None:
    # provider_context.protocol_ref -> input_params.protocol_ref
```

### 9.8 Error result helpers

```python theme={null}
def fatal_result(exc, *, default_code, error_detail=None) -> ExecutionResult:
    # converts FatalMovementError/KeyError into a FAILED ExecutionResult

def temporary_result(exc, *, provider_refs) -> ExecutionResult:
    # converts TemporaryMovementError into an OBSERVING ExecutionResult
    # includes retry_after_seconds
```

`temporary_result` is used to keep the observe state while scheduling a retry on transient errors.

***

## 10. Observe results and NodeState transitions

How the observer worker (`application/services/observe.py`) converts an executor's `observe()` result into a NodeState:

| ExecutionResult.next\_state | NodeState transition  | Follow-up behavior                                                    |
| --------------------------- | --------------------- | --------------------------------------------------------------------- |
| `"COMPLETED"`               | `NodeState.COMPLETED` | `advance_after_completion()` — activate next node or complete request |
| `"OBSERVING"`               | `NodeState.OBSERVING` | set `next_observe_at`, continue polling                               |
| `"FAILED"`                  | `NodeState.FAILED`    | `apply_request_state_from_nodes()` — update request state             |
| `"UNKNOWN"`                 | `NodeState.UNKNOWN`   | update request state, set frontier                                    |

### next\_observe\_at computation

```python theme={null}
def _compute_next_observe_at(*, action_type: str, retry_after_seconds: int | None):
    if retry_after_seconds is not None:
        return datetime.now(UTC) + timedelta(seconds=retry_after_seconds)
    interval = DEFAULT_OBSERVE_INTERVALS.get(action_type, 10)
    return datetime.now(UTC) + timedelta(seconds=interval)
```

If the executor returns `retry_after_seconds`, that value is used; otherwise the per-action\_type default applies.

***

## 11. DEFAULT\_OBSERVE\_INTERVALS

**File**: `application/services/observe.py`

```python theme={null}
DEFAULT_OBSERVE_INTERVALS = {
    'cex_withdrawal_status': 5,
    'cex_deposit_status': 10,
    'destination_chain_receive_observe': 10,
    'destination_chain_finality_observe': 15,
    'protocol_observe': 15,
    'cctp_attestation': 15,
    'cctp_mint_status': 5,
    'ccip_delivery': 60,
    'debridge_fulfillment': 30,
}
```

Unmatched action\_types default to **10 seconds**.

| action\_type                         | Default interval | Note                          |
| ------------------------------------ | ---------------- | ----------------------------- |
| `cex_withdrawal_status`              | 5s               | CEX withdrawal state          |
| `cex_deposit_status`                 | 10s              | CEX deposit state             |
| `destination_chain_receive_observe`  | 10s              | on-chain receive check        |
| `destination_chain_finality_observe` | 15s              | block confirmations           |
| `protocol_observe`                   | 15s              | protocol proof (general)      |
| `cctp_attestation`                   | 15s              | CCTP attestation              |
| `cctp_mint_status`                   | 5s               | CCTP mint state               |
| `ccip_delivery`                      | 60s              | CCIP delivery (future)        |
| `debridge_fulfillment`               | 30s              | deBridge fulfillment (future) |

***

## 12. Guide to adding a new Probe

### 12.1 Adding a ChainReceiveProbe — example

```python theme={null}
from qtg.domain import ChainReceiveProbeResult, ExecutionContext

class SolanaChainReceiveProbe:
    def __init__(self, rpc_url: str):
        self.rpc_url = rpc_url

    async def validate_context(self, context: ExecutionContext) -> None:
        # validate match_mode and required parameters
        ...

    async def check_receive(
        self, context: ExecutionContext, *, match_mode: str, targets: dict,
    ) -> ChainReceiveProbeResult:
        # check the transaction via Solana RPC
        ...
        return ChainReceiveProbeResult(
            matched=True,
            provider_state="observed",
            provider_refs={"txid": "..."},
            proof={"proof_source": "destination_chain", ...},
        )
```

Registration:

```python theme={null}
from qtg.infrastructure.executors.observe.registry import register_chain_receive_probe

register_chain_receive_probe("solana", SolanaChainReceiveProbe(rpc_url="..."))
```

node\_config:

```json theme={null}
{
    "probe_key": "solana",
    "match_mode": "txid",
    "chain_id": "solana-mainnet"
}
```

Use the existing `exec.observe.destination_chain_receive` executor as-is and just change `probe_key`.

### 12.2 Adding a ProtocolProbe — example

```python theme={null}
from qtg.domain import ProtocolProbeResult, ExecutionContext

class CcipDeliveryProbe:
    async def check_proof(
        self, context: ExecutionContext, *, protocol_ref: str, proof_mode: str | None,
    ) -> ProtocolProbeResult:
        # check CCIP message delivery state
        ...
        return ProtocolProbeResult(
            completed=True,
            provider_state="delivered",
            provider_refs={"protocol_ref": protocol_ref, "dest_txid": "..."},
            proof={"proof_source": "protocol", ...},
        )
```

Registration:

```python theme={null}
register_protocol_probe("ccip", CcipDeliveryProbe())
```

***

## Related documents

* [Executor Overview](/reference/executors/overview) — executor architecture, registry, local/remote distinction
* [CEX Lane](/reference/executors/cex-lane) — CEX executor details (does not use observe executors)
* [CCTP Lane](/reference/executors/bridges/cctp-lane) — how CCTP uses observe probes