stagegate

Public package namespace for stage-aware local pipeline execution.

class stagegate.Scheduler(*, resources: dict[str, int | float] | None = None, pipeline_parallelism: int = 1, task_parallelism: int | None = None, exception_exit_policy: str = 'fail_fast')

Single-process scheduler for stage-aware local pipelines and tasks.

Parameters:

resources – Optional abstract resource capacities such as {"cpu": 8}. None means no resource labels are configured. These are scheduler-defined admission labels, not OS-enforced quotas.
pipeline_parallelism – Maximum number of pipeline coordinator threads.
task_parallelism – Maximum number of concurrently admitted tasks. If None, the effective worker count is 1.
exception_exit_policy – Exceptional context-manager exit policy. "fail_fast" starts best-effort shutdown cleanup and lets the original block exception propagate immediately. "drain" waits for close() before the original block exception propagates.

shutdown_started() → bool

Return whether shutdown has begun.

Returns:: True once the scheduler has left the open state.
Return type:: bool

closed() → bool

Return whether the scheduler has fully closed.

Returns:

True only after live work has drained and runtime threads: have been joined.

Return type:

bool

snapshot() → SchedulerSnapshot

Return an immutable point-in-time snapshot of scheduler state.

Returns:

Detached aggregate view of scheduler state,: counts, and resources.

Return type:

SchedulerSnapshot

run_pipeline(pipeline: Pipeline, *, name: str | None = None) → PipelineHandle

Submit a pipeline instance for FIFO execution and return its handle.

Parameters:

pipeline – Pipeline instance to enqueue.
name – Optional submission-time monitoring label.

Returns:

Handle for observing the submitted pipeline.

Return type:

PipelineHandle

Raises:

TypeError – If pipeline is not a Pipeline instance.
RuntimeError – If shutdown has already started or the same pipeline instance was submitted previously.

wait_pipelines(handles: Iterable[PipelineHandle], timeout: float | None = None, return_when: str = 'ALL_COMPLETED') → tuple[set[PipelineHandle], set[PipelineHandle]]

Wait for pipeline handles owned by this scheduler.

Parameters:

handles – Pipeline handles created by this scheduler.
timeout – Maximum wait time in seconds. None means unbounded wait and 0 means immediate poll.
return_when – One of stagegate.FIRST_COMPLETED, stagegate.FIRST_EXCEPTION, or stagegate.ALL_COMPLETED.

Returns:

A (done, pending): pair.

Return type:

tuple[set[PipelineHandle], set[PipelineHandle]]

Raises:

TypeError – If a non-pipeline handle is provided.
ValueError – If handles is empty, ownership is wrong, timeout is invalid, or return_when is invalid.

shutdown(cancel_pending_pipelines: bool = False) → None

Start shutdown without waiting for full close.

Parameters:: cancel_pending_pipelines – Whether queued pipelines should be cancelled during shutdown processing.

close(cancel_pending_pipelines: bool = False) → None

Block until shutdown completes and runtime threads have exited.

Parameters:: cancel_pending_pipelines – Whether queued pipelines should be cancelled during close processing.
Raises:: RuntimeError – If called from a scheduler-owned runtime thread.

class stagegate.Pipeline

Base class for user-defined pipelines.

Subclass this type and override run() with the pipeline body. Pipeline-control APIs such as task(), wait(), and stage_forward() are valid only while run() is executing on the scheduler-owned coordinator thread.

run() → Any

Execute pipeline logic on a scheduler-owned coordinator thread.

Returns:

Arbitrary pipeline result object. Returning normally marks the: pipeline as succeeded.

Return type:

Any

Raises:

Exception – Any uncaught exception raised by user code marks the pipeline as failed and is later re-raised by the pipeline handle.

task(fn: Callable[..., Any], *, resources: dict[str, int | float] | None = None, args: tuple[Any, ...] = (), kwargs: dict[str, Any] | None = None, name: str | None = None) → TaskBuilder

Create a task builder for later submission via .run().

Parameters:

fn – Callable to execute later on a worker thread.
resources – Optional abstract resource requirements for admission control. None means no resources are required.
args – Positional arguments passed to fn.
kwargs – Keyword arguments passed to fn.
name – Optional user-facing task label.

Returns:

Builder object that can later be submitted with: .run().

Return type:

TaskBuilder

