orchestration

Scheduler-neutral Puzzletron campaign orchestration.

Classes

`AttemptSpec`	One submitted attempt for one work item or coordinated stage.
`CampaignController`	Scheduler-neutral durable controller for one campaign plan.
`CampaignPlan`	Compiled campaign plan bound to one experiment and runner.
`CommandSpec`	Shell command and environment for one attempt.
`ExecutionStrategy`	How one stage maps work items onto model instances.
`FailureClass`	Semantic failure classification for policy decisions.
`FailurePolicy`	Named retry/resume policy for campaign attempts.
`HaltPolicy`	When to stop the campaign after a stage attempt fails.
`JobHandle`	Opaque executor handle persisted for recovery.
`JobStatus`	Normalized executor status.
`ParallelMesh`	AutoModel-compatible parallel mesh for one model instance.
`ParallelMeshOverride`	Optional per-stage mesh override from execution config.
`RunnerEnvironment`	One scheduler-neutral runner environment bound to a campaign.
`StageExecutionSpec`	Execution semantics for one stage node.
`TaskLauncher`	How an executor starts processes inside one scheduler task.
`TaskTopology`	Scheduler-neutral task and distributed-group layout for one attempt.
`ShutdownAction`	User decision after requesting interactive controller shutdown.
`TerminalControls`	Read single-key controller commands without adding a terminal dependency.
`WorkItem`	One unit of work inside a stage plan.
`WorkPlan`	Deterministic work plan for one stage attempt.

Functions

`compile_campaign_plan`	Compile one campaign plan from experiment + runner + execution configs.
`normalize_vllm_topology`	Normalize vLLM topology, including its boolean expert-parallel mode.
`pack_gpu_allocation`	Pack independent instances onto nodes with cluster-safe exclusivity rules.
`plan_to_dict`	Serialize a campaign plan for durable storage.

class AttemptSpec

Bases: object

One submitted attempt for one work item or coordinated stage.

__init__(attempt_id, work_id, stage_id, command, allocation_nodes=1, allocation_gpus=1, exclusive=False, contract_hash='', metadata=<factory>, task_topology=<factory>)

Parameters:

attempt_id (str)
work_id (str)
stage_id (str)
command (CommandSpec)
allocation_nodes (int)
allocation_gpus (int)
exclusive (bool)
contract_hash (str)
metadata (Mapping[str, Any])
task_topology (TaskTopology)

Return type:

None

allocation_gpus: int = 1

allocation_nodes: int = 1

attempt_id: str

command: CommandSpec

contract_hash: str = ''

exclusive: bool = False

metadata: Mapping[str, Any]

stage_id: str

task_topology: TaskTopology

work_id: str

class CampaignController

Bases: object

Scheduler-neutral durable controller for one campaign plan.

__init__(plan, *, executor=None, poll_interval_seconds=5.0, local=False, logger=None, terminal_controls=None)

Parameters:

plan (CampaignPlan)
executor (Executor | None)
poll_interval_seconds (float)
local (bool)
logger (OrchestratorLogger | None)
terminal_controls (TerminalControls | None)

Return type:

None

run(*, overrides=None, once=False, max_iterations=None)

Run the controller until all stages complete or a fatal failure occurs.

Parameters:

overrides (list[str] | None)
once (bool)
max_iterations (int | None)

Return type:

dict[str, Any]

shutdown(*, reason='keyboard interrupt')

Cancel every live executor handle and mark attempts cancelled.

Parameters:: reason (str)
Return type:: int

class CampaignPlan

Bases: object

Compiled campaign plan bound to one experiment and runner.

__init__(experiment_config_path, puzzle_dir, experiment_config, runner, execution_defaults, stages, contract_hash, overrides=())

Parameters:

experiment_config_path (str)
puzzle_dir (Path)
experiment_config (Mapping[str, Any])
runner (RunnerEnvironment)
execution_defaults (Mapping[str, Any])
stages (tuple[StagePlanNode, ...])
contract_hash (str)
overrides (tuple[str, ...])

Return type:

None

contract_hash: str

execution_defaults: Mapping[str, Any]

experiment_config: Mapping[str, Any]

experiment_config_path: str

overrides: tuple[str, ...] = ()

puzzle_dir: Path

runner: RunnerEnvironment

