merge branch develop

Author: deeleeramone

.github/workflows/test-pywry.yml

@@ -57,21 +57,18 @@ jobs:
       run:
         working-directory: ${{ github.workspace }}
     outputs:
-      matrix: ${{ steps.matrix.outputs.matrix }}
+      full-matrix: ${{ steps.matrix.outputs.full-matrix }}
+      pr-matrix: ${{ steps.matrix.outputs.pr-matrix }}
       pytest-targets: ${{ steps.targets.outputs.targets }}
+      is-scoped: ${{ steps.targets.outputs.is-scoped }}
     steps:
       - id: matrix
         shell: bash
-        env:
-          IS_PR: ${{ github.event_name == 'pull_request' }}
-          HAS_FULL_LABEL: ${{ contains(github.event.pull_request.labels.*.name, 'full-test') }}
         run: |
-          if [[ "$IS_PR" == "true" && "$HAS_FULL_LABEL" != "true" ]]; then
-            MATRIX='{"include":[{"os":"ubuntu-24.04","python-version":"3.14"},{"os":"windows-2025","python-version":"3.14"},{"os":"macos-15-intel","python-version":"3.14"},{"os":"ubuntu-24.04-arm","python-version":"3.14"},{"os":"windows-11-arm","python-version":"3.14"},{"os":"macos-latest","python-version":"3.14"}]}'
-          else
-            MATRIX='{"include":[{"os":"ubuntu-24.04","python-version":"3.11"},{"os":"ubuntu-24.04","python-version":"3.14"},{"os":"windows-2025","python-version":"3.11"},{"os":"windows-2025","python-version":"3.14"},{"os":"macos-15-intel","python-version":"3.11"},{"os":"macos-15-intel","python-version":"3.14"},{"os":"ubuntu-24.04-arm","python-version":"3.11"},{"os":"ubuntu-24.04-arm","python-version":"3.14"},{"os":"windows-11-arm","python-version":"3.12"},{"os":"windows-11-arm","python-version":"3.14"},{"os":"macos-latest","python-version":"3.11"},{"os":"macos-latest","python-version":"3.14"}]}'
-          fi
-          echo "matrix=$MATRIX" >> "$GITHUB_OUTPUT"
+          FULL_MATRIX='{"include":[{"os":"ubuntu-24.04","python-version":"3.11"},{"os":"ubuntu-24.04","python-version":"3.14"},{"os":"windows-2025","python-version":"3.11"},{"os":"windows-2025","python-version":"3.14"},{"os":"macos-15-intel","python-version":"3.11"},{"os":"macos-15-intel","python-version":"3.14"},{"os":"ubuntu-24.04-arm","python-version":"3.11"},{"os":"ubuntu-24.04-arm","python-version":"3.14"},{"os":"windows-11-arm","python-version":"3.12"},{"os":"windows-11-arm","python-version":"3.14"},{"os":"macos-latest","python-version":"3.11"},{"os":"macos-latest","python-version":"3.14"}]}'
+          PR_MATRIX='{"include":[{"os":"ubuntu-24.04","python-version":"3.14"}]}'
+          echo "full-matrix=$FULL_MATRIX" >> "$GITHUB_OUTPUT"
+          echo "pr-matrix=$PR_MATRIX" >> "$GITHUB_OUTPUT"
 
       - name: Checkout for diff
         if: github.event_name == 'pull_request' && !contains(github.event.pull_request.labels.*.name, 'full-test')
@@ -90,6 +87,7 @@ jobs:
           if [[ "$IS_PR" != "true" || "$HAS_FULL_LABEL" == "true" ]]; then
             echo "Full suite (dispatch or full-test label)"
             echo "targets=tests/" >> "$GITHUB_OUTPUT"
+            echo "is-scoped=false" >> "$GITHUB_OUTPUT"
             exit 0
           fi
 
@@ -107,13 +105,16 @@ jobs:
           if [[ -n "$SOURCE_NON_TEST" || -n "$WORKFLOW_SELF" || -n "$INFRA_TEST" ]]; then
             echo "Full suite (source/workflow/test-infra changes)"
             echo "targets=tests/" >> "$GITHUB_OUTPUT"
