Plan Compilation

How templates are compiled into executable DAG plans

Plan Compilation

Source of truth: src/qtg/compiler/

This document describes the compilation pipeline that converts a movement plan template into an executable request. It covers graph validation, topological ordering, binding resolution, snapshot creation, and deterministic hashing.

---

Table of Contents

1. Overview: what is Plan Compilation
2. Compilation pipeline
3. Graph validation (validate.py)
4. Binding resolution (resolve.py)
5. Compiled Plan Snapshot (snapshot.py)
6. Deterministic Hash (hash.py)
7. Frontier node computation
8. Valid graphs vs invalid graphs
9. Completion Policy validation
10. v3.0 executability decision

---

Overview: what is Plan Compilation

Plan Compilation is the process of combining the abstract node/edge graph defined by a template with the input_params provided by the user, producing an immutable snapshot in which all information required for approval and execution has been finalized.

A compiled plan guarantees the following:

• Graph integrity: DAG (directed acyclic graph) property, no duplicate nodes
• Bindings finalized: the executor/signer to use for each node is decided
• Hash fixed: the plan at approval time and the plan at execution time can be verified as identical
• Topological order: the execution order of nodes is deterministic

---

Compilation pipeline

Public API

Source: src/qtg/compiler/__init__.py

__all__ = [
    "GraphValidationError",
    "GraphValidationResult",
    "build_compiled_snapshot",
    "compute_compiled_plan_hash",
    "compute_plan_graph_hash",
    "current_frontier_node_keys",
    "resolve_bindings",
    "validate_template_graph",
]

Step-by-step call order

Step	Function	Input	Output	Purpose
1	validate_template_graph()	nodes, edges, completion_policy	GraphValidationResult	DAG validation + topological ordering
2	resolve_bindings()	nodes	dict[str, dict]	finalize bindings from executor/signer selectors
3	build_compiled_snapshot()	all parameters	dict[str, Any]	construct the immutable snapshot
4	compute_compiled_plan_hash()	snapshot	str	deterministic SHA-256 hash

---

Graph validation (validate.py)

Source: src/qtg/compiler/validate.py

validate_template_graph()

def validate_template_graph(
    nodes: list[dict[str, Any]],
    edges: list[dict[str, Any]],
    completion_policy: dict[str, Any] | None = None,
) -> GraphValidationResult:

Input:

• nodes — the list of nodes defined by the template. Each node must contain at least node_key, node_kind, action_type, executor_selector, and has_side_effect.
• edges — connections between nodes. Each edge contains from, to, and edge_type, optionally with condition_expr.
• completion_policy — completion policy (optional). The completion_assurance field is required.

Returns: a GraphValidationResult dataclass.

@dataclass(frozen=True)
class GraphValidationResult:
    graph_shape: GraphShape          # LINEAR, BRANCHING, MERGING, SPLIT, HYBRID
    requires_graph_runtime: bool     # if True, cannot execute on v3.0
    executable_in_v3_0: bool         # whether executable on the v3.0 linear runtime
    non_executable_reasons: list[str]  # list of reasons execution is blocked
    topological_order: list[str]     # topologically ordered node keys

Validation rules

Graph validation is performed in the following order.

1. Duplicate node_key check

node_keys = [node["node_key"] for node in nodes]
if len(node_keys) != len(set(node_keys)):
    raise GraphValidationError("duplicate node_key detected")

If two or more nodes share the same node_key, validation fails immediately.

1-A. Dots prohibited in node_key

for node_key in node_keys:
    if "." in node_key:
        raise GraphValidationError("node_key must not contain '.'")

Because $ref:node_key.field syntax splits node and field on the first ., a node_key cannot contain ..

2. Minimum node count check

if not node_map:
    raise GraphValidationError("template must define at least one node")

An empty graph is not allowed.

3. Edge reference integrity

if from_key not in node_map:
    raise GraphValidationError(f"edge references unknown from node: {from_key}")
if to_key not in node_map:
    raise GraphValidationError(f"edge references unknown to node: {to_key}")

Validation fails if an edge’s from/to references a non-existent node.

