run-llama/workflows-py
1
0
Fork 0
mirror of https://github.com/run-llama/workflows-py.git synced 2026-07-20 19:57:02 -04:00
Code Issues 18 Packages Projects Releases 10 Wiki Activity

Concurrent step not cancelled after StopEvent in 2.14.0 #35

New Issue
Closed
opened 2026-02-16 02:16:13 -05:00 by yindo · 7 comments
yindo commented 2026-02-16 02:16:13 -05:00
Owner
Copy Link
Copy Source

Originally created by @schelv on GitHub (Feb 6, 2026).

After upgrading from llama-index-workflows 2.13.1 to 2.14.0, one of my workflow tests started failing.

The workflow has two concurrent steps. The test covers the case where one step finishes early and returns a StopEvent, which should cancel the other concurrent step.

Expected behavior (2.13.1)

StartEvent -> some_step       -> StopEvent
           -> some_other_step -X (cancelled)

Event stream:

[StopEvent]

Actual behavior (2.14.0)

StartEvent -> some_step       -> StopEvent
           -> some_other_step -------------> OtherEvent

Event stream:

[OtherEvent, StopEvent]

It looks like StopEvent no longer cancels the running concurrent steps in 2.14.0.

Originally created by @schelv on GitHub (Feb 6, 2026). After upgrading from **llama-index-workflows 2.13.1** to **2.14.0**, one of my workflow tests started failing. The workflow has two concurrent steps. The test covers the case where one step finishes early and returns a StopEvent, which should cancel the other concurrent step. #### Expected behavior (2.13.1) ``` StartEvent -> some_step -> StopEvent -> some_other_step -X (cancelled) ``` Event stream: ``` [StopEvent] ``` #### Actual behavior (2.14.0) ``` StartEvent -> some_step -> StopEvent -> some_other_step -------------> OtherEvent ``` Event stream: ``` [OtherEvent, StopEvent] ``` It looks like `StopEvent` no longer cancels the running concurrent steps in 2.14.0.
yindo closed this issue 2026-02-16 02:16:13 -05:00
yindo commented 2026-02-16 02:16:14 -05:00
Author
Owner
Copy Link
Copy Source

@adrianlyjak commented on GitHub (Feb 6, 2026):

@schelv can you share a bit more about the structure of your workflow and the test so I can recreate it precisely?

  • You're mentioning that OtherEvent appears in the event stream: are you calling ctx.write_event_to_stream within the steps? Or is this an implicit publish to the event stream via an InputRequiredEvent?
  • How are you tracking actual task cancellation? Just via the event stream? Or another mechanism
@adrianlyjak commented on GitHub (Feb 6, 2026): @schelv can you share a bit more about the structure of your workflow and the test so I can recreate it precisely? - You're mentioning that `OtherEvent` appears in the event stream: are you calling `ctx.write_event_to_stream` within the steps? Or is this an implicit publish to the event stream via an `InputRequiredEvent`? - How are you tracking actual task cancellation? Just via the event stream? Or another mechanism
yindo commented 2026-02-16 02:16:14 -05:00
Author
Owner
Copy Link
Copy Source

@adrianlyjak commented on GitHub (Feb 6, 2026):

Another thing I'm noticing is that you mention the event stream is

[OtherEvent, StopEvent]

This is the actual order of events you see? This seems like the simultaneous event is published before the final stop event step is fully committed. Not reflective of whether the first step has been cancelled or not, but rather a variation in timing introduced in the new release.

@adrianlyjak commented on GitHub (Feb 6, 2026): Another thing I'm noticing is that you mention the event stream is ``` [OtherEvent, StopEvent] ``` This is the actual order of events you see? This seems like the simultaneous event is published before the final stop event step is fully committed. Not reflective of whether the first step has been cancelled or not, but rather a variation in timing introduced in the new release.
yindo commented 2026-02-16 02:16:14 -05:00
Author
Owner
Copy Link
Copy Source

@adrianlyjak commented on GitHub (Feb 6, 2026):

@schelv I was able to recreate on a test comparing 2.13.1 to 2.14.0. Appreciate if you could retry your test on 2.14.1 (should be released by the time you read this). Feel free to re-open if you still see any issues!

@adrianlyjak commented on GitHub (Feb 6, 2026): @schelv I was able to recreate on a test comparing 2.13.1 to 2.14.0. Appreciate if you could retry your test on 2.14.1 (should be released by the time you read this). Feel free to re-open if you still see any issues!
yindo commented 2026-02-16 02:16:14 -05:00
Author
Owner
Copy Link
Copy Source

@schelv commented on GitHub (Feb 6, 2026):

I will test it after the weekend😀

@schelv commented on GitHub (Feb 6, 2026): I will test it after the weekend😀
yindo commented 2026-02-16 02:16:14 -05:00
Author
Owner
Copy Link
Copy Source

@schelv commented on GitHub (Feb 9, 2026):

