Step workflow

Skill datamaker-kr/synapse-claude-marketplace/plugins/synapse-plugin-helper/skills/step-workflow

Explains step-based workflow system for Synapse plugin actions. Use when the user mentions "BaseStep", "StepRegistry", "Orchestrator", "StepResult", "BaseStepContext", "step-based workflow", "workflow steps", "rollback", "progress_weight", or needs help with multi-step action development.From its SKILL.md

Install

npx -y skills add datamaker-kr/synapse-claude-marketplace --skill step-workflow

Assembled from the repository path, not quoted from the project. Check it against their README if it does not work.

2 things to look at

no licenseNo license file was found in the repository. Code published without one is not open source by default, so using it at work is a question for whoever answers licensing questions where you are.
1 stars1 stars. Stars are a popularity signal and not a quality one, but at this level it is likely that nobody has read this closely except its author, and you would be relying on your own review.

SKILL.md

4.0 KB, 872 tokens by cl100k_base, as published. Nobody here has run it

Step-based Workflow System

Synapse SDK provides a step-based workflow system for complex actions that need:

Multi-phase execution with progress tracking
Automatic rollback on failure
State sharing between steps
Conditional step execution

Core Components

Component	Purpose
`BaseStep`	Abstract step definition
`StepResult`	Step execution result
`StepRegistry`	Ordered step registration
`Orchestrator`	Step execution with rollback
`BaseStepContext`	State sharing between steps

Quick Start

from dataclasses import dataclass, field
from synapse_sdk.plugins.steps import (
    BaseStep,
    StepResult,
    StepRegistry,
    Orchestrator,
    BaseStepContext,
)
from synapse_sdk.plugins.context import RuntimeContext

# 1. Define context for state sharing
@dataclass
class ProcessContext(BaseStepContext):
    data: list = field(default_factory=list)
    processed: int = 0

# 2. Define steps
class LoadStep(BaseStep[ProcessContext]):
    @property
    def name(self) -> str:
        return 'load'

    @property
    def progress_weight(self) -> float:
        return 0.3

    def execute(self, ctx: ProcessContext) -> StepResult:
        ctx.data = load_data()
        return StepResult(success=True, data={'count': len(ctx.data)})

class ProcessStep(BaseStep[ProcessContext]):
    @property
    def name(self) -> str:
        return 'process'

    @property
    def progress_weight(self) -> float:
        return 0.7

    def execute(self, ctx: ProcessContext) -> StepResult:
        for item in ctx.data:
            process(item)
            ctx.processed += 1
            ctx.set_progress(ctx.processed, len(ctx.data))
        return StepResult(success=True)

# 3. Register and run
registry = StepRegistry[ProcessContext]()
registry.register(LoadStep())
registry.register(ProcessStep())

context = ProcessContext(runtime_ctx=runtime_ctx)
orchestrator = Orchestrator(registry, context)
result = orchestrator.execute()

Using with Specialized Actions

Override setup_steps() in specialized actions:

from synapse_sdk.plugins.actions.train import BaseTrainAction, TrainContext
from synapse_sdk.plugins.steps import StepRegistry

class MyTrainAction(BaseTrainAction[TrainParams]):
    def setup_steps(self, registry: StepRegistry[TrainContext]) -> None:
        registry.register(LoadDatasetStep())
        registry.register(TrainStep())
        registry.register(UploadModelStep())

Workflow Features

Automatic Progress Tracking

Progress is calculated based on step weights:

# Total weight = 0.2 + 0.6 + 0.2 = 1.0
LoadStep()    # progress_weight = 0.2 -> 0-20%
TrainStep()   # progress_weight = 0.6 -> 20-80%
UploadStep()  # progress_weight = 0.2 -> 80-100%

Automatic Rollback

On failure, executed steps are rolled back in reverse order:

class UploadStep(BaseStep[UploadContext]):
    def execute(self, ctx: UploadContext) -> StepResult:
        ctx.uploaded_files = upload_files()
        return StepResult(success=True)

    def rollback(self, ctx: UploadContext, result: StepResult) -> None:
        for file in ctx.uploaded_files:
            delete_file(file)

Conditional Execution

Skip steps based on context:

class OptionalStep(BaseStep[MyContext]):
    def can_skip(self, ctx: MyContext) -> bool:
        return not ctx.params.get('enable_validation', True)