Repository navigation
ChatWorkflow may leave tasks running after stream closure or cancellation #2876
Description
Issue
The concern affects ChatWorkflow.stream_many() and ChatWorkflow.stream_batch().
In the reviewed source, both methods start workflows through asyncio.as_completed(). Neither method has an explicit cleanup step for unfinished tasks when stream processing stops. Source.
Tests of the copied scheduling logic show two conditions:
- The caller closes the stream after the first result. The remaining workflow stays active and continues execution.
- The caller cancels the task that waits for the next result. Both unfinished workflows stay active and continue execution.
The event loop stays active during these tests. The closure test calls await stream.aclose(). It does not rely on a break statement or event-loop shutdown.
Why
Technical cause
Both methods pass workflow coroutines to asyncio.as_completed(). The reviewed code does not keep explicit task references for later cancellation. It also has no finally block to cancel unfinished tasks and wait for them to stop.
In the isolated tests, closure or cancellation stops stream processing but does not stop the scheduled workflows. When the test releases those workflows, they continue execution.
This report concerns cleanup in the Giskard methods, not a defect in asyncio.as_completed(). See the Python 3.13 reference.
Potential user impact
A caller can stop stream processing while work started by the stream remains active.
Giskard workflows can call a model or execute a tool. Continued workflow execution could therefore allow external operations after stream cancellation. See the workflow implementation.
Where
File: libs/giskard-agents/src/giskard/agents/workflow.py
| Method | Relevant code |
|---|---|
ChatWorkflow.stream_many() |
Creates workflow coroutines, then awaits and yields results from asyncio.as_completed(). No explicit task cleanup surrounds this loop. |
ChatWorkflow.stream_batch() |
Copies workflows for the input batch, then awaits and yields results from asyncio.as_completed(). No explicit task cleanup surrounds this loop. |
Activity
linear-code commentedon Oct 12, 2026
OSS-153
xz235 commentedon Oct 14, 2026
I've opened a PR for this: #2877
It keeps the
Taskhandles explicitly (asyncio.ensure_future) instead of handing barecoroutines to
as_completed(), and cancels whatever is still pending in afinallyblock,awaiting them with
gather(..., return_exceptions=True)so the workflows reach their owncleanup.
Verification (scripted generator, no API key needed): before the patch, cancelling the
consumer left 3 of 3 workflows running with 0 cancellations; after, all 3 are cancelled and
none are left running. Same for the
aclose()path. Added regression tests forstream_many(both paths) andstream_batch; they fail onmainand pass with the patch.pytest tests src -m "not functional"→ 193 passed.Happy to adjust the approach if you'd prefer a different one (e.g. scoping the cancellation
to only the not-yet-started workflows, or exposing it behind a flag).