4. Self-loop prohibition

if from_key == to_key:
    raise GraphValidationError("self-loop edges are not allowed")

A node cannot have an edge to itself.

5. DAG validation (cycle detection)

Topological ordering is performed using Kahn’s algorithm. If the resulting ordering contains fewer nodes than the total node count, a cycle exists.

queue = deque(sorted([key for key, degree in indegree.items() if degree == 0]))
topo: list[str] = []
indegree_working = dict(indegree)
while queue:
    current = queue.popleft()
    topo.append(current)
    for edge in sorted(outbound[current], key=lambda item: (item["to"], item.get("edge_type", ""))):
        child = edge["to"]
        indegree_working[child] -= 1
        if indegree_working[child] == 0:
            queue.append(child)

if len(topo) != len(node_map):
    raise GraphValidationError("template graph must be a DAG")

Deterministic ordering guarantee: nodes with the same indegree are ordered by sorted(), and outbound edges are also ordered by (to, edge_type). This ensures the same topological order is produced for the same input.

6. Config $ref validation

Immediately after topological order generation, each node’s top-level config string values are scanned for $ref:node_key.field.

• A malformed expression raises GraphValidationError.
• A reference to a node missing from the template raises GraphValidationError.
• A reference to a node that is not a DAG ancestor of the current node raises GraphValidationError.

At runtime a synthetic namespaced ref map is built from every completed node’s provider_refs, but those keys are not included in the provider_context that the executor sees.

7. Completion Policy validation

When completion_policy is provided, additional validation is performed (see the Completion Policy validation section below).

Graph Shape classification

After validation passes, the graph shape is classified:

has_branching = any(len(children) > 1 for children in outbound.values())
has_merging = any(len(parents) > 1 for parents in inbound.values())

GraphShape	Condition	Description
LINEAR	no branching, no merging	Single path. A -> B -> C
SPLIT	branching exists, no condition_expr	Unconditional branch (parallel execution)
BRANCHING	branching exists, condition_expr exists	Conditional branch
MERGING	merging exists, no branching	Multiple-input merge
HYBRID	branching and merging both exist	Branch + merge combination

if has_branching and has_merging:
    graph_shape = GraphShape.HYBRID
elif has_merging:
    graph_shape = GraphShape.MERGING
elif has_branching:
    graph_shape = GraphShape.BRANCHING if has_condition_expr else GraphShape.SPLIT
else:
    graph_shape = GraphShape.LINEAR

Topological ordering

The topological_order produced by Kahn’s algorithm provides a deterministic order for node execution:

• Predecessor nodes come first
• Nodes at the same level are ordered alphabetically by node_key
• Edges are also ordered by (to, edge_type)

This order is stored in GraphValidationResult.topological_order as a list[str] (list of node keys).

---

Binding resolution (application.services.binding)

Source: src/qtg/application/services/binding.py

async def resolve_bindings(
    session: AsyncSession,
    nodes: list[dict[str, Any]],
) -> dict[str, dict[str, Any]]:

Current behavior

At movement creation time, executor_selector / signer_selector are resolved to concrete bindings against the DB registry.

Executor selector contract:

• action_type required
• provider optional
• chain_family optional

Signer selector contract:

• chain_family required
• profile optional

Resolution rules:

• active registry rows are candidates
• every non-null selector field must match
• 0 matches -> BindingResolutionError
• 2 or more matches -> AmbiguousBindingError

Output: node_key -> {"executor": concrete binding, "signer": concrete binding | None}

Example:

nodes = [
    {
        "node_key": "withdrawal",
        "executor_selector": {"action_type": "cex_withdrawal", "provider": "upbit"},
        "signer_selector": None,
    },
]

result = await resolve_bindings(session, nodes)
# {
#   "withdrawal": {
#     "executor": {
#       "executor_key": "exec.cex.withdrawal_action",
#       "action_type": "cex_withdrawal",
#       "provider": "upbit",
#     },
#     "signer": None,
#   }
# }

---

Compiled Plan Snapshot (snapshot.py)