I was able to test the new release.
It looks like the problem is solved in some cases.

For my tests I've managed to fix the problem by adding some sleep timeout.

I've tried to create a minimal test example to see how adding sleep timeout solves the problem.
When I add some timeout everything works fine.

But when I set the sleep timeout to 0 I'm still able getting an event in the event stream that shouldn't be there (because the workflow should have already stopped).

Here is my attempt at a minimal example (it is probably not minimal):

import asyncio

from llama_index.core.workflow import Workflow, step, Context
from workflows.events import StopEvent, StartEvent, Event


class OtherEvent(Event):
    pass


class OtherStreamEvent(Event):
    pass


class CheckPassedEvent(Event):
    pass


class CheckNotPassedEvent(Event):
    pass


class InitializationCompleteEvent(Event):
    pass


sleep_duration = 0


class ExampleWorkflow(Workflow):

    def __init__(self, slow_step: str, passes_check=False, *args, **kwargs):
        super().__init__()
        self.slow_step = slow_step
        self.check_succeeds = passes_check

    @step
    async def check_step_that_can_stop_the_workflow_early(self, ctx: Context, ev: StartEvent) -> CheckPassedEvent | StopEvent:

        if self.slow_step == "check_step":
            await asyncio.sleep(sleep_duration)

        if self.check_succeeds:
            ctx.write_event_to_stream(CheckPassedEvent())
            return CheckPassedEvent()

        ctx.write_event_to_stream(CheckNotPassedEvent())
        return StopEvent()

    @step
    async def initialization_step(self, ctx: Context, ev: StartEvent) -> InitializationCompleteEvent:
        await ctx.store.set('a', 'b')
        return InitializationCompleteEvent()

    @step
    async def some_other_step(self, ctx: Context, ev: InitializationCompleteEvent) -> OtherEvent:

        if self.slow_step == "some_other_step":
            await asyncio.sleep(sleep_duration)

        ctx.write_event_to_stream(OtherStreamEvent())
        return OtherEvent()

    @step
    async def some_third_step(self, ctx: Context, ev: OtherEvent | CheckPassedEvent) -> StopEvent:

        result = ctx.collect_events(ev, [CheckPassedEvent, OtherEvent])
        if result is None:
            return None

        return StopEvent()


async def main():
    for slow_step in ["check_step", "some_other_step"]:
        for check_passes in [True, False]:
            handler = ExampleWorkflow(slow_step=slow_step, passes_check=check_passes).run()
            print(f'workflow with slow_step {slow_step} and check_passes {check_passes} has event stream {[event async for event in handler.stream_events()]}')


asyncio.run(main())

With sleep_duration = 0 gives output:

workflow with slow_step check_step and check_passes True has event stream [CheckPassedEvent(), OtherStreamEvent(), StopEvent()]
workflow with slow_step check_step and check_passes False has event stream [CheckNotPassedEvent(), OtherStreamEvent(), StopEvent()]
workflow with slow_step some_other_step and check_passes True has event stream [CheckPassedEvent(), OtherStreamEvent(), StopEvent()]
workflow with slow_step some_other_step and check_passes False has event stream [CheckNotPassedEvent(), StopEvent()]

note that OtherStreamEvent() appears even when check_passes=False, where the workflow should stop early.

@schelv commented on GitHub (Feb 9, 2026): I was able to test the new release. It looks like the problem is solved in some cases. For my tests I've managed to fix the problem by adding some sleep timeout. I've tried to create a minimal test example to see how adding sleep timeout solves the problem. When I add some timeout everything works fine. But when I set the sleep timeout to 0 I'm still able getting an event in the event stream that shouldn't be there (because the workflow should have already stopped). Here is my attempt at a minimal example (it is probably not minimal): ``` import asyncio from llama_index.core.workflow import Workflow, step, Context from workflows.events import StopEvent, StartEvent, Event class OtherEvent(Event): pass class OtherStreamEvent(Event): pass class CheckPassedEvent(Event): pass class CheckNotPassedEvent(Event): pass class InitializationCompleteEvent(Event): pass sleep_duration = 0 class ExampleWorkflow(Workflow): def __init__(self, slow_step: str, passes_check=False, *args, **kwargs): super().__init__() self.slow_step = slow_step self.check_succeeds = passes_check @step async def check_step_that_can_stop_the_workflow_early(self, ctx: Context, ev: StartEvent) -> CheckPassedEvent | StopEvent: if self.slow_step == "check_step": await asyncio.sleep(sleep_duration) if self.check_succeeds: ctx.write_event_to_stream(CheckPassedEvent()) return CheckPassedEvent() ctx.write_event_to_stream(CheckNotPassedEvent()) return StopEvent() @step async def initialization_step(self, ctx: Context, ev: StartEvent) -> InitializationCompleteEvent: await ctx.store.set('a', 'b') return InitializationCompleteEvent() @step async def some_other_step(self, ctx: Context, ev: InitializationCompleteEvent) -> OtherEvent: if self.slow_step == "some_other_step": await asyncio.sleep(sleep_duration) ctx.write_event_to_stream(OtherStreamEvent()) return OtherEvent() @step async def some_third_step(self, ctx: Context, ev: OtherEvent | CheckPassedEvent) -> StopEvent: result = ctx.collect_events(ev, [CheckPassedEvent, OtherEvent]) if result is None: return None return StopEvent() async def main(): for slow_step in ["check_step", "some_other_step"]: for check_passes in [True, False]: handler = ExampleWorkflow(slow_step=slow_step, passes_check=check_passes).run() print(f'workflow with slow_step {slow_step} and check_passes {check_passes} has event stream {[event async for event in handler.stream_events()]}') asyncio.run(main()) ``` With `sleep_duration = 0` gives output: ``` workflow with slow_step check_step and check_passes True has event stream [CheckPassedEvent(), OtherStreamEvent(), StopEvent()] workflow with slow_step check_step and check_passes False has event stream [CheckNotPassedEvent(), OtherStreamEvent(), StopEvent()] workflow with slow_step some_other_step and check_passes True has event stream [CheckPassedEvent(), OtherStreamEvent(), StopEvent()] workflow with slow_step some_other_step and check_passes False has event stream [CheckNotPassedEvent(), StopEvent()] ``` note that OtherStreamEvent() appears even when check_passes=False, where the workflow should stop early.
yindo commented 2026-02-16 02:16:14 -05:00
Author
Owner
Copy Link
Copy Source

