[Bugfix] Fix DPM replies timeout task context

.github/WORKFLOW_TASKS/issue-2-bugfix-dpm-replies-timeout-task-context.md

@@ -0,0 +1,61 @@
+# Issue 2: [Bugfix] Resolve `RuntimeError('Timeout should be used inside a task')` in `acsys.dpm.replies()` on Python 3.11+
+
+**Status:** Implemented on branch `fix/dpm-replies-task-context`; full pytest suite passes locally. Draft PR pending.
+
+## Summary
+
+Make `acsys.dpm.DPM.replies()` usable when its async generator is driven through an event loop on Python 3.11 and later, without exposing the `asyncio.wait_for()` task-context runtime error.
+
+## Reproduction and current evidence
+
+`tests/test_dpm_task_context.py` reproduces the problem in the RED phase by:
+
+1. Creating a bare `DPM` instance with a stubbed `__anext__`.
+2. Creating `replies_gen = dpm.replies(tmo=1.0)`.
+3. Driving the async generator once with `.send(None)` while marking a bare event loop as running, without wrapping the generator in an active task.
+4. Asserting that Python 3.11+ raises `RuntimeError` matching `Timeout should be used inside a task`.
+
+The GREEN regression test continues the manually driven generator after the fix and verifies that the first yielded object is a task and that the reply is returned successfully.
+
+The relevant implementation is `DPM.replies()` in `acsys/dpm/__init__.py`, where each reply is awaited through `asyncio.wait_for(self.__anext__(), tmo)`.
+
+## Scope
+
+- Diagnose the supported invocation patterns for `DPM.replies()` and the Python 3.11+ `asyncio` requirement involved in the failure.
+- Change the implementation and/or surrounding async boundary so the public replies API no longer fails solely because the caller drives the async generator without an existing task.
+- Preserve timeout semantics: `tmo` remains the maximum interval between replies and an `asyncio.TimeoutError` remains the expected timeout signal.
+- Replace the current regression test's failure expectation with a success/behavioral regression test that covers the supported calling pattern.
+
+## Acceptance criteria
+
+- The reproducer no longer raises `RuntimeError('Timeout should be used inside a task')` on Python 3.11+.
+- Iterating `DPM.replies(tmo=...)` from a normal task continues to yield replies as before.
+- A timeout between replies still raises `asyncio.TimeoutError` with the documented behavior.
+- Cancellation and generator cleanup do not leave pending tasks or unhandled coroutine warnings.
+- The regression coverage runs under the repository's supported Python versions, or any version-specific limitation is explicitly documented.
+- No unrelated changes to DPM request/reconnection behavior are introduced.
+
+## Verification
+
+- RED phase: the `.send(None)` reproducer failed before the implementation change with `RuntimeError('Timeout should be used inside a task')`.
+- GREEN phase: `python -m pytest -q tests/test_dpm_task_context.py -W error::RuntimeWarning` passed.
+- Full suite: `python -m pytest -q` passed (`1 passed`).
+- Focused regression test passed with runtime warnings treated as errors.
+- The fix creates a task for `self.__anext__()` and a task for the `asyncio.wait_for()` coroutine, keeping Python 3.11+ timeout context execution inside a task.
+
+## Dependencies and risks
+
+- Issue 1 should establish the pytest/pytest-asyncio configuration before this issue is implemented.
+- The exact repair must preserve the caller's event-loop ownership and avoid silently creating a competing loop.
+- The test's current bare-object setup bypasses normal `DPM` initialization; implementation should also be validated through the real public setup path where practical.
+
+## Review focus
+
+- Whether the fix addresses the task-context boundary rather than masking the exception.
+- Whether timeout, cancellation, and async-generator lifecycle semantics remain correct.
+- Whether tests assert observable API behavior instead of depending only on a particular `asyncio` implementation detail.
+
+## Out of scope
+
+- Replacing the synchronization primitives used by `set_many()` (Issue 3).
+- General modernization of all `asyncio.get_event_loop()` calls in the package.

.gitignore

@@ -140,6 +140,5 @@ cython_debug/
 
 # Application specific
 acsys.tgz
-tests/
 .python-version
 .vscode

acsys/dpm/__init__.py

@@ -411,7 +411,10 @@ async def replies(self, tmo=None):
 
         """
         while True:
-            ii = await asyncio.wait_for(self.__anext__(), tmo)
+            reply_task = asyncio.create_task(self.__anext__())
+            timeout_task = asyncio.create_task(
+                asyncio.wait_for(reply_task, tmo))
+            ii = await timeout_task
             if ii is None:
                 return
             yield ii

tests/test_dpm_task_context.py

@@ -0,0 +1,60 @@
+import asyncio
+import pytest
+import acsys.dpm
+
+
+def _new_loop():
+    """Create a test loop without leaking the loop created during import."""
+    try:
+        current_loop = asyncio.get_event_loop()
+    except RuntimeError:
+        current_loop = None
+    if current_loop is not None and not current_loop.is_running():
+        current_loop.close()
+
+    loop = asyncio.new_event_loop()
+    asyncio.set_event_loop(loop)
+    return loop
+
+
+def _drive_once_without_task(coro, loop):
+    """Advance a coroutine once without wrapping it in an asyncio.Task."""
+    asyncio.events._set_running_loop(loop)
+    try:
+        return coro.send(None)
+    finally:
+        asyncio.events._set_running_loop(None)
+
+
+def _close_loop(loop):
+    pending = asyncio.all_tasks(loop)
+    for task in pending:
+        task.cancel()
+    if pending:
+        loop.run_until_complete(asyncio.gather(*pending, return_exceptions=True))
+    asyncio.set_event_loop(None)
+    loop.close()
+
+
+def test_dpm_replies_timeout_uses_task_context():
+    """DPM.replies() can start without an existing asyncio.Task."""
+    dpm = object.__new__(acsys.dpm.DPM)
+
+    async def dummy_anext():
+        await asyncio.sleep(0.01)
+        return "data"
+
+    dpm.__anext__ = dummy_anext
+    loop = _new_loop()
+    coro = dpm.replies(tmo=1.0).__anext__()
+
+    try:
+        timeout_task = _drive_once_without_task(coro, loop)
+        assert isinstance(timeout_task, asyncio.Task)
+        assert loop.run_until_complete(timeout_task) == "data"
+        with pytest.raises(StopIteration) as stopped:
+            coro.send(None)
+        assert stopped.value.value == "data"
+    finally:
+        coro.close()
+        _close_loop(loop)