`rfgen.executors`¶

Concrete execution backends for Phase 1 sharded IQ generation. Local execution is synchronous; PySpark is the sole supported distributed backend. Dask and Ray are deferred. Every concrete class subclasses BaseDistributedExecutor and is selected at runtime by config (executor.name: ...). Cloud-specific subclasses live in optional extras and lazy-import their vendor SDK; importing rfgen.executors itself never pulls a cloud SDK. See Reference / Cloud Backends for which subclass ships in which extra.

Warning

Most executor classes on this page remain pre-implementation proposals. The Layer 19 run-control reference is implemented and has a deliberately narrower, in-memory scope. It is not a durable executor or cloud control plane.

Module summary¶

from rfgen.config import GenerationConfig
from rfgen.executors import SparkExecutor, ShardSpec
from rfgen.orchestration import build_shard_fn

cfg = GenerationConfig.model_validate({"run": {"run_id": "demo"}, "storage": {"path": "gs://b/d/v1"}})
cfg_hash = "sha256:demo-config"
dataset_uri = "gs://b/d/v1"
shard_fn = build_shard_fn(cfg)
executor = SparkExecutor.from_config({"app_name": "rfgen-demo", "max_executors": 16})
shards = [ShardSpec.for_shard(shard_index=i, num_samples=4096,
                    config_hash=cfg_hash, output_uri=dataset_uri)
          for i in range(64)]
handle = executor.submit(shard_fn, shards)
result = executor.wait(handle)
print(result.succeeded_shards, result.failed_shards, result.elapsed_s)

Every executor has the same surface: submit(shard_fn, shards) returns an ExecutorRunHandle; wait(handle) blocks and returns a RunResult; cancel(handle) aborts. What differs is the cluster type, credential chain, and how shards are dispatched to workers.

Run-control protocol¶

rfgen.run_control supplies the implemented, strict reference contract for submitting, observing, pausing, resuming, cancelling, and reconciling a run. Its InMemoryRunController uses an injected clock and process-local storage. It makes record shapes, state transitions, idempotency, pagination, and reconciliation executable; it does not submit Spark work, persist state across a process restart, or call a cloud provider. Durable control-plane adapters are deferred to Layer 29.

All V1 records are frozen Pydantic v2 models with strict parsing and no unknown fields. Records that carry a *_sha256 field use a lowercase SHA-256 digest of their RFC 8785 canonical JSON representation, excluding that record’s self-hash field. Timestamps must be UTC.

Public records and enums¶

API	Contract
RunState	Closed lifecycle: `PENDING`, `RUNNING`, `PAUSING`, `PAUSED`, `SUCCEEDED`, `FAILED`, `CANCELLING`, `CANCELLED`. Terminal states have no lease.
RunEventKind	Append-only lifecycle, unit, and reconciliation event kinds. Lifecycle events encode an allowed edge; unit events do not change the run lifecycle.
RetryPolicyV1	Sorted, unique retryable codes; 1 to 100 attempts; finite positive backoff bounds and multiplier at least one.
RunSpecV1	Run UUID, manifest revision, sorted unique unit IDs, retry policy, lease TTL of 30 to 86,400 seconds, in-flight limit of 1 to 10,000, and a self-hash.
RunHandleV1	Immutable run identity, specification hash, hashed idempotency key, controller backend, creation time, and self-hash.
RunStatusV1	Immutable observed lifecycle state, sequence number, unit counts, lease, optional failure, observation time, and self-hash. Counts must sum to the total.
RunEventV1	Ordered audit event with actor, manifest revision, optional unit attempt, provider handle or billing ID, and self-hash. Failure events require a RunErrorV1; other event kinds must not carry one.
RunEventPageV1	Immutable event page bound to a 24-hour snapshot, including a signed opaque continuation token when another page exists.
ReconciliationReportV1	Hash-verified report containing rows sorted by run ID, unit ID, and attempt. Each row records the provider observation, action, and before/after unit state.

Controller contract¶

`class rfgen.run_control.InMemoryRunController`¶

class InMemoryRunController:
    def __init__(
        self,
        *,
        now: Callable[[], datetime] | None = None,
        backend: str = "memory",
    ) -> None: ...

    def submit(
        self, spec: RunSpecV1, *, idempotency_key: str, deadline: datetime
    ) -> RunHandleV1: ...
    def status(self, handle: RunHandleV1, *, deadline: datetime) -> RunStatusV1: ...
    def pause(
        self, handle: RunHandleV1, *, expected_state: Literal[RunState.RUNNING],
        deadline: datetime,
    ) -> RunStatusV1: ...
    def resume(
        self, handle: RunHandleV1, *, expected_state: Literal[RunState.PAUSED],
        deadline: datetime,
    ) -> RunStatusV1: ...
    def cancel(
        self, handle: RunHandleV1, *, expected_state: RunState,
        deadline: datetime,
    ) -> RunStatusV1: ...
    def events(
        self, handle: RunHandleV1, *, after_sequence: int = 0, limit: int = 1000,
        page_token: str | None = None, deadline: datetime,
    ) -> RunEventPageV1: ...
    def reconcile(
        self, handle: RunHandleV1, *, now: datetime, deadline: datetime,
    ) -> ReconciliationReportV1: ...

Every public operation requires a UTC deadline. An expired deadline raises DeadlineExceededError before the operation changes controller state. submit hashes, rather than retains, the idempotency key. Reusing that key with the same specification returns the original handle; reusing it with a different specification raises IdempotencyConflictError. A handle produced by a different controller raises RunHandleError.

pause, resume, and cancel are compare-and-swap operations: callers provide the state they observed, and a stale or disallowed transition raises RunConflictError. A production provider adapter is responsible for driving the transitional PAUSING and CANCELLING states to their terminal counterparts. The reference-only reference_progress and reference_unit_* helpers exist to exercise that provider boundary in tests; they are not an execution backend API.

events creates a 24-hour, per-controller snapshot on the first request. Continuation tokens are bound to the handle, snapshot, limit, issuance and expiry times, canonical payload digest, and a controller-local secret. A foreign, malformed, modified, missing, or expired token raises PaginationTokenError with a machine-readable code. New events never alter an existing snapshot.