+            echo "is-scoped=false" >> "$GITHUB_OUTPUT"
           elif [[ -n "$TEST_FILES" ]]; then
             T=$(echo "$TEST_FILES" | sed 's|^pywry/||' | tr '\n' ' ' | sed 's/ *$//')
             echo "Scoped to changed test files: $T"
             echo "targets=$T" >> "$GITHUB_OUTPUT"
+            echo "is-scoped=true" >> "$GITHUB_OUTPUT"
           else
             echo "No pywry source/test changes; skipping pytest selection"
             echo "targets=__skip__" >> "$GITHUB_OUTPUT"
+            echo "is-scoped=false" >> "$GITHUB_OUTPUT"
           fi
 
   test:
@@ -130,7 +131,7 @@ jobs:
       # ``sqlcipher3`` is a source build against the SQLCipher C
       # library installed in the ``Build SQLCipher`` step below, so
       # any Python ABI works (no binary-wheel coverage gaps).
-      matrix: ${{ fromJSON(needs.gen-matrix.outputs.matrix) }}
+      matrix: ${{ fromJSON(github.event_name == 'pull_request' && !contains(github.event.pull_request.labels.*.name, 'full-test') ? needs.gen-matrix.outputs.pr-matrix : needs.gen-matrix.outputs.full-matrix) }}
 
     steps:
       - uses: actions/checkout@v5
@@ -274,7 +275,7 @@ jobs:
       - name: Install dependencies
         run: |
           python -m pip install --upgrade pip
-          pip install -e ".[dev]"
+          pip install -e ".[dev]" pytest-xdist
 
       - name: Verify Docker (Linux)
         if: runner.os == 'Linux'
@@ -327,7 +328,7 @@ jobs:
           NO_AT_BRIDGE: "1"
           PYTEST_TARGETS: ${{ needs.gen-matrix.outputs.pytest-targets }}
         run: |
-          dbus-run-session -- xvfb-run --auto-servernum --server-args="-screen 0 1920x1080x24" python -m pytest -c pytest.ini $PYTEST_TARGETS -v --tb=short
+          dbus-run-session -- xvfb-run --auto-servernum --server-args="-screen 0 1920x1080x24" python -m pytest -c pytest.ini $PYTEST_TARGETS -n auto -v --tb=short
           rc=$?
           # Exit 5 = no tests collected after platform filtering on a scoped run; treat as success.
           if [[ $rc -eq 5 && "$PYTEST_TARGETS" != "tests/" ]]; then rc=0; fi
@@ -342,7 +343,7 @@ jobs:
           PYTEST_TARGETS: ${{ needs.gen-matrix.outputs.pytest-targets }}
         shell: pwsh
         run: |
-          python -m pytest -c pytest.ini $env:PYTEST_TARGETS.Split(' ') -v --tb=short --ignore=tests/test_state_redis_integration.py --ignore=tests/test_auth_rbac_integration.py --ignore=tests/test_deploy_mode_integration.py --ignore=tests/test_e2e_deploy_mode.py --ignore=tests/test_e2e_rbac_widgets.py -m "not redis and not container"
+          python -m pytest -c pytest.ini $env:PYTEST_TARGETS.Split(' ') -n auto -v --tb=short --ignore=tests/test_state_redis_integration.py --ignore=tests/test_auth_rbac_integration.py --ignore=tests/test_deploy_mode_integration.py --ignore=tests/test_e2e_deploy_mode.py --ignore=tests/test_e2e_rbac_widgets.py -m "not redis and not container"
           if ($LASTEXITCODE -eq 5 -and $env:PYTEST_TARGETS -ne 'tests/') { exit 0 }
           exit $LASTEXITCODE
 
@@ -355,7 +356,7 @@ jobs:
           PYTEST_TARGETS: ${{ needs.gen-matrix.outputs.pytest-targets }}
         shell: pwsh
         run: |