stages: tuple[StagePlanNode, ...]

class CommandSpec

Bases: object

Shell command and environment for one attempt.

__init__(argv, env=<factory>, cwd=None, log_path=None, shell=False)

Parameters:

argv (tuple[str, ...])
env (Mapping[str, str])
cwd (str | None)
log_path (str | None)
shell (bool)

Return type:

None

argv: tuple[str, ...]

cwd: str | None = None

env: Mapping[str, str]

log_path: str | None = None

shell: bool = False

class ExecutionStrategy

Bases: str, Enum

How one stage maps work items onto model instances.

PERSISTENT_POOL = 'persistent_pool'

SHARDED = 'sharded'

SINGLE = 'single'

__new__(value)

class FailureClass

Bases: str, Enum

Semantic failure classification for policy decisions.

APPLICATION = 'application'

CANCELLED = 'cancelled'

CONFIG = 'config'

OOM = 'oom'

SUCCESS = 'success'

TIMEOUT_FATAL = 'timeout_fatal'

TIMEOUT_RESUMABLE = 'timeout_resumable'

TRANSIENT = 'transient'

UNKNOWN = 'unknown'

__new__(value)

class FailurePolicy

Bases: str, Enum

Named retry/resume policy for campaign attempts.

RESUME = 'resume'

STRICT = 'strict'

__new__(value)

class HaltPolicy

Bases: str, Enum

When to stop the campaign after a stage attempt fails.

DRAIN = 'drain'

FAIL_FAST = 'fail_fast'

__new__(value)

class JobHandle

Bases: object

Opaque executor handle persisted for recovery.

__init__(backend, handle_id, attempt_id, metadata=<factory>)

Parameters:

backend (str)
handle_id (str)
attempt_id (str)
metadata (Mapping[str, Any])

Return type:

None

attempt_id: str

backend: str

handle_id: str

metadata: Mapping[str, Any]

class JobStatus

Bases: object

Normalized executor status.

__init__(handle, state, exit_code=None, reason=None, log_paths=())

Parameters:

handle (JobHandle)
state (JobState)
exit_code (int | None)
reason (str | None)
log_paths (tuple[str, ...])

Return type:

None

exit_code: int | None = None

handle: JobHandle

log_paths: tuple[str, ...] = ()

reason: str | None = None

state: JobState

class ParallelMesh

Bases: object

AutoModel-compatible parallel mesh for one model instance.

__init__(tp=1, cp=1, pp=1, ep=1, dp_shard=1, dp_replicate=1)

Parameters:

tp (int)
cp (int)
pp (int)
ep (int)
dp_shard (int)
dp_replicate (int)

Return type:

None

as_dict()

Return type:: dict[str, int]

cp: int = 1

dp_replicate: int = 1

dp_shard: int = 1

ep: int = 1

classmethod from_mapping(mapping)

Parameters:: mapping (Mapping[str, Any])
Return type:: ParallelMesh

pp: int = 1

tp: int = 1

class ParallelMeshOverride

Bases: object

Optional per-stage mesh override from execution config.

__init__(tp=None, cp=None, pp=None, ep=None, dp_shard=None, dp_replicate=None)

Parameters:

tp (int | None)
cp (int | None)
pp (int | None)
ep (int | None)
dp_shard (int | None)
dp_replicate (int | None)

Return type:

None

cp: int | None = None

dp_replicate: int | None = None

dp_shard: int | None = None

ep: int | None = None

pp: int | None = None

tp: int | None = None

class RunnerEnvironment

Bases: object

One scheduler-neutral runner environment bound to a campaign.

__init__(kind, contract, slurm=None, baremetal=None, defaults=<factory>)

Parameters:

kind (str)
contract (ExecutionContract)
slurm (SlurmRunnerConfig | None)
baremetal (BareMetalRunnerConfig | None)
defaults (Mapping[str, Any])

Return type:

None

baremetal: BareMetalRunnerConfig | None = None

contract: ExecutionContract

defaults: Mapping[str, Any]

kind: str

slurm: SlurmRunnerConfig | None = None

class ShutdownAction

Bases: str, Enum

User decision after requesting interactive controller shutdown.

CANCEL = 'cancel'

CONTINUE = 'continue'

DETACH = 'detach'

__new__(value)

class StageExecutionSpec

Bases: object