reconcile compares the reference unit state with explicit provider observations. It can adopt a provider success, requeue a missing running unit, mark a failed unit, or cancel an orphan. Orphan cancellation crosses an explicit callback boundary configured with set_reference_orphan_canceller; without that callback the operation fails with RunConflictError rather than claiming that a provider cancellation occurred. Reconciliation appends a non-lifecycle-mutating event and returns an immutable report.

Lifecycle and unit rules¶

The permitted lifecycle edges are PENDING to RUNNING or CANCELLING, RUNNING to PAUSING, SUCCEEDED, FAILED, or CANCELLING, PAUSING to PAUSED, FAILED, or CANCELLING, PAUSED to RUNNING or CANCELLING, and CANCELLING to CANCELLED or FAILED. Terminal states cannot transition. A run cannot become SUCCEEDED until every unit has succeeded.

The reference helpers start a unit only while the run is RUNNING, enforce the configured in-flight limit, and require consecutive attempts. A retry is allowed only after a retryable failure whose code is listed in RetryPolicyV1.retryable_codes, and never beyond max_attempts. A unit can be committed once for an attempt; its optional billing ID is retained in the unit event. These helpers model control-protocol evidence, not Spark task execution.

Class index¶

Class	Kind	Notes
BaseDistributedExecutor	abc	ABC; subclass to add a custom backend. Documented in the Abstract base classes section below
SparkExecutor	concrete	BYO Spark cluster; takes a `SparkSession` factory. Cloud-managed Spark variants subclass this. Fully documented below
LocalExecutor	concrete	Synchronous one-host execution. Default for tests, CI, and small runs.
LocalRunner	convenience class	Synchronous single-process API for getting-started and testing; not a `BaseDistributedExecutor`
ManagedSparkServerlessExecutor	concrete	GCP Dataproc Serverless for Spark. Ships in `rfgen[gcp]`. (stub)
EMRExecutor	concrete	AWS EMR-on-EC2 / EMR Serverless. Ships in `rfgen[aws]`. (stub)
SynapseExecutor	concrete	Azure Synapse Spark pools. Ships in `rfgen[azure]`. (stub)
DatabricksExecutor	concrete	Multi-cloud Databricks Jobs. Ships in `rfgen[databricks]`. (stub)
ShardSpec	dataclass	One unit of work passed to `submit`
ShardError	dataclass	Serializable sample-level or shard-level failure record
ShardResult	dataclass	Terminal per-shard worker summary returned by `shard_fn`
ExecutorRunHandle	dataclass	Opaque handle returned by `submit`, consumed by `wait` / `cancel`
RunResult	dataclass	Terminal outcome returned by `wait`
SeedSchedule	dataclass	Deterministic shard/sample seed planner accepted by `build_shard_fn(..., seed_schedule=...)`
`ShardWorkerCollaborators`	dataclass	Advanced testing/alternate-runtime helper for `build_shard_fn(...)`; not part of the day-to-day export surface
build_shard_fn	factory function	Produces the per-shard worker callable passed to `submit`. Defined in `rfgen.orchestration`

For per-cloud setup (service accounts, network, quotas) see Reference / Compute Environment. For cluster selection at runtime see Reference / Cloud Backends § Executor configuration.

ABC: `BaseDistributedExecutor`¶

The executor ABC every backend implements. Defines submit, wait, cancel, and schema. It is part of the executor layer, not of rfgen.orchestration; rfgen.orchestration owns the shard worker factory and orchestration-side value types such as ShardSpec, RunResult, and SeedSchedule. This page is the normative reference for the ABC contract and its adjacent executor surfaces.

`class rfgen.executors.SparkExecutor`¶

class SparkExecutor(BaseDistributedExecutor):
    """BYO Spark distributed executor. Maps shards onto Spark tasks using a
    user-supplied `SparkSession` factory; runs against any Spark cluster
    (local, standalone, YARN, Kubernetes) without depending on a vendor SDK.

    Cloud-managed variants (`ManagedSparkServerlessExecutor`, `EMRExecutor`,
    `SynapseExecutor`, `DatabricksExecutor`) subclass this class and override
    only the `SparkSession` construction and credential plumbing.
    """

    name: str = "spark"

Constructor¶

SparkExecutor(
    *,
    app_name: str = "rfgen",
    session_factory: Callable[[], "SparkSession"] | None = None,
    max_executors: int = 8,
    executor_cores: int = 4,
    executor_memory: str = "8g",
    parallelism_per_executor: int = 1,
    shard_partitioning: Literal["one_per_task", "coalesce", "repartition"] = "one_per_task",
    fail_fast: bool = False,
    poll_interval_s: float = 5.0,
    credentials: BaseCredentialsProvider | None = None,
)

Stateless aside from configuration. The SparkSession is created lazily on first submit; wait and cancel reuse the session held by the ExecutorRunHandle.

Class attributes¶

Attribute	Type	Value	Purpose
`name`	str	`"spark"`	Registry key under the `rfgen.executors` entry-point group

Constructor parameters¶

Name	Type	Default	Description
`app_name`	str	`"rfgen"`	`spark.app.name`; surfaces in the Spark UI and cluster logs
`session_factory`	callable	`None`	Zero-arg factory returning a configured `SparkSession`. If `None`, the executor builds one from the remaining parameters using `SparkSession.builder`
`max_executors`	int	`8`	`spark.dynamicAllocation.maxExecutors`. Caps cluster fan-out
`executor_cores`	int	`4`	`spark.executor.cores`; one shard per core unless `parallelism_per_executor` overrides
`executor_memory`	str	`"8g"`	`spark.executor.memory`. Sized for one in-flight scene plus IQ buffers
`parallelism_per_executor`	int	`1`	Concurrent shards per executor JVM. Above 1 only safe when emitters/channels are GIL-free
`shard_partitioning`	str	`"one_per_task"`	Strategy mapping shards to Spark partitions. `"one_per_task"` is canonical; `"coalesce"` / `"repartition"` reshape the RDD before dispatch
`fail_fast`	bool	`False`	If `True`, the first failed shard cancels the run; otherwise failures accumulate into `RunResult.failed_shards`
`poll_interval_s`	float	`5.0`	How often `wait` polls Spark job status
`credentials`	BaseCredentialsProvider	`None`	Resolves storage credentials for shard outputs. `None` defers to the cluster’s ambient identity (typical on managed Spark)