stage_forward() → None

Advance the pipeline stage used by future task submissions.

Raises:: RuntimeError – If called outside the active coordinator-thread control context for the pipeline.

wait(handles: Iterable[TaskHandle], timeout: float | None = None, return_when: str = 'ALL_COMPLETED') → tuple[set[TaskHandle], set[TaskHandle]]

Wait for task handles created by this pipeline.

Parameters:

handles – Task handles created by this pipeline.
timeout – Maximum wait time in seconds. None means unbounded wait and 0 means immediate poll.
return_when – One of stagegate.FIRST_COMPLETED, stagegate.FIRST_EXCEPTION, or stagegate.ALL_COMPLETED.

Returns:

A (done, pending) pair.

Return type:

tuple[set[TaskHandle], set[TaskHandle]]

Raises:

RuntimeError – If called outside the active coordinator-thread control context for the pipeline.
TypeError – If a non-task handle is provided.
ValueError – If handles is empty, ownership is wrong, timeout is invalid, or return_when is invalid.

class stagegate.TaskHandle(record: TaskRecord)

Handle for a task submitted from a pipeline.

Task handles are thin public wrappers around scheduler-owned task records. They provide observation, waiting, and pre-start cancellation / cooperative-termination requests.

cancel() → bool

Cancel the task if it has not started yet.

Returns:

True if the task was still queued or ready and could be: cancelled, otherwise False.

Return type:

bool

name() → str | None: Return the user-facing task name, if one was supplied.

request_terminate() → bool

Request cooperative termination for this task.

Returns:

True if the request was recorded or the task followed the: pre-start cancel path, otherwise False for already terminal tasks.

Return type:

bool

done() → bool

Return whether the task is terminal.

Returns:

True for succeeded, failed, terminated, or cancelled: tasks.

Return type:

bool

running() → bool

Return whether the task callable is actively running.

Returns:

True only while the task callable is executing on a: worker thread.

Return type:

bool

cancelled() → bool

Return whether the task ended in the cancelled state.

Returns:: True only for the cancelled terminal state.
Return type:: bool

result(timeout: float | None = None) → Any

Return the task result or raise its stored terminal outcome.

Parameters:

timeout – Maximum wait time in seconds. None means unbounded wait and 0 means immediate check.

Returns:

Stored task result value for succeeded tasks.

Return type:

Any

Raises:

TimeoutError – If the task is not terminal before the timeout.
CancelledError – If the task was cancelled before start.
BaseException – The original stored task exception, including TerminatedError for cooperative terminate.

exception(timeout: float | None = None) → BaseException | None

Return the task exception object, if any.

Parameters:

timeout – Maximum wait time in seconds. None means unbounded wait and 0 means immediate check.

Returns:

Stored exception for failed or terminated: tasks, or None for succeeded tasks.

Return type:

BaseException | None

Raises:

TimeoutError – If the task is not terminal before the timeout.
CancelledError – If the task was cancelled before start.

class stagegate.PipelineHandle(record: PipelineRecord)

Handle for a pipeline submitted to a scheduler.

Pipeline handles provide pipeline-level observation, queued-pipeline cancellation, immutable snapshots, and explicit post-terminal disposal via discard().

cancel() → bool

Cancel the pipeline if it has not started yet.

Returns:

True if the pipeline was still queued and could be: cancelled, otherwise False.

Return type:

bool

name() → str | None

Return the submission-time pipeline name, if one was supplied.

Raises:: DiscardedHandleError – If the handle was discarded.

discard() → None

Discard this handle and release retained terminal outcome data.

After discard, observation methods on this handle raise DiscardedHandleError. Separately held task handles remain valid.

Raises:: RuntimeError – If the pipeline is not yet terminal.

done() → bool

Return whether the pipeline is terminal.

Returns:: True for succeeded, failed, or cancelled pipelines.
Return type:: bool
Raises:: DiscardedHandleError – If the handle was discarded.

running() → bool

Return whether the pipeline run() method is active.

Returns:

True only while the scheduler-owned coordinator thread is: executing Pipeline.run().

Return type:

bool

Raises:

DiscardedHandleError – If the handle was discarded.

cancelled() → bool

Return whether the pipeline ended in the cancelled state.

Returns:: True only for the cancelled terminal state.
Return type:: bool
Raises:: DiscardedHandleError – If the handle was discarded.

