Python Executor

​

Responsibilities and boundaries

• Receiving ExecuteCall / ResumeCall / CancelCall / HeartbeatAck and emitting CallWaiting / CallCompleted / CallFailed / Heartbeat over a single long-lived UDS connection.
• Compiling and caching function modules by source digest, isolating module namespaces per task, and fetching source from the Worker by digest when it is missing.
• Using greenlet to suspend synchronous-style user code at SDK call points and resume it after ResumeCall.
• Running cancellation and deadline / budget checks at safe points, and preempting pure-CPU loops with a SIGALRM timer.
• Reporting process metrics (RSS, CPU utilization, cumulative CPU seconds, call counts per state) every heartbeat interval.

• Deciding the call dispatch order or which Executor to pick (that is the Worker dispatcher’s job).
• Accessing BlockDB / node RPC / Function Code View directly. All SDK IO is intercepted and goes out through the Worker.
• Persisting or recovering any call across processes. After a process restart, all context is rebuilt.

• User code runs only in greenlets on the main thread, one at a time; the UDS reader / writer / heartbeat threads are separate and never switch greenlets.
• An ExecContext has at most one pending requestId at any time; a ResumeCall must match callId, the resumeKind corresponding to the state, and requestId all at once, otherwise it is dropped as stale.
• A second ExecuteCall for the same callId is treated as a new attempt: the old greenlet is marked stale and terminated without emitting a terminal state; the new attempt gets a new greenlet.
• Every discarded context emits exactly one terminal message (CallCompleted or CallFailed), except on attempt replacement.
• Compiled artifacts are shared across tasks by digest; exec’d module namespaces are isolated per task_id and released when the task’s last call finishes.

​

Code location

​

Core types and interfaces

class Provider(Protocol):
    def read_req(self, backend: str, operation: str, params: Mapping[str, Any], *, cache_key: Optional[str] = None) -> Any: ...
    def write_req(self, backend: str, operation: str, params: Mapping[str, Any]) -> Any: ...
    def capability_req(self, backend: str, operation: str, params: Mapping[str, Any], *, cache_key: str = "") -> Any: ...
    def subcall(self, function_id: str, args: Any) -> Any: ...
    def sleep(self, seconds: float) -> None: ...

​

User function programming model

def _(p):
    child = function.call("adder", {"left": p["left"], "right": p["right"]})
    total = Decimal(str(child["sum"])) + Decimal("0.50")
    return {
        "child": child,
        "encoded": json.dumps({"sum": child["sum"]}, sort_keys=True, separators=(",", ":")),
        "total": str(total),
        "day": date(2026, 6, 10).isoformat(),
        "stamp": datetime(2026, 6, 10, 9, 30).isoformat(),
    }

from leafage import ChainState

def _(p):
    cs = ChainState(p["chain"], p["block_id"])
    code = cs.get_address_code(p["address"])
    bal = cs.get_address_balance(p["address"])
    return len(code) + bal

​

Data flow / execution flow

​

State and lifecycle

• On resume, the state does not first go back to RUNNABLE: resume_call only writes resume_payload and enqueues with queued=True, and _perform_wait switches directly to RUNNING after switch() returns. Whether a waiting context is already queued is expressed by the queued flag, not by the state.
• A DONE context is removed from _contexts immediately, so inflight_count() is just len(_contexts).
• WAITING_FUNCTION_CODE is used for fetching source by digest. Concurrent misses on the same digest are coalesced locally (singleflight) in SdkBridge._function_code_fetches; only the leader emits CallWaiting(waitKind=function_code).

​

Configuration

​

Extension points

• Connect a new external SDK to the Worker IO channel: write a bridge module (see blockdb_bridge.py / noderpc_bridge.py for reference) that replaces the SDK’s transport seam with calls to provider.read_req / write_req, and install it in ExecutorRuntime.__init__; the backend name must match the backend registered in the Worker-side IO subsystem (see IO access subsystem). If the SDK is gRPC unary, reuse BridgeChannel(provider_getter, backend=..., streaming_methods=frozenset()) directly.
• Add a new SDK entry point for user code: only add a thin wrapper under blockx_sdk that calls get_provider(); do not hold connections in the SDK. For restricted mode, also update the whitelist in blockx_audit/tables.py and the restricted_runtime facade.
• Add a new ResumeCall error type: add a RemoteCallError subclass in sdk_bridge.py and extend _map_resume_error; use register_io_error_kinds for backend-specific mappings.
• Add a heartbeat field or message type: add a constructor in messages.py, a constant in wire_keys.py, a branch in runtime._handle_inbound_message_typed, and update api/uds/types.go in step.
• Change safe points or accounting: all expiry decisions go through ExecContext.deadline_expired_reason and raise_if_deadline_expired; do not write separate checks elsewhere.

​

Testing

uv sync --project python                                    # first time or after uv.lock changes; the Go E2E tests use this .venv too
PYTHONPATH=python uv run --project python python -m unittest discover -s python/tests
PYTHONPATH=python uv run --project python python -m unittest python.tests.test_executor_protocol_contract   # single module

• Runtime: test_executor_runtime_scheduler.py, test_executor_runtime_termination.py, test_executor_runtime_attempt.py, test_executor_budget.py, test_executor_subcall.py, test_executor_oom_hardening.py, test_terminal_bounds.py.
• Protocol: test_executor_protocol_contract.py (ExecuteCall validation, inbound control messages, heartbeat fields), test_uds_codec.py, test_uds_session.py, test_envelope_*.py.
• SDK hooks: test_executor_sdk_hook.py, test_sdk_bridge_provider.py, test_sdk_bridge_capability.py, test_perform_wait_emit_failure.py, test_blockdb_bridge.py, test_noderpc_bridge.py, test_sleep_patch.py.
• Loading and isolation: test_module_registry.py, test_restricted_runtime.py, test_executor_audit_gate.py.
• Context and metrics: test_exec_context_*.py, test_process_metrics_sampler.py, test_logging_setup.py, test_tracing_setup.py, test_debug_capture.py.
• Auditor: test_audit_*.py.

​

Related docs

• docs/specs/python-function-executor.md — the in-process design of the Executor: scheduling, states, attempt / requestId, cancellation and deadlines, resource bounds (the key one).
• docs/specs/worker-executor-connection-and-python-sdk-hook.md §6–§8 — the SDK hook approach, the SdkBridge synchronous contract, the suspend / resume sequence.
• docs/specs/python-executor-process-metrics.md — sampling and degradation of process metrics in heartbeats.
• docs/specs/2026-07-23-execute-call-function-code-reference.md — the protocol for fetching source by digest.
• docs/specs/executor-sandbox-isolation.md — the containerd sandbox spawn mode.
• docs/specs/2026-07-30-python-executor-cpu-optimization.md — the CPU optimization plan.
• docs/specs/python-test-organization.md — the unit test organization standard.
• docs/specs/function-code-python-whitelist.md, docs/specs/function-code-audit-design.md — user code whitelist and auditing.

• Call execution subsystem — Worker-side dispatcher, executor adapter, subfunctions.
• Protocols and interfaces — UDS frame format and message fields.
• Function Code View — where source comes from and blockx_audit.
• IO access subsystem — where CallWaiting(waitKind=io) goes on the Worker side.
• Observability — log fields, metrics, tracing.