Method: `submit`¶

def submit(
    self,
    shard_fn: Callable[[ShardSpec], ShardResult],
    shards: list[ShardSpec],
) -> ExecutorRunHandle

Dispatch shards to the Spark cluster, one task per ShardSpec. shard_fn must be picklable (Spark serializes it to workers); the rfgen runtime ships a closure that loads the config, instantiates emitter / channel / store, and writes one shard’s worth of samples.

Parameters¶

Name	Type	Required	Default	Description
`shard_fn`	`Callable[[ShardSpec], ShardResult]`	yes	–	Per-shard worker function. Writes to `shard.output_uri` and returns a terminal ShardResult summary. Must be picklable
`shards`	`list[ShardSpec]`	yes	–	Shard specs. Length determines the RDD partition count under `"one_per_task"`

Returns¶

An ExecutorRunHandle with backend_name="spark" and backend_id set to the Spark application ID (spark.app.id). Pass this to wait or cancel.

Raises¶

ConfigurationError if session_factory is None and the builder cannot resolve a master URL from environment / Spark conf.
RuntimeError if shard_fn is not picklable; surfaces the underlying pickle.PicklingError as the cause.

Example¶

from rfgen.executors import SparkExecutor, ShardSpec
from rfgen.orchestration import build_shard_fn

executor = SparkExecutor(app_name="rfgen-demo", max_executors=16)
shards = [
    ShardSpec.for_shard(
        shard_index=i,
        num_samples=4096,
        config_hash="sha256:abc...",
        output_uri="gs://rf-fm-datasets-synth/demo/v1",
    )
    for i in range(64)
]
handle = executor.submit(build_shard_fn(cfg), shards)
result = executor.wait(handle)
assert not result.failed_shards

Method: `wait`¶

def wait(
    self,
    handle: ExecutorRunHandle,
    timeout: float | None = None,
) -> RunResult

Block until every Spark task associated with handle reaches a terminal state, or until timeout seconds elapse. Polls Spark’s REST API at poll_interval_s; aggregates per-task outcomes into RunResult. Worker callables return serializable ShardResult summaries; executors classify a shard as failed from that returned summary (for example, when ShardResult.error_records carries a terminal ShardFailureError) instead of depending on the worker to raise out of the task.

Returns¶

A RunResult. succeeded_shards lists shard IDs that wrote successfully; failed_shards lists shard IDs that ended in failure; errors carries the flattened ShardError records from those shards; elapsed_s is wall-clock time from submit to terminal.

Raises¶

TimeoutError if timeout elapses before all tasks terminate. The handle remains valid; pass it to cancel or call wait again.

Method: `cancel`¶

def cancel(self, handle: ExecutorRunHandle) -> None

Abort all in-flight tasks for handle via SparkContext.cancelJobGroup. Idempotent: cancelling an already-terminal run is a no-op.

Method: `schema`¶

def schema(self) -> type[BaseModel]

Returns the Pydantic v2 model used by the config validator for the executor: config block. Field names mirror the constructor parameters.

Schema: `SparkExecutorConfig`¶

class SparkExecutorConfig(BaseModel):
    name: Literal["spark"] = "spark"
    app_name: str = "rfgen"
    max_executors: int = 8
    executor_cores: int = 4
    executor_memory: str = "8g"
    parallelism_per_executor: int = 1
    shard_partitioning: Literal["one_per_task", "coalesce", "repartition"] = "one_per_task"
    fail_fast: bool = False
    poll_interval_s: float = 5.0

Returned by SparkExecutor.schema() and consumed by the Hydra config layer.

Notes¶

No vendor SDK. SparkExecutor imports only pyspark; cloud-managed variants subclass it and lazy-import their vendor SDK in __init__.
Idempotency. Worker re-runs of a shard write the same sample_id paths (content-hashed); Spark’s speculative execution and task retries are safe. See the idempotency note on ZarrStore.
Credentials. When credentials is None the cluster’s ambient identity is used (Dataproc service account, EMR instance profile, Databricks passthrough). For BYO Spark, pass an explicit BaseCredentialsProvider.
Determinism. Each ShardSpec.config_hash plus shard_index seeds the per-shard RNG inside shard_fn; the executor itself contributes no randomness.

`class rfgen.executors.LocalExecutor`¶

class LocalExecutor(BaseDistributedExecutor):
    """Synchronous executor for tests, CI smoke runs, and small
    single-host generation."""

    name: str = "local"

submit runs every incomplete shard before returning its terminal handle; wait aggregates those retained results. ExecutorConfig.parallelism is ignored by local execution and controls the requested partition count only for the live PySpark executor. Importing LocalExecutor pulls no third-party dependency.

`class rfgen.orchestration.local.LocalRunner`¶

class LocalRunner:
    """Synchronous single-process convenience runner.

    Runs Phase 1 IQ generation in-process, returning Records directly.
    Not a BaseDistributedExecutor; no shards, no submit/wait. Intended for
    getting-started scripts, integration tests, and notebook exploration.
    For production multi-host generation use SparkExecutor.
    Lives in src/rfgen/orchestration/local.py alongside LocalExecutor.
    """

Kind. Convenience class. NOT a subclass of BaseDistributedExecutor.

Module. rfgen.orchestration.local (src/rfgen/orchestration/local.py).

Constructor¶

LocalRunner(cfg: GenerationConfig)

Name	Type	Description
`cfg`	GenerationConfig	Fully validated generation config. Use LocalRunner.build_config to construct one from a preset, or pass a manually validated config directly.

Class method: `build_config`¶

@classmethod
def build_config(
    cls,
    preset: str,
    overrides: dict[str, object] | None = None,
) -> GenerationConfig

Build a validated GenerationConfig from a named preset and an optional dict of Hydra-dotted-path overrides.

Parameters¶

Name	Type	Description
`preset`	str	Name of a shipped preset (see Scenario Presets), e.g. `"narrowband_classifier_baseline"`.
`overrides`	`dict[str, object]` \| None	Hydra dotted-path overrides applied on top of the preset defaults, e.g. `{"run.num_samples": 5, "storage.path": "./out"}`. Keys follow the same convention as `+key=value` on the Hydra CLI.

Returns¶