result(timeout: float | None = None) → Any

Return the pipeline result or raise its stored terminal outcome.

Parameters:

timeout – Maximum wait time in seconds. None means unbounded wait and 0 means immediate check.

Returns:

Stored pipeline result for succeeded pipelines.

Return type:

Any

Raises:

TimeoutError – If the pipeline is not terminal before the timeout.
CancelledError – If the pipeline was cancelled before start.
DiscardedHandleError – If the handle was discarded.
BaseException – The stored pipeline exception for failed pipelines.

exception(timeout: float | None = None) → BaseException | None

Return the pipeline exception object, if any.

Parameters:

timeout – Maximum wait time in seconds. None means unbounded wait and 0 means immediate check.

Returns:

Stored pipeline exception for failed: pipelines, or None for succeeded pipelines.

Return type:

BaseException | None

Raises:

TimeoutError – If the pipeline is not terminal before the timeout.
CancelledError – If the pipeline was cancelled before start.
DiscardedHandleError – If the handle was discarded.

snapshot() → PipelineSnapshot

Return an immutable point-in-time snapshot for this pipeline.

Returns:

Detached aggregate view of the pipeline and its: child-task counts.

Return type:

PipelineSnapshot

Raises:

DiscardedHandleError – If the handle was discarded.

exception stagegate.CancelledError

Raised when a cancelled handle is observed as if it had a result.

This exception is used by both task and pipeline handles when result() or exception() is requested after a queued item was cancelled before start.

exception stagegate.DiscardedHandleError: Raised when an operation is requested from a discarded handle.

exception stagegate.SchedulerAbortError

Raised when a running pipeline is aborted during fail-fast cleanup.

This exception is used for pipelines that were already running when a scheduler context-manager block exited with an uncaught exception under the fail-fast exceptional-exit policy.

exception stagegate.TerminatedError(*, argv: tuple[str, ...], pid: int | None, returncode: int | None, forced_kill: bool)

Raised when a task terminates cooperatively after a terminate request.

argv: Command-line arguments associated with the terminated subprocess, or an empty tuple when no subprocess metadata exists.

pid: Process identifier for the terminated subprocess, if any.

returncode: Observed subprocess return code, if any.

forced_kill: Whether the subprocess helper had to escalate from SIGTERM to SIGKILL.

exception stagegate.UnknownResourceError: Raised when a task requests a resource label unknown to the scheduler.

exception stagegate.UnschedulableTaskError: Raised when a single task can never fit within scheduler capacity.

stagegate.terminate_requested() → bool

Return whether the current task has received a terminate request.

Returns:: True only while user code is running on a worker thread with an ambient task context whose terminate-request flag is set.
Return type:: bool

stagegate.terminate_tracked_subprocesses() → int

Send SIGTERM to all currently tracked subprocess process groups.

This helper is intended for user-managed shutdown cleanup, for example from a Python-level signal handler or KeyboardInterrupt cleanup path. stagegate does not install signal handlers itself.

Returns:: Number of tracked process groups that were targeted.
Return type:: int
Raises:: NotImplementedError – If called on Windows, where the current process-group signaling contract is not supported.

stagegate.run_subprocess(argv: Sequence[str | PathLike[str]], *, terminate_grace_seconds: float | None = 5.0) → int

Run a subprocess that cooperates with task termination requests.

The child is started in its own process group. If a task terminate request is observed first, the helper sends SIGTERM to that process group and, if necessary, escalates to SIGKILL after the grace timeout.

This helper is currently intended for POSIX platforms such as Linux, macOS, and BSD. Its terminate path depends on process-group signaling semantics that are not currently documented or supported on Windows. stagegate itself may still be usable on Windows for workloads that do not rely on this helper.

Parameters:

argv – Executable plus arguments. shell=True is never used.
terminate_grace_seconds – Seconds to wait after SIGTERM before sending SIGKILL. None means wait indefinitely after SIGTERM. 0 is allowed and means immediate escalation if the process has not already exited.

Returns:

Process return code on normal completion.

Return type:

int

Raises:

ValueError – If argv is empty or the grace timeout is negative.
NotImplementedError – If called on Windows, where the current process-group terminate path is not supported.
TerminatedError – If the terminate path is taken and the child exits after the terminate request.