Execution semantics for one stage node.

__init__(stage_id, strategy, instances=1, failure_policy=FailurePolicy.STRICT, mesh_override=None, gpus_per_node=None, partition=None, resource='gpu')

Parameters:

stage_id (str)
strategy (ExecutionStrategy)
instances (int)
failure_policy (FailurePolicy)
mesh_override (ParallelMeshOverride | None)
gpus_per_node (int | None)
partition (str | None)
resource (str)

Return type:

None

failure_policy: FailurePolicy = 'strict'

gpus_per_node: int | None = None

instances: int = 1

mesh_override: ParallelMeshOverride | None = None

partition: str | None = None

resource: str = 'gpu'

stage_id: str

strategy: ExecutionStrategy

class TaskLauncher

Bases: str, Enum

How an executor starts processes inside one scheduler task.

DIRECT = 'direct'

TORCHRUN = 'torchrun'

__new__(value)

class TaskTopology

Bases: object

Scheduler-neutral task and distributed-group layout for one attempt.

__init__(task_count=1, gpus_per_task=None, tasks_per_group=1, launcher=TaskLauncher.DIRECT, placement='block')

Parameters:

task_count (int)
gpus_per_task (int | None)
tasks_per_group (int)
launcher (TaskLauncher)
placement (str)

Return type:

None

gpus_per_task: int | None = None

launcher: TaskLauncher = 'direct'

placement: str = 'block'

task_count: int = 1

tasks_per_group: int = 1

class TerminalControls

Bases: object

Read single-key controller commands without adding a terminal dependency.

__init__(*, input_stream=None, output_stream=None)

Parameters:

input_stream (TextIO | None)
output_stream (TextIO | None)

Return type:

None

static action_for_choice(choice)

Parameters:: choice (str)
Return type:: ShutdownAction | None

choose_revisions(prompt, revision_ids)

Select revision IDs in cooked mode using numbers or exact IDs.

Parameters:

prompt (str)
revision_ids (tuple[str, ...])

Return type:

tuple[str, …]

choose_shutdown()

Prompt in cooked mode and return one explicit shutdown decision.

Return type:: ShutdownAction

poll_quit()

Return whether an available keypress requests the quit menu.

Return type:: bool

start()

Return type:: None

stop()

Return type:: None

class WorkItem

Bases: object

One unit of work inside a stage plan.

__init__(work_id, stage_id, shard_index, shard_count, gpus_per_instance, local_gpu_ids=(), metadata=<factory>)

Parameters:

work_id (str)
stage_id (str)
shard_index (int)
shard_count (int)
gpus_per_instance (int)
local_gpu_ids (tuple[int, ...])
metadata (Mapping[str, Any])

Return type:

None

gpus_per_instance: int

local_gpu_ids: tuple[int, ...] = ()

metadata: Mapping[str, Any]

shard_count: int

shard_index: int

stage_id: str

work_id: str

class WorkPlan

Bases: object

Deterministic work plan for one stage attempt.

__init__(stage_id, strategy, items, aggregate_required=False)

Parameters:

stage_id (str)
strategy (ExecutionStrategy)
items (tuple[WorkItem, ...])
aggregate_required (bool)

Return type:

None

aggregate_required: bool = False

items: tuple[WorkItem, ...]

stage_id: str

strategy: ExecutionStrategy

compile_campaign_plan(*, experiment_config_path, runner, execution, overrides=None, stage_filter=None)

Compile one campaign plan from experiment + runner + execution configs.

Parameters:

experiment_config_path (str | Path)
runner (RunnerEnvironment)
execution (Mapping[str, Any])
overrides (list[str] | None)
stage_filter (str | None)

Return type:

CampaignPlan

normalize_vllm_topology(topology)

Normalize vLLM topology, including its boolean expert-parallel mode.

Parameters:: topology (Mapping[str, Any])
Return type:: dict[str, Any]

pack_gpu_allocation(*, mesh, instances, gpus_per_node)

Pack independent instances onto nodes with cluster-safe exclusivity rules.

Parameters:

mesh (ParallelMesh)
instances (int)
gpus_per_node (int)

Return type:

GpuAllocation

plan_to_dict(plan)

Serialize a campaign plan for durable storage.

Parameters:: plan (CampaignPlan)
Return type:: dict[str, Any]