A fully merged, validated GenerationConfig. Raises pydantic.ValidationError if the merged result is invalid.

Method: `run`¶

def run(self) -> Iterator[Record]

Generate all samples defined by cfg.run.num_samples synchronously in the calling process. Yields one Record per scene.

Returns¶

An Iterator[Record]. Callers can collect all records with list(runner.run()) or iterate lazily to bound memory. The generator respects cfg.storage.path only for manifest writing; the Record objects yielded in-process are fully assembled (IQ + labels + metadata). IQ is not written to disk unless cfg.storage.write_mode is set.

Raises¶

pydantic.ValidationError if config constraints are violated at runtime.
SceneError if the scene composer receives an invalid runtime contract, such as a geometry mismatch or a time-placement strategy that returns no start samples for a sampled emitter.

Example¶

from rfgen.orchestration.local import LocalRunner

cfg = LocalRunner.build_config(
    preset="narrowband_classifier_baseline",
    overrides={"run.num_samples": 5, "storage.path": "./out/demo"},
)
runner = LocalRunner(cfg)
records = list(runner.run())         # returns list[Record]

record = records[0]
print(record.iq.shape)               # torch.Size([2, 1_000_000])
print(record.scene.scene_geometry_hash)
print(len(record.bboxes))            # number of bounding boxes

Notes¶

Not sharded. LocalRunner runs all num_samples records sequentially in the calling process. For parallel execution use SparkExecutor with build_shard_fn.
RNG is deterministic. The run uses cfg.run.seed; the same config and seed reproduce the same records.
Phase 2 not included. run() is Phase 1 only. To annotate the generated records, run the annotation pipeline separately against the store path.

`class rfgen.executors.ManagedSparkServerlessExecutor`¶

class ManagedSparkServerlessExecutor(SparkExecutor):
    """GCP Dataproc Serverless for Spark. Subclasses `SparkExecutor`;
    constructs the `SparkSession` against a Dataproc Serverless batch and
    plugs in `GoogleADCProvider` credentials."""

    name: str = "gcp_serverless_spark"

Constructor adds project_id, region, service_account, machine_type, network_uri, subnetwork_uri. Ships in rfgen[gcp]; lazy-imports google.cloud.dataproc_v1 on first construction. (stub.)

`class rfgen.executors.EMRExecutor`¶

class EMRExecutor(SparkExecutor):
    """AWS EMR-on-EC2 (cluster_id) or EMR Serverless (application_id).
    Subclasses `SparkExecutor`; submits step-runs against an existing EMR
    cluster using `boto3`."""

    name: str = "emr"

Constructor adds cluster_id or application_id, release_label, instance_type, instance_role_arn, log_uri. Ships in rfgen[aws]; lazy-imports boto3. (stub.)

`class rfgen.executors.SynapseExecutor`¶

class SynapseExecutor(SparkExecutor):
    """Azure Synapse Spark pool executor. Subclasses `SparkExecutor`;
    submits Spark batch jobs against a named Synapse pool."""

    name: str = "synapse"

Constructor adds workspace_name, spark_pool_name, tenant_id, subscription_id, resource_group. Ships in rfgen[azure]; lazy-imports azure.synapse.spark. (stub.)

`class rfgen.executors.DatabricksExecutor`¶

class DatabricksExecutor(SparkExecutor):
    """Multi-cloud Databricks executor. Subclasses `SparkExecutor`; submits
    one-time Jobs Runs against an existing Databricks workspace and reuses
    its job clusters."""

    name: str = "databricks"

Constructor adds host, token_env, cluster_id or job_cluster_spec, notebook_path or python_file_uri. Ships in rfgen[databricks]; lazy-imports databricks.sdk. Works on AWS, Azure, and GCP Databricks deployments. (stub.)

`class rfgen.executors.ShardSpec`¶

@dataclass(frozen=True)
class ShardSpec:
    shard_id: str
    shard_index: int
    num_samples: int
    config_hash: str
    output_uri: str

One unit of work passed into submit. Frozen and hash-stable so the same spec can be retried byte-identically. shard_id must be the canonical content-addressed id for (config_hash, shard_index); use ShardSpec.for_shard(...) to derive it.

Field	Type	Description
`shard_id`	str	Canonical content-addressed shard identifier, derived from `config_hash` plus `shard_index`. Used in logs and RunResult
`shard_index`	int	Zero-based index. Combined with `config_hash` to seed the per-shard RNG
`num_samples`	int	Number of Record records this shard must produce
`config_hash`	str	`"sha256:..."` digest of the resolved Hydra config; pinned for reproducibility
`output_uri`	str	Where the shard’s records land. URL scheme picks the storage filesystem (`file://`, `gs://`, `s3://`, `az://`)

Method: `canonical_shard_id`¶

@classmethod
def canonical_shard_id(
    cls,
    *,
    config_hash: str,
    shard_index: int,
) -> str: ...

Returns "sha256:<lowercase hex digest>" for the canonical CBOR encoding of {"config_hash": config_hash, "shard_index": shard_index}. The same two inputs always return the same ID; changing either input changes the hash input.

Parameter	Type	Contract
`config_hash`	str	Required non-empty string. Leading and trailing whitespace is stripped before hashing. The method does not parse or verify a `sha256:` prefix.
`shard_index`	int	Zero-based integer greater than or equal to `0`; booleans and non-integers are rejected.

Returns str. Invalid parameters raise ConfigError naming ShardSpec.config_hash or ShardSpec.shard_index.

Method: `for_shard`¶

@classmethod
def for_shard(
    cls,
    *,
    shard_index: int,
    num_samples: int,
    config_hash: str,
    output_uri: str,
) -> ShardSpec: ...

Calls canonical_shard_id(…) first, then constructs and returns a fully validated ShardSpec with the resulting value in shard_id.

Parameter	Type	Contract
`shard_index`	int	Zero-based integer greater than or equal to `0`; booleans and non-integers are rejected.
`num_samples`	int	Number of records requested from this shard; must be at least `1`, with booleans and non-integers rejected.
`config_hash`	str	Required non-empty string used by `canonical_shard_id(...)`; trimmed during validation.
`output_uri`	str	Required non-empty destination URI or path; trimmed during validation.

