fix: archive task messages for skipped steps and handle late callbacks

[jumski]

Archive Task Messages for Skipped Steps

This PR adds message archiving functionality to prevent "zombie" tasks from being processed after a step has been skipped or completed. Key improvements:

1. Added guards in complete_task and fail_task to handle late callbacks:

   • Detects when callbacks arrive after a step is no longer in 'started' state
   • Archives the task message to prevent stuck work
   • Returns current task state without mutating step or run state

2. Enhanced _cascade_force_skip_steps to archive queued/started task messages for skipped steps:

   • Ensures all task messages are archived when a step is force-skipped
   • Prevents workers from processing tasks for steps that are already skipped

3. Modified fail_task to archive sibling task messages when a step is skipped:

   • When a task failure causes a step to be skipped, all sibling task messages are archived
   • Prevents double-decrement of remaining_steps when multiple tasks fail on the same step

4. Improved start_tasks to reject tasks for skipped steps:

   • Added guard conditions to only start tasks for steps in 'started' state
   • Prevents race conditions where a task might be started after its step was skipped

Added comprehensive tests to verify all these behaviors work correctly.

[changeset-bot]

⚠️ No Changeset found

Latest commit: 714b3b6

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]

• fix: optimize step skipping and improve concurrent task handling #613
• feat: add constraints for input patterns and improve cascade skip tests #611
• fix: archive task messages for skipped steps and handle late callbacks #609 👈 (View in Graphite)
• fix: improve condition handling and event emission for failed steps #607
• fix: correctly decrement remaining_deps for multiple skipped parents #606
• rename when_failed to when_exhausted for clarity #602
• feat: add support for skipped steps in client #601
• feat: add conditional flow execution and retry handling #600
• docs: add conditional steps documentation and skipped step styling #594
• feat: add conditional step execution with skip infrastructure #572 : 1 other dependent PR (#573 )
• main

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

[nx-cloud]

View your CI Pipeline Execution ↗ for commit 785877d

Command	Status	Duration	Result
nx test:types:health dsl	✅ Succeeded	12s	View ↗

---

☁️ Nx Cloud last updated this comment at 2026-03-17 19:36:39 UTC

[github-actions]

🔍 Preview Deployment: Website

✅ Deployment successful!

🔗 Preview URL: https://pr-609.pgflow.pages.dev

📝 Details:

• Branch: 02-14-fix_missing_skip_archival
• Commit: 2e42c42eca6c598ee3fa2f7cd2716d7b4f125c1f
• View Logs

_Last updated: _

[jumski]

Merge activity

• Mar 17, 7:20 PM UTC: A user started a stack merge that includes this pull request via Graphite.
• Mar 17, 7:35 PM UTC: Graphite rebased this pull request as part of a merge.
• Mar 17, 7:36 PM UTC: @jumski merged this pull request with Graphite.

[github-actions]

🚀 Production Deployment: Website

✅ Successfully deployed to production!

🔗 Production URL: https://pgflow.dev

📝 Details:

• Commit: 6873dbe814bcb98adfc4300cad3e0d0f7572bbeb
• View Logs

Deployed at: 2026-03-17T20:36:29+01:00