stagegate.run_shell(command: str, *, terminate_grace_seconds: float | None = 5.0) → int

Run a shell command that cooperates with task termination requests.

The command is executed via POSIX-minimum /bin/sh -c. The shell is started in its own process group so a terminate request can signal the entire shell-controlled process tree.

Parameters:

command – Shell command string executed by /bin/sh -c.
terminate_grace_seconds – Seconds to wait after SIGTERM before sending SIGKILL. None means wait indefinitely after SIGTERM. 0 is allowed and means immediate escalation if the shell process group has not already exited.

Returns:

Shell return code on normal completion.

Return type:

int

Raises:

TypeError – If command is not a string.
ValueError – If the grace timeout is negative.
NotImplementedError – If called on Windows, where the current process-group terminate path is not supported.
TerminatedError – If the terminate path is taken and the shell exits after the terminate request.

class stagegate.ResourceSnapshot(label: str, capacity: int | float, in_use: int | float, available: int | float)

Immutable point-in-time view of one configured resource label.

label

Scheduler-defined resource label.

Type:: str

capacity

Total configured capacity for the resource.

Type:: int | float

in_use

Current reserved amount.

Type:: int | float

available

Remaining unreserved amount.

Type:: int | float

class stagegate.RunningPipelineSummary(pipeline_id: int, name: str | None)

Immutable scheduler-inspection summary for one running pipeline.

pipeline_id

Stable local identifier for the pipeline.

Type:: int

name

Optional submission-time monitoring label.

Type:: str | None

class stagegate.TaskCountsSnapshot(queued: int, admitted: int, running: int, succeeded: int, failed: int, terminated: int, cancelled: int, total: int)

Immutable aggregate task counts.

queued

Number of queued tasks.

Type:: int

admitted

Number of admitted tasks, including running tasks.

Type:: int

running

Number of actively running tasks.

Type:: int

succeeded

Number of succeeded tasks.

Type:: int

failed

Number of failed tasks.

Type:: int

terminated

Number of cooperatively terminated tasks.

Type:: int

cancelled

Number of cancelled tasks.

Type:: int

total

Total number of tasks represented by the snapshot.

Type:: int

class stagegate.PipelineCountsSnapshot(queued: int, running: int, succeeded: int, failed: int, cancelled: int, total: int)

Immutable aggregate pipeline counts.

queued

Number of queued pipelines.

Type:: int

running

Number of running pipelines.

Type:: int

succeeded

Number of succeeded pipelines.

Type:: int

failed

Number of failed pipelines.

Type:: int

cancelled

Number of cancelled pipelines.

Type:: int

total

Total number of pipelines represented by the snapshot.

Type:: int

class stagegate.SchedulerSnapshot(shutdown_started: bool, closed: bool, pipeline_parallelism: int, task_parallelism: int, pipelines: PipelineCountsSnapshot, tasks: TaskCountsSnapshot, resources: tuple[ResourceSnapshot, ...], running_pipelines: tuple[RunningPipelineSummary, ...])

Immutable scheduler-wide snapshot.

shutdown_started

Whether shutdown has begun.

Type:: bool

closed

Whether the scheduler is fully closed.

Type:: bool

pipeline_parallelism

Configured number of pipeline coordinator slots.

Type:: int

task_parallelism

Effective task parallelism.

Type:: int

pipelines

Aggregate pipeline counts.

Type:: stagegate.snapshots.PipelineCountsSnapshot

tasks

Aggregate task counts.

Type:: stagegate.snapshots.TaskCountsSnapshot

resources

Deterministically ordered resource snapshots.

Type:: tuple[stagegate.snapshots.ResourceSnapshot, …]

running_pipelines

Optional lightweight running-pipeline summaries.

Type:: tuple[stagegate.snapshots.RunningPipelineSummary, …]

class stagegate.PipelineSnapshot(pipeline_id: int, name: str | None, state: str, stage_index: int, tasks: TaskCountsSnapshot)

Immutable per-pipeline snapshot.

pipeline_id

Stable local identifier for the pipeline.

Type:: int

name

Optional submission-time monitoring label.

Type:: str | None

state

Public string representation of the pipeline state.

Type:: str

stage_index

Current pipeline stage index.

Type:: int

tasks

Aggregate task counts for that pipeline only.

Type:: stagegate.snapshots.TaskCountsSnapshot