Returns ShardSpec. Invalid fields raise ConfigError naming the corresponding ShardSpec field. Direct ShardSpec(...) construction additionally raises ConfigError when a caller-supplied shard_id does not equal the value returned by canonical_shard_id(...) for the same config_hash and shard_index.

Mirrored from the ABC definition; re-exported here for convenience.

`class rfgen.orchestration.ShardError`¶

@dataclass(frozen=True)
class ShardError:
    sample_idx: int
    error_class: str
    message: str
    traceback: str

Immutable failure record carried by ShardResult and flattened into RunResult. It stores strings and integers only, so a distributed executor can pickle or otherwise serialize it without preserving the original exception object.

Field	Type	Valid values	Description
`sample_idx`	int	`-1` or greater; booleans rejected	Zero-based failed sample index. `-1` identifies a shard-level failure that is not attributable to one sample, including storage, channel setup, RT asset preparation, or a shard failure-threshold record.
`error_class`	str	Non-empty after trimming	Exception class name, such as `ValueError`, `StorageError`, or `ShardFailureError`.
`message`	str	Non-empty after trimming	Human-readable exception message.
`traceback`	str	Non-empty after trimming	Captured traceback text. Synthetic shard-level errors use a one-line `"<ErrorClass>: <message>"` value when no active traceback exists.

Construction trims all three string fields. Invalid field types, a sample_idx below -1, bool/float/string substitutes for the integer index, or an empty string raise ConfigError naming the invalid field.

`class rfgen.orchestration.ShardResult`¶

@dataclass(frozen=True)
class ShardResult:
    shard_id: str
    succeeded_samples: int
    failed_samples: int
    elapsed_s: float
    error_records: tuple[ShardError, ...] = ()

Terminal summary returned by one shard worker. Executors aggregate these per-shard outcomes into RunResult.

A shard has two separately stored parts: its records and a small shard-complete flag. The flag is the authoritative statement that every requested record finished. A failed attempt can therefore leave usable records behind without the flag; the next run sees the missing flag and tries the shard again.

Here, commit means publishing that flag after all records finish. It does not promise database-style rollback of records written earlier in the attempt.

Each record has an ID derived from its content, so a deterministic retry regenerates the same ID. Writing identical content under an ID that already exists is an idempotent replay: the store treats it as a no-op rather than creating a duplicate.

Field	Type	Description
`shard_id`	str	Canonical shard identifier matching the submitted ShardSpec
`succeeded_samples`	int	Count of records that completed scene composition and labeling during this attempt. A completed-shard skip reports `ShardSpec.num_samples`.
`failed_samples`	int	Count of caught sample-local composition or labeling failures. A shard-level storage, channel, or RT asset-preparation failure reports `ShardSpec.num_samples` instead.
`elapsed_s`	float	Worker wall-clock seconds for this shard attempt
`error_records`	`tuple[ShardError, ...]`	Terminal shard-level and sample-level failures captured without raising out of the worker

For a normal attempt, the worker increments succeeded_samples immediately before yielding each labeled record to StoreHandle.write_shard(…). The count therefore describes generated and labeled records, not records guaranteed to remain stored when a later sample failure aborts the shard commit. A backend that stages the entire shard may discard every yielded record, while a backend that writes records individually may retain them without publishing the shard-complete flag.

With RunConfig.fail_fast = False, sample iteration attempts every requested sample, so succeeded_samples + failed_samples == ShardSpec.num_samples. With fail_fast = True, iteration stops after the first sample-local failure, so succeeded_samples + failed_samples <= ShardSpec.num_samples; a strict inequality means the remaining samples were not attempted. Shard-level storage, channel, and RT asset-preparation failures normalize the result to succeeded_samples = 0 and failed_samples = ShardSpec.num_samples.

Defined in rfgen.orchestration and consumed by BaseDistributedExecutor.submit() implementations through the Callable[[ShardSpec], ShardResult] worker contract documented on this page.

`class rfgen.executors.ExecutorRunHandle`¶

@dataclass(frozen=True)
class ExecutorRunHandle:
    backend_name: str
    backend_id: str

Opaque handle returned by submit and consumed by wait / cancel.

Field	Type	Description
`backend_name`	str	Matches the executor’s `name` class attribute, e.g. `"spark"`, `"local"`
`backend_id`	str	Vendor-native Spark application or managed-service job identifier.

Mirrored from the ABC definition.

`class rfgen.executors.RunResult`¶

@dataclass(frozen=True)
class RunResult:
    succeeded_shards: tuple[str, ...]
    failed_shards: tuple[str, ...]
    elapsed_s: float
    errors: tuple[ShardError, ...]

Terminal outcome returned by wait.

Field	Type	Description
`succeeded_shards`	`tuple[str, ...]`	Shard IDs whose `shard_fn` returned a terminal success summary
`failed_shards`	`tuple[str, ...]`	Shard IDs whose `shard_fn` returned one or more terminal failures
`elapsed_s`	float	Wall-clock seconds from `submit` to terminal
`errors`	`tuple[ShardError, ...]`	Flattened shard-level and sample-level error records, in arrival order across the run

Mirrored from the ABC definition.

Abstract base classes¶

`BaseDistributedExecutor`¶

Phase 1 sharded IQ generation. Execution is synchronous locally or distributed through PySpark and its managed-service variants.

from dataclasses import dataclass

@dataclass(frozen=True)
class ShardSpec:
    shard_id: str
    shard_index: int
    num_samples: int
    config_hash: str
    output_uri: str

@dataclass(frozen=True)
class ExecutorRunHandle:
    backend_name: str
    backend_id: str       # e.g. Spark or managed-service batch ID

@dataclass(frozen=True)
class RunResult:
    succeeded_shards: tuple[str, ...]
    failed_shards: tuple[str, ...]
    elapsed_s: float
    errors: tuple[ShardError, ...]


class BaseDistributedExecutor(ABC):
    """Phase 1 sharded IQ generation orchestrator."""

    name: str             # registry key: "local", "spark", "gcp_serverless_spark", ...

    @abstractmethod
    def submit(
        self,
        shard_fn: Callable[[ShardSpec], ShardResult],
        shards: list[ShardSpec],
    ) -> ExecutorRunHandle: ...

    @abstractmethod
    def wait(
        self,
        handle: ExecutorRunHandle,
        timeout: float | None = None,
    ) -> RunResult: ...

    @abstractmethod
    def cancel(self, handle: ExecutorRunHandle) -> None: ...

    @abstractmethod
    def schema(self) -> type[BaseModel]:
        """Pydantic config model accepted by this executor's constructor."""

