diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py index 3822a78..145bb0e 100644 --- a/tests/test_pipeline.py +++ b/tests/test_pipeline.py @@ -3,6 +3,8 @@ into the brain's replies.""" import asyncio +import pytest + from wsai.backends.mock import ( MockBrain, MockFrameSource, @@ -10,6 +12,7 @@ from wsai.backends.mock import ( MockTTS, MockVision, ) +from wsai.interfaces import Frame from wsai.pipeline import Pipeline @@ -56,6 +59,44 @@ def test_voice_only_pipeline_runs_without_eyes(): 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(): pipe = Pipeline( source=MockFrameSource(limit=0), diff --git a/wsai/pipeline.py b/wsai/pipeline.py index a938137..e5e1ca0 100644 --- a/wsai/pipeline.py +++ b/wsai/pipeline.py @@ -102,9 +102,16 @@ class Pipeline: # -- lifecycle --------------------------------------------------------- # 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: - 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: await self.aclose()