-          python -m pytest -c pytest.ini $env:PYTEST_TARGETS.Split(' ') -v --tb=short --ignore=tests/test_state_redis_integration.py --ignore=tests/test_auth_rbac_integration.py --ignore=tests/test_deploy_mode_integration.py --ignore=tests/test_e2e_deploy_mode.py --ignore=tests/test_e2e_rbac_widgets.py --ignore=tests/test_inline_ssl.py -m "not redis and not container"
+          python -m pytest -c pytest.ini $env:PYTEST_TARGETS.Split(' ') -n auto -v --tb=short --ignore=tests/test_state_redis_integration.py --ignore=tests/test_auth_rbac_integration.py --ignore=tests/test_deploy_mode_integration.py --ignore=tests/test_e2e_deploy_mode.py --ignore=tests/test_e2e_rbac_widgets.py --ignore=tests/test_inline_ssl.py -m "not redis and not container"
           if ($LASTEXITCODE -eq 5 -and $env:PYTEST_TARGETS -ne 'tests/') { exit 0 }
           exit $LASTEXITCODE
 
@@ -365,7 +366,7 @@ jobs:
           PYWRY_HEADLESS: "1"
           PYTEST_TARGETS: ${{ needs.gen-matrix.outputs.pytest-targets }}
         run: |
-          python -m pytest -c pytest.ini $PYTEST_TARGETS -v --tb=short
+          python -m pytest -c pytest.ini $PYTEST_TARGETS -n auto -v --tb=short
           rc=$?
           if [[ $rc -eq 5 && "$PYTEST_TARGETS" != "tests/" ]]; then rc=0; fi
           exit $rc
@@ -377,7 +378,7 @@ jobs:
           PYWRY_DEPLOY__STATE_BACKEND: "memory"
           PYTEST_TARGETS: ${{ needs.gen-matrix.outputs.pytest-targets }}
         run: |
-          python -m pytest -c pytest.ini $PYTEST_TARGETS -v --tb=short --ignore=tests/test_state_redis_integration.py --ignore=tests/test_auth_rbac_integration.py --ignore=tests/test_deploy_mode_integration.py --ignore=tests/test_e2e_deploy_mode.py --ignore=tests/test_e2e_rbac_widgets.py -m "not redis and not container"
+          python -m pytest -c pytest.ini $PYTEST_TARGETS -n auto -v --tb=short --ignore=tests/test_state_redis_integration.py --ignore=tests/test_auth_rbac_integration.py --ignore=tests/test_deploy_mode_integration.py --ignore=tests/test_e2e_deploy_mode.py --ignore=tests/test_e2e_rbac_widgets.py -m "not redis and not container"
           rc=$?
           if [[ $rc -eq 5 && "$PYTEST_TARGETS" != "tests/" ]]; then rc=0; fi
           exit $rc

pywry/pywry/chat/manager.py