`submit()`¶

Abstract method on BaseDistributedExecutor. Dispatch a list of ShardSpec to the cluster, returning an ExecutorRunHandle.

`wait()`¶

Abstract method on BaseDistributedExecutor. Block until every shard reaches a terminal state (or timeout elapses); returns a RunResult.

`cancel()`¶

Abstract method on BaseDistributedExecutor. Abort all in-flight tasks for the supplied handle. Idempotent.

`schema()`¶

Abstract method on BaseDistributedExecutor. Returns the Pydantic config model accepted by this executor.

Implementations:

Subclass	Cloud	Extra
LocalExecutor	none	core
SparkExecutor	none (any Spark cluster)	core

`class rfgen.rng.SeedSchedule`¶

from rfgen.rng import SeedSchedule

class SeedSchedule:
    global_seed: int
    num_shards: int
    samples_per_shard: int

    @classmethod
    def plan(
        cls,
        global_seed: int,
        num_shards: int,
        samples_per_shard: int,
    ) -> SeedSchedule: ...

    def sample_seed(
        self,
        shard_index: int,
        sample_idx: int,
        *,
        stage: str | int = 0,
        emitter_idx: int = 0,
        rx_idx: int = 0,
    ) -> int: ...

    def shard_sample_seed(
        self,
        shard_id: str,
        sample_idx: int,
        *,
        stage: str | int = 0,
        emitter_idx: int = 0,
        rx_idx: int = 0,
    ) -> int: ...

Module. rfgen.rng (src/rfgen/rng.py). Re-exported from rfgen.orchestration for backward compatibility with existing executor and test code.

Deterministic seed planner for Phase 1 shard execution. Executors rarely construct one directly; the main user-facing seam is the optional seed_schedule= override on build_shard_fn.

Constructor surface¶

Prefer the validated class method:

schedule = SeedSchedule.plan(
    global_seed=1337,
    num_shards=64,
    samples_per_shard=4096,
)

SeedSchedule.plan(...) validates that all three dimensions are real integers and that num_shards > 0 and samples_per_shard > 0.

Fields¶

Name	Type	Description
`global_seed`	`int`	Run-level root seed threaded into seed_for
`num_shards`	`int`	Finite shard count for this run
`samples_per_shard`	`int`	Planned samples per shard, used for index validation and the default per-shard memoization path

Methods¶

Method	Returns	Description
`plan(global_seed, num_shards, samples_per_shard)`	SeedSchedule	Validated constructor used by orchestration and tests
`sample_seed(shard_index, sample_idx, *, stage=0, emitter_idx=0, rx_idx=0)`	`int`	Index-based seed lookup for a synthetic shard id (`shard-000123`)
`shard_sample_seed(shard_id, sample_idx, *, stage=0, emitter_idx=0, rx_idx=0)`	`int`	Seed lookup bound to an explicit canonical shard id string

Notes¶

The default worker path memoizes each shard’s (emitter_idx=0, rx_idx=0, stage=0) per-sample seeds on first use so repeated sample_idx lookups do not rebuild a fresh NumPy SeedSequence chain every time.
The default worker path keeps only the most recently used shard’s (emitter_idx=0, rx_idx=0, stage=0) per-sample seeds memoized at a time, so the hot loop stays amortized without letting long-lived workers retain one full seed tuple per historical shard.
The memoized cache is worker-local runtime state. Pickle round-trips clear it so distributed executors do not ship warmed per-shard seed tuples between processes.
build_shard_fn(...) validates any explicit schedule override against the run-derived shard geometry before returning the worker callable.

Advanced Testing Helper: `ShardWorkerCollaborators`¶

from rfgen.orchestration import ShardWorkerCollaborators
from rfgen.emitters.base import EmitterLike

@dataclass(frozen=True)
class ShardWorkerCollaborators:
    store_factory: Callable[[GenerationConfig], BaseStore] | None = None
    channel_factory: Callable[[GenerationConfig], BaseChannel | ChannelPipeline] | None = None
    composer_factory: Callable[
        [GenerationConfig, BaseChannel | ChannelPipeline],
        BaseSceneComposer,
    ] | None = None
    labeler_factory: Callable[[GenerationConfig], BaseLabeler] | None = None
    emitter_pool_factory: Callable[[GenerationConfig], Mapping[str, EmitterLike]] | None = None

Optional override bundle for tests and alternate runtimes that need a custom worker-local store, channel, composer, labeler, or emitter pool. The default build_shard_fn(cfg) call shape stays narrow; day-to-day callers should ignore this helper even though it is exported by from rfgen.orchestration import *.

Each factory is optional. When omitted, build_shard_fn(...) uses the normal framework path for that collaborator. When provided, the factory must be pickleable and have the runtime call shape that orchestration uses:

store_factory(cfg) returning a BaseStore
channel_factory(cfg) returning a BaseChannel or ChannelPipeline
composer_factory(cfg, channel) returning a BaseSceneComposer; its build(...) method accepts the validated scene config, emitter pool, channel, and RNG, and returns a Signal
labeler_factory(cfg) returning a BaseLabeler
emitter_pool_factory(cfg) returning Mapping[str, EmitterLike]

Advanced alternate-device runtimes can also implement the public structural protocol below on either the resolved channel or composer:

from typing import Protocol, runtime_checkable

from rfgen.orchestration import SupportsSampleRngDevice

@runtime_checkable
class SupportsSampleRngDevice(Protocol):
    def sample_rng_device(self) -> torch.device | str | None: ...

Orchestration checks the composer first. If the composer hook returns a valid non-None device, that device wins and the channel hook is not consulted. The channel hook is used only when the composer omits the hook or returns None. Without either explicit hook, orchestration keeps the sample RNG on CPU.

build_shard_fn(...) validates the factory signatures at construction time. The worker validates collaborator method signatures, the channel return type, and the opened store-handle protocol before it runs any shard records.

`build_shard_fn`¶

from rfgen.orchestration import build_shard_fn

