plexus.cli.procedure.lua_dsl.primitives.stage module

Stage Primitive - Workflow progress tracking.

Provides: - Stage.current() - Get current stage name - Stage.set(stage) - Set current stage - Stage.advance() - Move to next stage - Stage.is(stage) - Check if in specific stage - Stage.history() - Get stage transition history

class plexus.cli.procedure.lua_dsl.primitives.stage.StagePrimitive(declared_stages: List[str] | None = None, lua_sandbox=None)

Bases: object

Manages workflow stage tracking for procedures.

Enables workflows to: - Track current workflow stage - Validate stage transitions - Query stage state - Maintain stage history

Initialize Stage primitive.

Args:

declared_stages: List of valid stage names from YAML config lua_sandbox: LuaSandbox for creating Lua tables (optional)

__init__(declared_stages: List[str] | None = None, lua_sandbox=None)

Initialize Stage primitive.

Args:

declared_stages: List of valid stage names from YAML config lua_sandbox: LuaSandbox for creating Lua tables (optional)

advance() → str | None

Move to next stage in declared sequence.

Returns:

New current stage name, or None if at end

Raises:

ValueError: If current stage not set or not in declared stages

Example (Lua):

Stage.set(“setup”) – … do work … Stage.advance() – Moves to next stage in sequence Log.info(“Advanced to”, {stage = Stage.current()})

clear_history() → None

Clear stage transition history (mainly for testing).

Note: Does not affect current stage.

count() → int

Get count of stage transitions.

Returns:

Number of transitions in history

current() → str | None

Get current stage name.

Returns:

Current stage name or None if not set

Example (Lua):

local stage = Stage.current() if stage == “processing” then

Log.info(“Currently processing”)

end

history()

Get stage transition history.

Returns:

Lua table (if lua_sandbox available) or Python list of transition records

Example (Lua):

local history = Stage.history() for i, transition in ipairs(history) do

Log.debug(“Transition”, {

from = transition.from_stage, to = transition.to_stage, timestamp = transition.timestamp

})

end

is_current(stage: str) → bool

Check if currently in specific stage.

Args:

stage: Stage name to check

Returns:

True if current stage matches, False otherwise

Example (Lua):

if Stage.is(“processing”) then – ‘is’ is mapped to is_current()

Log.info(“Currently processing”)

end

set(stage: str) → None

Set current stage.

Args:

stage: Stage name to set

Raises:

ValueError: If stage not in declared stages list

Example (Lua):

Stage.set(“processing”) Log.info(“Moved to processing stage”)