feat: add conditional step execution with skip/fail behavior

[jumski]

Conditional Step Execution in PGFlow

This PR adds support for conditional step execution in PGFlow, allowing steps to be conditionally executed based on flow input or dependency outputs.

Key features:

• Added condition_pattern and when_unmet columns to the steps table
• Implemented condition evaluation in the start_ready_steps function
• Added three handling modes for unmet conditions:
  • fail: Fails the step and the entire run
  • skip: Skips just the current step
  • skip-cascade: Skips the step and all its dependents

The implementation evaluates conditions before starting any steps:

• For root steps, conditions are evaluated against the flow input
• For dependent steps, conditions are evaluated against the aggregated outputs of dependencies

The DSL has been updated to support the new options:

flow.step(
  {
    slug: 'conditional_step',
    condition: { enabled: true },
    whenUnmet: 'skip'
  },
  () => 'result'
)

Comprehensive tests have been added to verify all condition evaluation scenarios and handling modes.

[changeset-bot]

⚠️ No Changeset found

Latest commit: 7372203

Merging this PR will not cause a version bump for any packages. If these changes should not result in a new version, you're good to go. If these changes should result in a version bump, you need to add a changeset.

This PR includes no changesets

When changesets are added to this PR, you'll see the packages that this PR includes changesets for and the associated semver types

Click here to learn what changesets are, and how to add one.

Click here if you're a maintainer who wants to add a changeset to this PR

[jumski]

Warning

This pull request is not mergeable via GitHub because a downstack PR is open. Once all requirements are satisfied, merge this PR as a stack on Graphite.
Learn more

• refactor: rename condition_pattern to required_input_pattern and condition_not_pattern to forbidden_input_pattern #587
• feat: add whenUnmet and whenFailed condition modes to flow shapes #585
• feat: add ifNot condition pattern for negative condition checks #584
• feat: improve type safety for flow step outputs and skippable dependencies #583
• feat: add type safety for skippable step dependencies #582
• feat: add ContainmentPattern type for JSON condition matching #576
• feat: add whenFailed option for error handling after retries exhausted #575
• feat: add conditional step execution with skip/fail behavior #573 👈 (View in Graphite)
• feat: add conditional step execution with skip infrastructure #572
• main

This stack of pull requests is managed by Graphite. Learn more about stacking.

[nx-cloud]

View your CI Pipeline Execution ↗ for commit 7372203

Command	Status	Duration	Result
nx run edge-worker:test:integration	✅ Succeeded	3m 48s	View ↗
nx run client:e2e	✅ Succeeded	1m 12s	View ↗
nx run core:pgtap	✅ Succeeded	1m 43s	View ↗
nx affected -t verify-exports --base=origin/mai...	✅ Succeeded	3s	View ↗
nx affected -t build --configuration=production...	✅ Succeeded	3s	View ↗
nx affected -t lint typecheck test --parallel -...	✅ Succeeded	50s	View ↗
nx run cli:e2e	✅ Succeeded	4s	View ↗
nx run edge-worker:e2e	✅ Succeeded	41s	View ↗

---

☁️ Nx Cloud last updated this comment at 2026-01-08 11:26:50 UTC