def build_shard_fn(
    cfg: GenerationConfig,
    *,
    seed_schedule: SeedSchedule | None = None,
    collaborators: ShardWorkerCollaborators | None = None,
) -> Callable[[ShardSpec], ShardResult]: ...

Module. rfgen.orchestration (defined in src/rfgen/orchestration.py).

Factory that produces the per-shard worker callable passed to BaseDistributedExecutor.submit(). The returned callable is self-contained: it holds a closed-over copy of the resolved config and carries everything a worker process needs to generate one shard without any other context.

Parameters¶

Name	Type	Description
`cfg`	GenerationConfig	Fully resolved and validated generation config. Must be fully merged (all Hydra defaults applied, all validators passed) before this call. The factory snapshots it with `GenerationConfig.model_copy(deep=True)` before closing over it, so changes to `cfg` after `build_shard_fn` returns have no effect.
`seed_schedule`	SeedSchedule \| `None`	Optional explicit shard seed schedule. When omitted, the factory derives one from `cfg.run.seed`, `cfg.run.num_samples`, and `cfg.run.shard_size` as `SeedSchedule.plan(global_seed=cfg.run.seed, num_shards=ceil(num_samples / shard_size), samples_per_shard=shard_size)`. `SeedSchedule` validates the shard/sample bounds and resolves seeds through seed_for on demand; the default worker path memoizes only the most recently used shard’s `(emitter_idx=0, rx_idx=0, stage=0)` sample seeds, which keeps repeated per-sample lookups hot without retaining one tuple per historical shard. Executors may still pass an explicit schedule override, but `build_shard_fn(...)` validates at factory time that the override’s `num_shards` matches `ceil(cfg.run.num_samples / cfg.run.shard_size)` and its `samples_per_shard` matches `cfg.run.shard_size`.
`collaborators`	`ShardWorkerCollaborators` \| `None`	Advanced testing/alternate-runtime escape hatch. When omitted, `build_shard_fn(...)` uses the default storage, channel, composer, labeler, and emitter-pool construction paths.

Returns¶

A Callable[[ShardSpec], ShardResult] with the following contract:

Input: one ShardSpec.
Output: one ShardResult. Records are still written to shard.output_uri as a side effect.
Picklable: the returned callable must survive pickle.dumps / pickle.loads, which Spark uses to ship it to executor JVMs. The snapshotted Pydantic config and the seed schedule are carried directly inside the worker object. Any collaborator factory inside ShardWorkerCollaborators must itself be pickleable. Worker-local runtime state such as prepared RT assets, collaborator instances, and the asset cache root is recomputed after unpickling, so deserialized workers do not reuse the sender process’s local temp paths.
Validated overrides: collaborator-factory signatures and pickleability are checked at build_shard_fn(...) construction time. On every shard, the worker validates the store’s bound dataset_uri() contract before probing has_shard(...), so a divergent ShardSpec.output_uri fails even on a completed-shard rerun. The completed-shard fast path still avoids instantiating heavy non-store collaborators, so actual non-store collaborator-object validation happens only on an incomplete shard, where the worker validates collaborator method signatures, sample-RNG-device hooks, and the opened store-handle protocol before any shard records run. Invalid overrides fail with ConfigError naming the offending collaborator field.
Bound dataset URI wins: when the resolved store exposes dataset_uri(), ShardSpec.output_uri must match that bound dataset URI exactly before the shard-complete probe runs. Orchestration then opens StoreMode.APPEND on the bound URI, so the shard-complete probe and the append target cannot drift to different datasets.

What the returned callable does¶

def _shard_fn(spec: ShardSpec) -> ShardResult:
    # 1. Read the already-validated config snapshot from closure
    cfg = snapshotted_cfg
    schedule = seed_schedule or SeedSchedule.plan(
        global_seed=cfg.run.seed,
        num_shards=ceil(cfg.run.num_samples / cfg.run.shard_size),
        samples_per_shard=cfg.run.shard_size,
    )

    # 2. Resolve the store, validate the bound dataset URI, then probe the shard-complete marker
    store = cached_store or build_store(cfg.storage)
    append_uri = cached_bound_dataset_uri or store.dataset_uri() or spec.output_uri
    if append_uri != spec.output_uri:
        raise ConfigError("ShardSpec.output_uri must match the bound store dataset URI.")

    if store.has_shard(spec.shard_id):
        return ShardResult(
            shard_id=spec.shard_id,
            succeeded_samples=spec.num_samples,
            failed_samples=0,
            elapsed_s=elapsed_s,
        )

    # 3. Resolve and validate any remaining collaborators only for incomplete shards
    channel = cached_channel or build_channel_pipeline(cfg.channel)
    composer = cached_composer or build_composer(cfg, channel=channel)
    labeler = cached_labeler or build_labeler(cfg)
    emitter_pool = cached_emitter_pool or build_emitter_pool(cfg)
    validate_composer_signature(composer.build)
    validate_labeler_signature(labeler.label)
    validate_emitter_pool(emitter_pool)
    sample_rng = cached_sample_rng or new_sample_rng(composer=composer, channel=channel)

    succeeded = 0
    errors = []

    def records():
        nonlocal succeeded
        for sample_idx in range(spec.num_samples):
            sample_rng.manual_seed(
                schedule.shard_sample_seed(spec.shard_id, sample_idx)
            )
            try:
                scene_signal = composer.build(
                    scene_cfg=cfg.scene,
                    emitter_pool=emitter_pool,
                    channel=channel,
                    rng=sample_rng,
                )
                labels = labeler.label(
                    iq=scene_signal.iq,
                    emitters=tuple(
                        component.metadata
                        for component in scene_signal.component_signals
                    ),
                    scene=scene_signal.metadata,
                )
            except Exception as exc:
                errors.append(ShardError.from_exception(spec, sample_idx, exc))
                if cfg.run.fail_fast:
                    break
                continue
            succeeded += 1
            yield labels

        if errors:
            # Internal control flow: stop write_shard before it can publish its
            # backend-specific shard-complete marker.
            raise AbortShardCommit

    with store.open(append_uri, mode=StoreMode.APPEND) as handle:
        try:
            handle.write_shard(spec.shard_id, records())
        except AbortShardCommit:
            pass

    failed = len(errors)
    if failed / spec.num_samples >= cfg.run.shard_failure_threshold:
        errors.append(shard_failure_error(spec.shard_id))
    return ShardResult(
        shard_id=spec.shard_id,
        succeeded_samples=succeeded,
        failed_samples=failed,
        elapsed_s=elapsed_s,
        error_records=tuple(errors),
    )