Source: src/qtg/compiler/snapshot.py

def build_compiled_snapshot(
    *,
    template_key: str,
    template_version: int,
    completion_policy: dict[str, Any] | None = None,
    nodes: list[dict[str, Any]],
    edges: list[dict[str, Any]],
    input_params: dict[str, Any],
    resolved_bindings: dict[str, Any],
    risk_controls: dict[str, Any],
    callback_config: dict[str, Any] | None = None,
    schema_version: str = "v3.0",
) -> dict[str, Any]:

Snapshot structure

Full schema of the returned dict:

{
    "schema_version": "v3.0",
    "template_key": "cex_transfer",
    "template_version": 1,
    "completion_policy": {
        "completion_assurance": "destination_credited"
    },
    "nodes": [
        {
            "node_key": "withdrawal_action",
            "node_kind": "action",
            "action_type": "cex_withdrawal",
            "executor_selector": {"action_type": "cex_withdrawal", "provider": "upbit"},
            "signer_selector": null,
            "has_side_effect": true,
            "config": {}
        }
    ],
    "edges": [
        {
            "from": "withdrawal_action",
            "to": "withdrawal_observe",
            "edge_type": "on_success"
        }
    ],
    "input_params": {
        "source_exchange": "upbit",
        "destination_exchange": "binance",
        "asset": "XRP",
        "network": "XRP",
        "amount": "100",
        "address": "rXXXX...",
        "memo": "12345"
    },
    "resolved_bindings": {
        "withdrawal_action": {
            "executor": {
                "executor_key": "exec.cex.withdrawal_action",
                "action_type": "cex_withdrawal",
                "provider": "upbit"
            },
            "signer": null
        },
        "withdrawal_observe": {
            "executor": {
                "executor_key": "exec.cex.withdrawal_observe",
                "action_type": "cex_withdrawal_status",
                "provider": "upbit"
            },
            "signer": null
        }
    },
    "risk_controls": {
        "hardcap_usd": 10000,
        "rate_limit_per_hour": 5
    },
    "callback_config": {
        "callback_url": "https://dashboard.example.com/webhook"
    }
}

Field descriptions

Field	Type	Description
schema_version	str	Snapshot schema version. Default "v3.0"
template_key	str	Identifier of the template used
template_version	int	Version of the template used
completion_policy	dict	Completion decision policy. Defaults to empty dict
nodes	list[dict]	Original node definitions from the template
edges	list[dict]	Original edge definitions from the template
input_params	dict	Execution parameters provided by the user/system
resolved_bindings	dict	Result of resolve_bindings(). Maps node_key to executor/signer
risk_controls	dict	Risk control parameters defined on the plan version. Free-form and carried whole into the hash; only max_amount (and min_amount, for route filtering) is read at runtime — the illustrative hardcap_usd / rate_limit_per_hour above are inert. See the v3 endpoints reference
callback_config	dict	Callback configuration. Defaults to empty dict

Immutability guarantee

After creation, this snapshot is not modified. It is persisted to the DB at MovementRequest creation and is referenced throughout approval and execution. The compiled_plan_hash can be used to verify whether it has been tampered with.

---

Deterministic Hash (hash.py)

Source: src/qtg/compiler/hash.py

Two hash functions exist; they serve different purposes.

compute_plan_graph_hash()

Purpose: a template-level graph structure hash. Identical template structures produce identical hashes.

def compute_plan_graph_hash(
    *,
    template_key: str,
    template_version: int,
    nodes: list[dict[str, Any]],
    edges: list[dict[str, Any]],
    approval_policy: dict[str, Any],
    completion_policy: dict[str, Any],
    risk_controls: dict[str, Any],
) -> str:

Fields included in the hash:

Category	Included fields
Metadata	schema_version, template_key, template_version
Node	node_key, node_kind, action_type, executor_selector, signer_selector, config, has_side_effect
Edge	from, to, edge_type, condition_expr
Policy	approval_policy, completion_policy, risk_controls

Characteristics: input_params and resolved_bindings are not included. The same template structure produces the same graph hash regardless of input values.

---

