fix(pipeline): cancel sibling loops on failure (TaskGroup, no close-during-use)
Pipeline.run() used asyncio.gather, so if one loop raised, the failing coroutine propagated while the sibling loops kept running detached; aclose() in the finally then closed a source/stt out from under a still-live loop. Switch to asyncio.TaskGroup so a failing loop cancels+awaits the siblings before teardown. Add a regression test asserting an error in the conversation loop cancels the perception loop and still closes every source.
This commit is contained in:
@@ -3,6 +3,8 @@ into the brain's replies."""
|
|||||||
|
|
||||||
import asyncio
|
import asyncio
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
from wsai.backends.mock import (
|
from wsai.backends.mock import (
|
||||||
MockBrain,
|
MockBrain,
|
||||||
MockFrameSource,
|
MockFrameSource,
|
||||||
@@ -10,6 +12,7 @@ from wsai.backends.mock import (
|
|||||||
MockTTS,
|
MockTTS,
|
||||||
MockVision,
|
MockVision,
|
||||||
)
|
)
|
||||||
|
from wsai.interfaces import Frame
|
||||||
from wsai.pipeline import Pipeline
|
from wsai.pipeline import Pipeline
|
||||||
|
|
||||||
|
|
||||||
@@ -56,6 +59,44 @@ def test_voice_only_pipeline_runs_without_eyes():
|
|||||||
assert "아직 화면을 못 읽었어요" in replies[0]
|
assert "아직 화면을 못 읽었어요" in replies[0]
|
||||||
|
|
||||||
|
|
||||||
|
def test_error_in_one_loop_cancels_siblings_and_closes():
|
||||||
|
"""If the conversation loop raises, the perception loop must be cancelled
|
||||||
|
(not left running detached) and every source must still be closed — i.e. no
|
||||||
|
close-during-use and no orphaned task."""
|
||||||
|
closed = {"source": False, "stt": False}
|
||||||
|
|
||||||
|
class ForeverSource:
|
||||||
|
async def frames(self):
|
||||||
|
while True:
|
||||||
|
await asyncio.sleep(0.01)
|
||||||
|
yield Frame(data=b"", width=1, height=1, ts=0.0)
|
||||||
|
|
||||||
|
async def aclose(self):
|
||||||
|
closed["source"] = True
|
||||||
|
|
||||||
|
class BoomSTT(MockSTT):
|
||||||
|
async def aclose(self):
|
||||||
|
closed["stt"] = True
|
||||||
|
|
||||||
|
class BoomBrain(MockBrain):
|
||||||
|
async def respond(self, user_text, screen, history):
|
||||||
|
raise RuntimeError("boom")
|
||||||
|
|
||||||
|
pipe = Pipeline(
|
||||||
|
source=ForeverSource(),
|
||||||
|
vision=MockVision(),
|
||||||
|
brain=BoomBrain(),
|
||||||
|
stt=BoomSTT(script=["hi"], interval=0.01),
|
||||||
|
tts=MockTTS(),
|
||||||
|
)
|
||||||
|
|
||||||
|
with pytest.raises(BaseException): # TaskGroup raises an ExceptionGroup
|
||||||
|
asyncio.run(asyncio.wait_for(pipe.run(), timeout=5))
|
||||||
|
|
||||||
|
assert closed["source"] is True, "perception source was not closed (orphaned loop)"
|
||||||
|
assert closed["stt"] is True, "stt was not closed"
|
||||||
|
|
||||||
|
|
||||||
def test_history_is_bounded():
|
def test_history_is_bounded():
|
||||||
pipe = Pipeline(
|
pipe = Pipeline(
|
||||||
source=MockFrameSource(limit=0),
|
source=MockFrameSource(limit=0),
|
||||||
|
|||||||
@@ -102,9 +102,16 @@ class Pipeline:
|
|||||||
|
|
||||||
# -- lifecycle --------------------------------------------------------- #
|
# -- lifecycle --------------------------------------------------------- #
|
||||||
async def run(self) -> None:
|
async def run(self) -> None:
|
||||||
loops = [self._perceive(), self._listen_voice(), self._listen_text()]
|
# A TaskGroup (not bare gather) so that if ONE loop raises, the others
|
||||||
|
# are cancelled and awaited before teardown. With plain gather the
|
||||||
|
# failing loop propagated while the siblings kept running detached, and
|
||||||
|
# aclose() in the finally then closed a source/stt out from under a
|
||||||
|
# still-live loop (close-during-use).
|
||||||
try:
|
try:
|
||||||
await asyncio.gather(*loops)
|
async with asyncio.TaskGroup() as tg:
|
||||||
|
tg.create_task(self._perceive())
|
||||||
|
tg.create_task(self._listen_voice())
|
||||||
|
tg.create_task(self._listen_text())
|
||||||
finally:
|
finally:
|
||||||
await self.aclose()
|
await self.aclose()
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user