AbortShardCommit, ShardError.from_exception(...), and shard_failure_error(...) above are descriptive pseudocode. The implementation uses private helpers for the same control flow and constructs the public frozen result records explicitly. The labeler returns the complete Record; there is no separate record-assembly call after labeling.

Notes¶

Streaming commit boundary. The worker passes a lazy record iterator to StoreHandle.write_shard(…). A backend may persist each yielded record immediately, but it must publish its shard-complete marker only after exhausting the iterator normally. If any sample fails, the iterator raises the private abort after its last successful record; write_shard(...) therefore does not return and must not publish the marker. The worker catches only that private abort and still returns the accumulated sample results.
Idempotency. The worker writes through the store’s content-addressed sample IDs. A shard retry replays the same successful records to the same sample_id values, and store-owned skip-if-exists handling makes those sample writes a no-op. Because the failed attempt has no shard-complete marker, has_shard(...) remains false and the retry enters the generation path instead of taking the completed-shard fast path.
No Phase 2. The callable is Phase 1 only: it does not call any inference backend or annotator. Phase 2 is a separate job that appends to the store after Phase 1 completes.
Error handling. Exceptions from scene composition or labeling are sample-local: each becomes one ShardError, and generation continues unless cfg.run.fail_fast is true. Any sample-local failure aborts the storage shard commit before a shard-complete marker is published. Storage or RT asset-preparation failures are returned as shard-level error records with sample_idx = -1. After sample iteration stops, the worker computes failed_samples / spec.num_samples; the denominator is always the requested shard size, including when fail-fast leaves later samples unattempted. A ratio greater than or equal to shard_failure_threshold appends a terminal ShardFailureError with sample_idx = -1. Threshold breaches stay inside the returned ShardResult so distributed executors can serialize, aggregate, and classify the shard without relying on task-level exceptions.

Example¶

from rfgen.config import GenerationConfig
from rfgen.executors import SparkExecutor, ShardSpec
from rfgen.orchestration import build_shard_fn
from omegaconf import OmegaConf

raw = OmegaConf.to_container(OmegaConf.load("configs/config.yaml"), resolve=True)
cfg = GenerationConfig.model_validate(raw)

shard_fn = build_shard_fn(cfg)
cfg_hash = "sha256:cfg"

# Verify picklability before submitting to Spark
import pickle
pickle.loads(pickle.dumps(shard_fn))    # must not raise

executor = SparkExecutor(app_name="rfgen-demo", max_executors=16)
shards = [
    ShardSpec.for_shard(
        shard_index=i, num_samples=4096,
        config_hash=cfg_hash,
        output_uri="gs://my-bucket/dataset/v1",
    )
    for i in range(64)
]
handle = executor.submit(shard_fn, shards)
result = executor.wait(handle)
assert not result.failed_shards

rfgen.executors¶

Module summary¶

Run-control protocol¶

Public records and enums¶

Controller contract¶

class rfgen.run_control.InMemoryRunController¶

Lifecycle and unit rules¶

Class index¶

ABC: BaseDistributedExecutor¶

class rfgen.executors.SparkExecutor¶

Constructor¶

Class attributes¶

Constructor parameters¶

Method: submit¶

Parameters¶

Returns¶

Raises¶

Example¶

Method: wait¶

Returns¶

Raises¶

Method: cancel¶

Method: schema¶

Schema: SparkExecutorConfig¶

Notes¶

See Also¶

class rfgen.executors.LocalExecutor¶

class rfgen.orchestration.local.LocalRunner¶

Constructor¶

Class method: build_config¶

Parameters¶

Returns¶

Method: run¶

Returns¶

Raises¶

Example¶

Notes¶

class rfgen.executors.ManagedSparkServerlessExecutor¶

class rfgen.executors.EMRExecutor¶

class rfgen.executors.SynapseExecutor¶

class rfgen.executors.DatabricksExecutor¶

class rfgen.executors.ShardSpec¶

Method: canonical_shard_id¶

Method: for_shard¶

class rfgen.orchestration.ShardError¶

class rfgen.orchestration.ShardResult¶

class rfgen.executors.ExecutorRunHandle¶

class rfgen.executors.RunResult¶

Abstract base classes¶

BaseDistributedExecutor¶

submit()¶

wait()¶

cancel()¶

schema()¶

class rfgen.rng.SeedSchedule¶

Constructor surface¶

Fields¶

Methods¶

Notes¶

Advanced Testing Helper: ShardWorkerCollaborators¶

build_shard_fn¶

Parameters¶

Returns¶

What the returned callable does¶

Notes¶

Example¶

See Also¶

`rfgen.executors`¶

`class rfgen.run_control.InMemoryRunController`¶

ABC: `BaseDistributedExecutor`¶

`class rfgen.executors.SparkExecutor`¶

Method: `submit`¶

Method: `wait`¶

Method: `cancel`¶

Method: `schema`¶

Schema: `SparkExecutorConfig`¶

`class rfgen.executors.LocalExecutor`¶

`class rfgen.orchestration.local.LocalRunner`¶

Class method: `build_config`¶

Method: `run`¶

`class rfgen.executors.ManagedSparkServerlessExecutor`¶

`class rfgen.executors.EMRExecutor`¶

`class rfgen.executors.SynapseExecutor`¶

`class rfgen.executors.DatabricksExecutor`¶

`class rfgen.executors.ShardSpec`¶

Method: `canonical_shard_id`¶

Method: `for_shard`¶

`class rfgen.orchestration.ShardError`¶

`class rfgen.orchestration.ShardResult`¶

`class rfgen.executors.ExecutorRunHandle`¶

`class rfgen.executors.RunResult`¶

`BaseDistributedExecutor`¶

`submit()`¶

`wait()`¶

`cancel()`¶

`schema()`¶

`class rfgen.rng.SeedSchedule`¶

Advanced Testing Helper: `ShardWorkerCollaborators`¶

`build_shard_fn`¶