compute_compiled_plan_hash()

Purpose: full hash of a compiled plan for a specific request. Core mechanism for approval integrity verification.

def compute_compiled_plan_hash(snapshot: dict[str, Any]) -> str:

Fields included in the hash:

Category	Included fields
Metadata	schema_version, template_key, template_version
Completion policy	completion_policy
Node	node_key, node_kind, action_type, executor_selector, signer_selector, config, has_side_effect
Edge	from, to, edge_type
Input values	input_params
Bindings	resolved_bindings
Risk	risk_controls

Differences vs compute_plan_graph_hash:

• input_params and resolved_bindings are included — request-specific
• Edge condition_expr is excluded — runtime conditions are already finalized in the compiled plan
• approval_policy is excluded — the approval hash is managed externally

Hash computation method

Both functions follow the same pattern:

hashable = {
    "schema_version": "v3.0",
    "template_key": ...,
    # ... ordered fields
    "nodes": sorted([...], key=lambda node: node["node_key"]),
    "edges": sorted([...], key=lambda edge: (edge["from"], edge["to"], edge["edge_type"])),
}
canonical = json.dumps(hashable, sort_keys=True, separators=(",", ":"), ensure_ascii=False)
return "sha256:" + hashlib.sha256(canonical.encode("utf-8")).hexdigest()

Determinism guarantee mechanism:

1. Node ordering: alphabetical by node_key
2. Edge ordering: composite key (from, to, edge_type)
3. JSON normalization: sort_keys=True — dict keys ordered
4. Fixed separators: separators=(",", ":") — compact JSON with no whitespace
5. Unicode preserved: ensure_ascii=False — non-ASCII characters such as Hangul are preserved
6. UTF-8 encoding: hashing is done at the byte level

Hash format: "sha256:" prefix + 64-character hex digest

Example: "sha256:a1b2c3d4e5f6..."

Why the hash matters

1. Approval integrity: verifies that the plan approved by the approver is the same plan being executed
2. Tamper detection: detects whether the compiled snapshot has been altered in the DB
3. Audit trail: tracks which plan construction was executed via the hash
4. ExecutionContext propagation: compiled_plan_hash is propagated to every executor, allowing the executor to verify it is executing as part of the correct plan

---

Frontier node computation

Source: src/qtg/compiler/validate.py

def current_frontier_node_keys(
    nodes: list[dict[str, Any]],
    edges: list[dict[str, Any]],
) -> list[str]:
    node_keys = [node["node_key"] for node in nodes]
    downstream_nodes = {edge["to"] for edge in edges}
    return [node_key for node_key in node_keys if node_key not in downstream_nodes]

Purpose: find nodes with no predecessor (root nodes) in the graph. At request creation these nodes start in the READY state.

Logic: collect every node that appears as an edge’s to target; nodes not in that set are frontier (root) nodes.

Example:

A -> B -> C
D -> C

frontier = [A, D]   # B, C are downstream

Note: this function is used to compute the initial template-level frontier. Runtime advancement of the current frontier is managed by advance_after_completion().

---

Valid graphs vs invalid graphs

Valid graph examples

1. Linear (executable on v3.0)

withdrawal_action -> withdrawal_observe -> deposit_observe

• shape: LINEAR
• executable_in_v3_0: True
• topological_order: ["withdrawal_action", "withdrawal_observe", "deposit_observe"]

2. Split (parallel, not executable on v3.0)

• shape: SPLIT
• executable_in_v3_0: False
• reason: "graph shape 'split' requires graph runtime"

3. Diamond / Hybrid (not executable on v3.0)

• shape: HYBRID
• executable_in_v3_0: False

Invalid graph examples

1. Cycle

A -> B -> C -> A

• Error: GraphValidationError("template graph must be a DAG")

2. Duplicate node_key

nodes = [
    {"node_key": "step1", ...},
    {"node_key": "step1", ...},  # duplicate!
]

• Error: GraphValidationError("duplicate node_key detected")

3. Reference to a non-existent node

edges = [{"from": "step1", "to": "nonexistent", "edge_type": "on_success"}]