@@ -1160,23 +1160,26 @@ async def _handle_async_stream(
         """Stream from an async generator."""
         state = _StreamState(message_id)
         typing_hidden = False
-        async for item in agen:
+        try:
+            async for item in agen:
+                if not typing_hidden:
+                    typing_hidden = True
+                    self._emit(
+                        "chat:typing-indicator",
+                        {"typing": False, "threadId": thread_id},
+                    )
+                if cancel.is_set():
+                    self._handle_cancel(state, thread_id)
+                    return
+                self._process_handler_item(item, state, thread_id, ctx)
             if not typing_hidden:
-                typing_hidden = True
                 self._emit(
                     "chat:typing-indicator",
                     {"typing": False, "threadId": thread_id},
                 )
-            if cancel.is_set():
-                self._handle_cancel(state, thread_id)
-                return
-            self._process_handler_item(item, state, thread_id, ctx)
-        if not typing_hidden:
-            self._emit(
-                "chat:typing-indicator",
-                {"typing": False, "threadId": thread_id},
-            )
-        self._finalize_stream(state, thread_id)
+            self._finalize_stream(state, thread_id)
+        finally:
+            await agen.aclose()
 
     def _inject_context(
         self,

pywry/pywry/chat/providers/anthropic.py

@@ -137,10 +137,13 @@ async def prompt(
             temperature=session.get("temperature", 0.7),
             max_tokens=session.get("max_tokens", 4096),
         ) as stream:
-            async for text in stream.text_stream:
-                if cancel_event and cancel_event.is_set():
-                    raise GenerationCancelledError()
-                yield AgentMessageUpdate(text=text)
+            try:
+                async for text in stream.text_stream:
+                    if cancel_event and cancel_event.is_set():
+                        raise GenerationCancelledError()
+                    yield AgentMessageUpdate(text=text)
+            finally:
+                await stream.text_stream.aclose()
 
     async def cancel(self, session_id: str) -> None:
         """Cancel is handled cooperatively via ``cancel_event``.

pywry/pywry/chat/providers/callback.py

@@ -143,10 +143,14 @@ def _wrap(item: Any) -> Any:
             return AgentMessageUpdate(text=item) if isinstance(item, str) else item
 
         if hasattr(result, "__aiter__"):
-            async for item in result:
-                if cancel_event and cancel_event.is_set():
-                    raise GenerationCancelledError()
-                yield _wrap(item)
+            try:
+                async for item in result:
+                    if cancel_event and cancel_event.is_set():
+                        raise GenerationCancelledError()
+                    yield _wrap(item)
+            finally:
+                if hasattr(result, "aclose"):
+                    await result.aclose()
         elif hasattr(result, "__iter__"):
             for item in result:
                 if cancel_event and cancel_event.is_set():

pywry/pywry/chat/providers/magentic.py

@@ -150,10 +150,14 @@ async def prompt(
             output_types=[AsyncStreamedStr],
         )
         chat = await chat.asubmit()
-        async for chunk in chat.last_message.content:
-            if cancel_event and cancel_event.is_set():
-                raise GenerationCancelledError()
-            yield AgentMessageUpdate(text=chunk)
+        try:
+            async for chunk in chat.last_message.content:
+                if cancel_event and cancel_event.is_set():
+                    raise GenerationCancelledError()
+                yield AgentMessageUpdate(text=chunk)
+        finally:
+            if hasattr(chat.last_message.content, "aclose"):
+                await chat.last_message.content.aclose()
 
     async def cancel(self, session_id: str) -> None:
         """Cancel is handled cooperatively via ``cancel_event``.

pywry/pywry/state/redis.py

@@ -310,8 +310,10 @@ async def subscribe(self, channel: str) -> AsyncIterator[EventMessage]:
         pubsub = r.pubsub()
         await pubsub.subscribe(self._channel_name(channel))
 
+        listen_gen = None
         try:
-            async for message in pubsub.listen():
+            listen_gen = pubsub.listen()
+            async for message in listen_gen:
                 if message["type"] == "message":
                     try:
                         data = json.loads(message["data"])
@@ -327,6 +329,8 @@ async def subscribe(self, channel: str) -> AsyncIterator[EventMessage]:
                     except json.JSONDecodeError:
                         continue
         finally:
+            if listen_gen:
+                await listen_gen.aclose()
             await pubsub.unsubscribe(self._channel_name(channel))
             await pubsub.close()
 

pywry/tests/test_chat_providers_anthropic.py

@@ -0,0 +1,160 @@
+"""Tests for the Anthropic chat provider.
+
+Source: ``pywry/chat/providers/anthropic.py``.
+
+The Anthropic client is mocked via ``unittest.mock`` so tests run without
+network or an installed ``anthropic`` package.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import builtins
+import sys
+
+from unittest.mock import MagicMock
+
+import pytest
+
+from pywry.chat.models import GenerationCancelledError, TextPart
+from pywry.chat.session import ClientCapabilities
+from pywry.chat.updates import AgentMessageUpdate
+
+
+# =============================================================================
+# Anthropic client surrogate
+# =============================================================================
+
+
+class _FakeAnthStream:
+    """Async-context-manager + async-iterable text stream surrogate."""
+
+    def __init__(self, chunks: list[str]):
+        self._chunks = chunks
+        self._gen = None
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, exc_type, exc, tb):
+        if self._gen:
+            await self._gen.aclose()
+        return False
+
+    @property
+    def text_stream(self):
+        async def _gen():
+            for c in self._chunks:
+                yield c
+
+        if self._gen is None:
+            self._gen = _gen()
+        return self._gen
+
+
+def _make_anthropic_client(chunks: list[str]) -> MagicMock:
+    """Build a fake AsyncAnthropic whose ``.messages.stream`` returns a context-manager."""
+    client = MagicMock()
+    stream_obj = _FakeAnthStream(chunks)
+
+    def _stream(**_kwargs):
+        return stream_obj
+
+    client.messages.stream = MagicMock(side_effect=_stream)
+    return client
+
+
+@pytest.fixture
+def anthropic_module(monkeypatch):
+    """Install a fake ``anthropic`` module whose AsyncAnthropic returns an
+    empty MagicMock — useful for tests that exercise paths that don't drive
+    the client."""
+    fake_module = MagicMock()
+    fake_module.AsyncAnthropic = lambda **_kwargs: MagicMock()
+    monkeypatch.setitem(sys.modules, "anthropic", fake_module)
+    return fake_module
+
+
+# =============================================================================
+# Tests
+# =============================================================================
+
+
+class TestAnthropicProvider:
+    def test_import_error_when_anthropic_missing(self, monkeypatch):
+        real_import = builtins.__import__
+
+        def fake_import(name, *args, **kwargs):
+            if name == "anthropic":
+                raise ImportError("no anthropic")
+            return real_import(name, *args, **kwargs)
+
+        monkeypatch.setattr(builtins, "__import__", fake_import)
+        from pywry.chat.providers.anthropic import AnthropicProvider
+
+        with pytest.raises(ImportError, match="pywry\\[anthropic\\]"):
+            AnthropicProvider(api_key="x")
+
+    def test_constructor_creates_client(self, monkeypatch):
+        fake_client = MagicMock()
+        fake_module = MagicMock()
+        fake_module.AsyncAnthropic = lambda **_kwargs: fake_client
+        monkeypatch.setitem(sys.modules, "anthropic", fake_module)
+        from pywry.chat.providers.anthropic import AnthropicProvider
+
+        provider = AnthropicProvider(api_key="sk-test")
+        assert provider._client is fake_client
+        assert provider._sessions == {}
+
+    async def test_initialize_returns_image_capabilities(self, anthropic_module):
+        from pywry.chat.providers.anthropic import AnthropicProvider
+
+        provider = AnthropicProvider()
+        caps = await provider.initialize(ClientCapabilities())
+        assert caps.prompt_capabilities is not None
+        assert caps.prompt_capabilities.image is True
+
+    async def test_new_session_returns_id_with_prefix(self, anthropic_module):
+        from pywry.chat.providers.anthropic import AnthropicProvider
+
+        provider = AnthropicProvider()
+        sid = await provider.new_session("/tmp")
+        assert sid.startswith("ant_")
+        assert provider._sessions[sid]["model"] == "claude-sonnet-4-20250514"
+
+    async def test_prompt_yields_agent_message_updates(self, monkeypatch):
+        client = _make_anthropic_client(["Hello", " world"])
+        fake_module = MagicMock()
+        fake_module.AsyncAnthropic = lambda **_kwargs: client
+        monkeypatch.setitem(sys.modules, "anthropic", fake_module)
+        from pywry.chat.providers.anthropic import AnthropicProvider
+
+        provider = AnthropicProvider()
+        sid = await provider.new_session("/tmp")
+        updates = []
+        async for u in provider.prompt(sid, [TextPart(text="hi")]):
+            updates.append(u)
+        assert [u.text for u in updates] == ["Hello", " world"]
+        assert all(isinstance(u, AgentMessageUpdate) for u in updates)
+
+    async def test_prompt_respects_cancel_event(self, monkeypatch):
+        client = _make_anthropic_client(["a", "b", "c"])
+        fake_module = MagicMock()
+        fake_module.AsyncAnthropic = lambda **_kwargs: client
+        monkeypatch.setitem(sys.modules, "anthropic", fake_module)
+        from pywry.chat.providers.anthropic import AnthropicProvider
+
+        provider = AnthropicProvider()
+        sid = await provider.new_session("/tmp")
+        cancel = asyncio.Event()
+        cancel.set()
+
+        with pytest.raises(GenerationCancelledError):
+            async for _ in provider.prompt(sid, [TextPart(text="hi")], cancel_event=cancel):
+                pass
+
+    async def test_cancel_is_noop(self, anthropic_module):
+        from pywry.chat.providers.anthropic import AnthropicProvider
+
+        provider = AnthropicProvider()
+        assert await provider.cancel("any-session") is None