@adrianlyjak commented on GitHub (Feb 9, 2026):

@schelv, from what I understand, this looks expected.

Steps for the same event run concurrently (here, there's two parallel paths triggered from StartEvent, that creates a race condition in the structure of the workflow. For the case of slow_step="check_step" and check_passes=False, with a sleep(0), that's enough yield to allow the other fast path to quickly iterate through its sequential initialization_step -> some_other_step. Once some_other_step is running, it will quickly reach the write_event_to_stream.

Structurally I'd recommend some kind of waiting condition if you need to avoid the publishing until you're confident that the check passes. Some ideas:

  • move ctx.write_event_to_stream(OtherStreamEvent()) into some_third_step after the ctx.collect_events are collected, or
  • Trigger initialization_step from CheckPassedEvent instead of StartEvent
  • move the collect CheckPassedEvent events earlier to guard from publishing, either from some_third_step into the initialization_step or some_other_step
@adrianlyjak commented on GitHub (Feb 9, 2026): @schelv, from what I understand, this looks expected. Steps for the same event run concurrently (here, there's two parallel paths triggered from `StartEvent`, that creates a race condition in the structure of the workflow. For the case of `slow_step="check_step"` and `check_passes=False`, with a `sleep(0)`, that's enough yield to allow the other fast path to quickly iterate through its sequential `initialization_step -> some_other_step`. Once `some_other_step` is running, it will quickly reach the `write_event_to_stream`. Structurally I'd recommend some kind of waiting condition if you need to avoid the publishing until you're confident that the check passes. Some ideas: - move `ctx.write_event_to_stream(OtherStreamEvent())` into `some_third_step` after the `ctx.collect_events` are collected, or - Trigger `initialization_step` from `CheckPassedEvent` instead of `StartEvent` - move the collect `CheckPassedEvent` events earlier to guard from publishing, either from `some_third_step` into the `initialization_step` or `some_other_step`
yindo commented 2026-02-16 02:16:14 -05:00
Author
Owner
Copy Link
Copy Source

@schelv commented on GitHub (Feb 10, 2026):

@adrianlyjak Thank you for looking into it.
The problem was mainly that my previous tests had some implicit assumption about order.
The new version of the package brought those assumptions to the surface and made me fall into the asyncio rabbit hole.
I've crawled out, fixed my test, and am moving on. 😀
Thank you again for looking into things!

@schelv commented on GitHub (Feb 10, 2026): @adrianlyjak Thank you for looking into it. The problem was mainly that my previous tests had some implicit assumption about order. The new version of the package brought those assumptions to the surface and made me fall into the asyncio rabbit hole. I've crawled out, fixed my test, and am moving on. 😀 Thank you again for looking into things!
Sign in to join this conversation.
No Branch/Tag Specified
Branches Tags
Labels
Clear labels
bug
enhancement
enhancement
good first issue
pull-request
Mirrored from GitHub Pull Request
 question
No Label
Milestone
No items
No Milestone
Projects
Clear projects
No project
Assignees
Clear assignees
yindo
No Assignees
1 Participants
Notifications
Due Date
No due date set.
Dependencies

No dependencies set.

Reference: run-llama/workflows-py#35
Delete Branch "%!s()"

Deleting a branch is permanent. Although the deleted branch may continue to exist for a short time before it actually gets removed, it CANNOT be undone in most cases. Continue?