• Error: GraphValidationError("edge references unknown to node: nonexistent")

4. Self-loop

edges = [{"from": "step1", "to": "step1", "edge_type": "on_failure"}]

• Error: GraphValidationError("self-loop edges are not allowed")

5. Empty graph

nodes = []

• Error: GraphValidationError("template must define at least one node")

---

Completion Policy validation

When completion_policy is provided, _validate_completion_policy() performs additional validation.

CompletionAssurance mapping

Depending on the CompletionAssurance value, a terminal node must belong to a specific “family”:

CompletionAssurance	Required terminal node family	Description
provider_completed	source_provider_observe	Confirms source exchange withdrawal completion
destination_observed	destination_chain_receive_observe	Confirms destination chain receive
destination_credited	destination_provider_observe	Confirms destination exchange deposit completion
destination_finalized	destination_chain_finality_observe	Confirms destination chain finality
protocol_finalized	protocol_observe	Confirms protocol-level completion

Node family classification

The mapping table that determines the node family from action_type:

NODE_FAMILY_BY_ACTION_TYPE = {
    "cex_withdrawal_status": "source_provider_observe",
    "source_provider_observe": "source_provider_observe",
    "cex_deposit_status": "destination_provider_observe",
    "destination_provider_observe": "destination_provider_observe",
    "destination_chain_receive_observe": "destination_chain_receive_observe",
    "wallet_receive_observe": "destination_chain_receive_observe",
    "mint_receive_observe": "destination_chain_receive_observe",
    "destination_chain_finality_observe": "destination_chain_finality_observe",
    "chain_finality_observe": "destination_chain_finality_observe",
    "protocol_observe": "protocol_observe",
    "bridge_delivery_observe": "protocol_observe",
    "cctp_attestation": "protocol_observe",
    "source_chain_finality_observe": "source_chain_finality_observe",
}

Validation rules

1. The completion_assurance field must be present.
2. It must be a valid CompletionAssurance enum value.
3. At least one terminal node (a node with no outbound edges) must belong to the required family.
4. Family-specific additional configuration validation:
   • destination_provider_observe / destination_chain_receive_observe: config.match_mode required
   • destination_chain_finality_observe: config.confirmations_required > 0 required

Validation failure examples

# When completion_assurance is "destination_credited" but the
# terminal node is withdrawal_observe (source_provider_observe family):
# -> GraphValidationError("completion_assurance 'destination_credited'
#    requires terminal node family 'destination_provider_observe'")

# When the terminal node has the correct family but config.match_mode is missing:
# -> GraphValidationError("terminal node family 'destination_provider_observe'
#    requires config.match_mode")

---

v3.0 executability decision

The executable_in_v3_0 field on GraphValidationResult indicates whether the current v3.0 runtime can execute this graph.

Conditions that block execution

Condition	Reason message
graph_shape != LINEAR	"graph shape '{shape}' requires graph runtime"
At least one edge with non-null condition_expr	"non-null condition_expr is not executable in v3.0"

Both conditions are appended to non_executable_reasons.

non_executable_reasons: list[str] = []
if graph_shape is not GraphShape.LINEAR:
    non_executable_reasons.append(f"graph shape '{graph_shape.value}' requires graph runtime")
if has_condition_expr:
    non_executable_reasons.append("non-null condition_expr is not executable in v3.0")

requires_graph_runtime = bool(non_executable_reasons)
executable_in_v3_0 = not non_executable_reasons

Summary of executable conditions

To execute on v3.0:

• The graph is LINEAR (single path, no branch/merge)
• Every edge’s condition_expr is None

If these conditions are not satisfied, template registration/validation still succeeds, but actual request creation and execution are only possible in a later version once the graph runtime is introduced.

---

Cross-References

• Executor/Signer protocols: domain/executor-signer-protocols.md
• State transitions: domain/states-and-transitions.md
• Types and enumerations: domain/types-and-enums.md
• CEX executor lane: executors/cex-lane.md
• CCTP executor lane: executors/cctp-lane.md
• DB model: infrastructure/data-model.md