pathforge.execution¶

Scheduler-neutral execution contracts used by local processes, Dask, and generated SLURM arrays.

Plan and execute idempotent PathForge work units on local or HPC schedulers.

class pathforge.execution.distributed.WorkRecord[source]¶

Bases: BaseModel

One immutable, scheduler-neutral unit of distributed PathForge work.

schema_version: int¶

stage: WorkStage¶

work_id: str¶

config_path: str¶

payload: dict[str, Any]¶

dependencies: list[str]¶

model_config = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pathforge.execution.distributed.ExecutionPlan[source]¶

Bases: BaseModel

Metadata and manifest locations for one distributed experiment plan.

schema_version: int¶

plan_id: str¶

created_at: str¶

config_path: str¶

plan_dir: str¶

feature_manifest: str¶

benchmark_manifest: str¶

status_dir: str¶

results_dir: str¶

num_feature_jobs: int¶

num_benchmark_jobs: int¶

num_optimization_workers: int¶

model_config = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pathforge.execution.distributed.WorkStatus[source]¶

Bases: BaseModel

Machine-readable lifecycle record written atomically by one worker.

schema_version: int¶

work_id: str¶

stage: WorkStage¶

state: Literal['running', 'success', 'failed', 'skipped']¶

started_at: str¶

finished_at: str | None¶

hostname: str¶

pid: int¶

slurm_job_id: str | None¶

slurm_array_task_id: str | None¶

result: dict[str, Any]¶

error: str | None¶

model_config = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

pathforge.execution.distributed.read_work_record(manifest_path: str | Path, index: int) → WorkRecord[source]¶

Read one zero-based work record from a JSON-lines manifest.

Parameters:

manifest_path (str | Path)
index (int)

Return type:

WorkRecord

pathforge.execution.distributed.create_execution_plan(config_path: str | Path, plan_dir: str | Path) → ExecutionPlan[source]¶

Materialize deduplicated feature and benchmark manifests plus SLURM files.

Parameters:

config_path (str | Path)
plan_dir (str | Path)

Return type:

ExecutionPlan

pathforge.execution.distributed.load_plan(path: str | Path) → ExecutionPlan[source]¶

Load and validate an execution plan JSON file.

Parameters:: path (str | Path)
Return type:: ExecutionPlan

pathforge.execution.distributed.execute_work_record(plan_path: str | Path, *, stage: Literal['features', 'benchmark'], index: int, resume: bool = True) → WorkStatus[source]¶

Execute one manifest record and atomically persist its lifecycle status.

Parameters:

plan_path (str | Path)
stage (Literal['features', 'benchmark'])
index (int)
resume (bool)

Return type:

WorkStatus

pathforge.execution.distributed.execute_stage(plan_path: str | Path, *, stage: Literal['features', 'benchmark'], backend: Literal['local', 'dask'] = 'local', scheduler_address: str | None = None) → list[dict[str, Any]][source]¶

Execute every record in one stage through local processes or Dask.

Parameters:

plan_path (str | Path)
stage (Literal['features', 'benchmark'])
backend (Literal['local', 'dask'])
scheduler_address (str | None)

Return type:

list[dict[str, Any]]

pathforge.execution.distributed.aggregate_plan(plan_path: str | Path) → Path[source]¶

Aggregate isolated worker statuses into one deterministic CSV report.

Parameters:: plan_path (str | Path)
Return type:: Path