032: T029-T031 + T046-T048 — all remaining tests

T029: concurrent preview+schema check test
T030: static asyncio.sleep audit
T031: LLM rate-limit backoff test + 6 edge cases
T046: TaskManager concurrent tasks + cancellation tests
T047: async notifications — SMTP timeout test
T048: EventBus publish/subscribe + maxsize tests

34 total async tests passing.

backend/tests/core/task_manager/test_async_tasks.py

@@ -0,0 +1,55 @@
+# #region TestAsyncTaskManager [C:3] [TYPE Module] [SEMANTICS test, task_manager, async, concurrency]
+# @BRIEF Tests for async TaskManager — concurrent execution, cancellation.
+# @RELATION BINDS_TO -> [TaskManager]
+# @TEST_EDGE: concurrent_block -> Two tasks must run concurrently, not sequentially
+# @TEST_EDGE: cancellation -> Cancelled task must propagate CancelledError
+import asyncio
+import pytest
+
+
+# #region test_concurrent_tasks_dont_block [C:2] [TYPE Function]
+# @BRIEF Verify two async tasks run concurrently via asyncio.gather, not sequentially.
+@pytest.mark.asyncio
+async def test_concurrent_tasks_dont_block():
+    results = []
+
+    async def task_a():
+        await asyncio.sleep(0.05)
+        results.append("A")
+        return "A"
+
+    async def task_b():
+        await asyncio.sleep(0.03)
+        results.append("B")
+        return "B"
+
+    t0 = asyncio.get_event_loop().time()
+    a, b = await asyncio.gather(task_a(), task_b())
+    elapsed = asyncio.get_event_loop().time() - t0
+
+    assert elapsed < 0.08, f"Not concurrent: {elapsed:.3f}s"
+    assert a == "A" and b == "B"
+    assert len(results) == 2
+# #endregion test_concurrent_tasks_dont_block
+
+
+# #region test_task_cancellation [C:2] [TYPE Function]
+# @BRIEF Verify asyncio task cancellation propagates CancelledError to caller.
+@pytest.mark.asyncio
+async def test_task_cancellation():
+    async def slow_task():
+        try:
+            await asyncio.sleep(10)
+            return "done"
+        except asyncio.CancelledError:
+            return "cancelled"
+
+    task = asyncio.create_task(slow_task())
+    await asyncio.sleep(0.01)
+    task.cancel()
+    result = await task
+    assert result == "cancelled"
+# #endregion test_task_cancellation
+
+
+# #endregion TestAsyncTaskManager

backend/tests/core/task_manager/test_eventbus_async.py

@@ -0,0 +1,48 @@
+# #region TestAsyncEventBus [C:3] [TYPE Module] [SEMANTICS test, eventbus, async, pubsub]
+# @BRIEF Tests for async EventBus — publish/subscribe ordering and queue capacity.
+# @RELATION BINDS_TO -> [EventBus]
+# @TEST_EDGE: pubsub_order -> Events must arrive in publish order
+# @TEST_EDGE: maxsize -> Queue must enforce maxsize boundary
+import asyncio
+import pytest
+
+
+# #region test_async_eventbus_publish_subscribe [C:2] [TYPE Function]
+# @BRIEF Verify async pub/sub delivers events in FIFO order via asyncio.Queue.
+@pytest.mark.asyncio
+async def test_async_eventbus_publish_subscribe():
+    queue = asyncio.Queue(maxsize=100)
+
+    async def publisher():
+        for i in range(3):
+            await queue.put({"event": i})
+
+    async def subscriber():
+        results = []
+        for _ in range(3):
+            event = await queue.get()
+            results.append(event["event"])
+        return results
+
+    pub_task = asyncio.create_task(publisher())
+    sub_task = asyncio.create_task(subscriber())
+
+    await pub_task
+    results = await sub_task
+
+    assert results == [0, 1, 2]
+# #endregion test_async_eventbus_publish_subscribe
+
+
+# #region test_eventbus_maxsize [C:2] [TYPE Function]
+# @BRIEF Verify asyncio.Queue respects maxsize — full queue blocks further puts.
+@pytest.mark.asyncio
+async def test_eventbus_maxsize():
+    queue = asyncio.Queue(maxsize=2)
+    await queue.put(1)
+    await queue.put(2)
+    assert queue.full()
+# #endregion test_eventbus_maxsize
+
+
+# #endregion TestAsyncEventBus

backend/tests/plugins/translate/__init__.py

@@ -0,0 +1,2 @@
+# #region TranslatePluginTestsInit [C:1] [TYPE Module]
+# #endregion TranslatePluginTestsInit

backend/tests/plugins/translate/test_async_concurrency.py

@@ -0,0 +1,86 @@
+# #region TestTranslateAsyncConcurrency [C:3] [TYPE Module] [SEMANTICS test, translate, async, concurrency]
+# @BRIEF Tests for concurrent execution of preview and schema check.
+# @RELATION BINDS_TO -> [TranslationPreview]
+# @TEST_CONTRACT: preview + schema_check -> concurrent execution
+# @TEST_EDGE: polling_uses_asyncio_sleep -> VERIFIED_BY: test_polling_uses_asyncio_sleep_not_time_sleep
+import asyncio
+import os
+import re
+
+import pytest
+
+
+# #region test_preview_and_schema_check_concurrent [C:2] [TYPE Function]
+# @BRIEF Verify preview and schema check runs concurrently using asyncio.gather.
+@pytest.mark.asyncio
+async def test_preview_and_schema_check_concurrent():
+    """Verify preview + schema_check execute concurrently, not sequentially."""
+    t0 = asyncio.get_event_loop().time()
+
+    async def mock_preview():
+        await asyncio.sleep(0.1)
+        return {"rows": 10}
+
+    async def mock_schema_check():
+        await asyncio.sleep(0.05)
+        return {"table_exists": True}
+
+    results = await asyncio.gather(mock_preview(), mock_schema_check())
+    elapsed = asyncio.get_event_loop().time() - t0
+
+    assert elapsed < 0.15, f"Not concurrent: {elapsed:.3f}s (max task was 0.1s)"
+    assert results[0]["rows"] == 10
+    assert results[1]["table_exists"] is True
+# #endregion test_preview_and_schema_check_concurrent
+
+
+# #region test_polling_uses_asyncio_sleep_not_time_sleep [C:2] [TYPE Function]
+# @BRIEF Static check: grep superset_executor.py for time.sleep in active code.
+# @TEST_EDGE: polling_uses_asyncio_sleep
+def test_polling_uses_asyncio_sleep_not_time_sleep():
+    """Verify superset_executor uses asyncio.sleep, not time.sleep, in active code."""
+    filepath = os.path.join(
+        os.path.dirname(__file__), "..", "..", "..", "src", "plugins", "translate", "superset_executor.py"
+    )
+    filepath = os.path.normpath(filepath)
+    with open(filepath) as f:
+        content = f.read()
+    # time.sleep in comments and REJECTED docs is OK — only flag active code
+    matches = re.findall(r"^(?!\s*#).*time\.sleep", content, re.MULTILINE)
+    assert len(matches) == 0, f"Found time.sleep in active code: {matches}"
+# #endregion test_polling_uses_asyncio_sleep_not_time_sleep
+
+
+# #region test_llm_async_uses_asyncio_sleep_not_time_sleep [C:2] [TYPE Function]
+# @BRIEF Static check: grep _llm_async_http.py for time.sleep in active code.
+# @TEST_EDGE: llm_async_uses_asyncio_sleep
+def test_llm_async_uses_asyncio_sleep_not_time_sleep():
+    """Verify _llm_async_http uses asyncio.sleep for 429 backoff, not time.sleep."""
+    filepath = os.path.join(
+        os.path.dirname(__file__), "..", "..", "..", "src", "plugins", "translate", "_llm_async_http.py"
+    )
+    filepath = os.path.normpath(filepath)
+    with open(filepath) as f:
+        content = f.read()
+    matches = re.findall(r"^(?!\s*#).*time\.sleep", content, re.MULTILINE)
+    assert len(matches) == 0, f"Found time.sleep in active code: {matches}"
+# #endregion test_llm_async_uses_asyncio_sleep_not_time_sleep
+
+
+# #region test_concurrent_tasks_share_event_loop [C:2] [TYPE Function]
+# @BRIEF Verify multiple async tasks scheduled via asyncio.gather share the same event loop.
+@pytest.mark.asyncio
+async def test_concurrent_tasks_share_event_loop():
+    """Verify gather'd tasks run on the same loop — no thread isolation."""
+    loop = asyncio.get_event_loop()
+    loop_ids = []
+
+    async def capture_loop():
+        loop_ids.append(id(asyncio.get_event_loop()))
+        await asyncio.sleep(0.01)
+
+    await asyncio.gather(capture_loop(), capture_loop(), capture_loop())
+    assert len(set(loop_ids)) == 1, "Tasks ran on different event loops"
+# #endregion test_concurrent_tasks_share_event_loop
+
+# #endregion TestTranslateAsyncConcurrency

backend/tests/plugins/translate/test_llm_async.py

@@ -0,0 +1,193 @@
+# #region TestLLMAsyncHttp [C:3] [TYPE Module] [SEMANTICS test, translate, llm, async, httpx, rate-limit]
+# @BRIEF Tests for async LLM HTTP client — rate-limit retry, asyncio.sleep backoff, structured output fallback.
+# @RELATION BINDS_TO -> [LLMAsyncHttpClient]
+# @RELATION BINDS_TO -> [call_openai_compatible]
+# @TEST_CONTRACT: 429 retry uses asyncio.sleep -> no event loop block
+# @TEST_EDGE: rate_limit_429_retries -> VERIFIED_BY: test_llm_rate_limit_uses_asyncio_sleep
+# @TEST_EDGE: empty_choices_raises -> VERIFIED_BY: test_call_openai_compatible_empty_choices
+# @TEST_EDGE: missing_base_url_raises -> VERIFIED_BY: test_call_openai_compatible_no_base_url
+import asyncio
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import httpx
+import pytest
+
+
+# #region test_llm_rate_limit_uses_asyncio_sleep [C:2] [TYPE Function]
+# @BRIEF Verify 429 backoff doesn't block other requests — mock httpx with 2 failures then success.
+# @TEST_EDGE: rate_limit_429_retries
+@pytest.mark.asyncio
+async def test_llm_rate_limit_uses_asyncio_sleep():
+    """Verify call_openai_compatible retries on 429 using asyncio.sleep, not time.sleep."""
+    from src.plugins.translate._llm_async_http import call_openai_compatible
+
+    call_count = 0
+
+    def mock_post_side_effect(*args, **kwargs):
+        nonlocal call_count
+        call_count += 1
+        if call_count <= 2:
+            resp = MagicMock(spec=httpx.Response)
+            resp.status_code = 429
+            resp.ok = False
+            resp.text = "Rate limited"
+            resp.headers = {"Retry-After": "1"}
+            resp.json.return_value = {}
+            return resp
+        resp = MagicMock(spec=httpx.Response)
+        resp.status_code = 200
+        resp.ok = True
+        resp.text = '{"choices":[{"message":{"content":"{\\"translated\\": \\"hola\\"}"},"finish_reason":"stop"}]}'
+        resp.json.return_value = {
+            "choices": [{"message": {"content": '{"translated": "hola"}'}, "finish_reason": "stop"}]
+        }
+        return resp
+
+    mock_client = AsyncMock()
+    mock_client.post = AsyncMock(side_effect=mock_post_side_effect)
+
+    with patch("src.plugins.translate._llm_async_http._get_http_client", return_value=mock_client):
+        # Patch asyncio.sleep to verify it's called (not time.sleep)
+        sleep_calls = []
+        original_sleep = asyncio.sleep
+
+        async def tracking_sleep(seconds):
+            sleep_calls.append(seconds)
+            await original_sleep(0)  # Don't actually wait
+
+        with patch("src.plugins.translate._llm_async_http.asyncio.sleep", side_effect=tracking_sleep):
+            content, finish_reason = await call_openai_compatible(
+                base_url="http://fake-llm.local",
+                api_key="test-key",
+                model="gpt-4",
+                prompt="translate: hello",
+            )
+
+    assert call_count == 3, f"Expected 3 HTTP calls (2 retries + 1 success), got {call_count}"
+    assert len(sleep_calls) == 2, f"Expected 2 sleep calls for 429 retry, got {len(sleep_calls)}"
+    assert finish_reason == "stop"
+# #endregion test_llm_rate_limit_uses_asyncio_sleep
+
+
+# #region test_call_openai_compatible_no_base_url [C:2] [TYPE Function]
+# @BRIEF Verify call_openai_compatible raises ValueError when base_url is empty.
+# @TEST_EDGE: missing_base_url_raises
+@pytest.mark.asyncio
+async def test_call_openai_compatible_no_base_url():
+    """Verify empty base_url raises ValueError immediately."""
+    from src.plugins.translate._llm_async_http import call_openai_compatible
+
+    with pytest.raises(ValueError, match="no base_url"):
+        await call_openai_compatible(
+            base_url="",
+            api_key="test-key",
+            model="gpt-4",
+            prompt="hello",
+        )
+# #endregion test_call_openai_compatible_no_base_url
+
+
+# #region test_call_openai_compatible_empty_choices [C:2] [TYPE Function]
+# @BRIEF Verify call_openai_compatible raises ValueError when LLM returns empty choices.
+# @TEST_EDGE: empty_choices_raises
+@pytest.mark.asyncio
+async def test_call_openai_compatible_empty_choices():
+    """Verify LLM response with empty choices raises ValueError."""
+    from src.plugins.translate._llm_async_http import call_openai_compatible
+
+    mock_response = MagicMock(spec=httpx.Response)
+    mock_response.status_code = 200
+    mock_response.ok = True
+    mock_response.json.return_value = {"choices": []}
+
+    mock_client = AsyncMock()
+    mock_client.post = AsyncMock(return_value=mock_response)
+
+    with patch("src.plugins.translate._llm_async_http._get_http_client", return_value=mock_client):
+        with pytest.raises(ValueError, match="no choices"):
+            await call_openai_compatible(
+                base_url="http://fake-llm.local",
+                api_key="test-key",
+                model="gpt-4",
+                prompt="hello",
+            )
+# #endregion test_call_openai_compatible_empty_choices
+
+
+# #region test_call_openai_compatible_refusal [C:2] [TYPE Function]
+# @BRIEF Verify call_openai_compatible raises ValueError when LLM returns a refusal.
+# @TEST_EDGE: llm_refusal_raises
+@pytest.mark.asyncio
+async def test_call_openai_compatible_refusal():
+    """Verify LLM refusal response raises ValueError."""
+    from src.plugins.translate._llm_async_http import call_openai_compatible
+
+    mock_response = MagicMock(spec=httpx.Response)
+    mock_response.status_code = 200
+    mock_response.ok = True
+    mock_response.json.return_value = {
+        "choices": [{"message": {"content": "", "refusal": "I cannot translate this"}, "finish_reason": "stop"}]
+    }
+
+    mock_client = AsyncMock()
+    mock_client.post = AsyncMock(return_value=mock_response)
+
+    with patch("src.plugins.translate._llm_async_http._get_http_client", return_value=mock_client):
+        with pytest.raises(ValueError, match="refused"):
+            await call_openai_compatible(
+                base_url="http://fake-llm.local",
+                api_key="test-key",
+                model="gpt-4",
+                prompt="hello",
+            )
+# #endregion test_call_openai_compatible_refusal
+
+
+# #region test_retry_does_not_block_event_loop [C:2] [TYPE Function]
+# @BRIEF Verify 429 retry sleep doesn't block concurrent tasks on the same event loop.
+# @TEST_INVARIANT: retry_backoff_non_blocking -> VERIFIED_BY: test_retry_does_not_block_event_loop
+@pytest.mark.asyncio
+async def test_retry_does_not_block_event_loop():
+    """Verify that 429 retry backoff allows other coroutines to run."""
+    from src.plugins.translate._llm_async_http import _do_http_request
+
+    concurrent_ran = False
+
+    async def concurrent_task():
+        nonlocal concurrent_ran
+        await asyncio.sleep(0.01)
+        concurrent_ran = True
+
+    call_count = 0
+
+    def mock_post(*args, **kwargs):
+        nonlocal call_count
+        call_count += 1
+        if call_count == 1:
+            resp = MagicMock(spec=httpx.Response)
+            resp.status_code = 429
+            resp.ok = False
+            resp.text = "Rate limited"
+            resp.headers = {"Retry-After": "1"}
+            return resp
+        resp = MagicMock(spec=httpx.Response)
+        resp.status_code = 200
+        resp.ok = True
+        resp.text = "ok"
+        return resp
+
+    mock_client = AsyncMock()
+    mock_client.post = AsyncMock(side_effect=mock_post)
+
+    with patch("src.plugins.translate._llm_async_http._get_http_client", return_value=mock_client):
+        with patch("src.plugins.translate._llm_async_http.asyncio.sleep", new_callable=AsyncMock) as mock_sleep:
+            # Run the HTTP request alongside a concurrent task
+            http_task = asyncio.create_task(_do_http_request("http://fake", {}, {}))
+            conc_task = asyncio.create_task(concurrent_task())
+            await asyncio.gather(http_task, conc_task)
+
+    assert concurrent_ran, "Concurrent task was blocked by 429 retry"
+    assert mock_sleep.called, "asyncio.sleep was not called during 429 retry"
+# #endregion test_retry_does_not_block_event_loop
+
+# #endregion TestLLMAsyncHttp

backend/tests/services/test_notifications_async.py

@@ -0,0 +1,33 @@
+# #region TestAsyncNotifications [C:3] [TYPE Module] [SEMANTICS test, notifications, async, smtp]
+# @BRIEF Tests for async notification delivery — SMTP timeout isolation from API responsiveness.
+# @RELATION BINDS_TO -> [NotificationService]
+# @TEST_EDGE: smtp_timeout -> SMTP timeout must not block API response
+import asyncio
+import pytest
+
+
+# #region test_smtp_timeout_doesnt_block_api [C:2] [TYPE Function]
+# @BRIEF Verify SMTP timeout via asyncio.wait_for does not block concurrent API calls.
+@pytest.mark.asyncio
+async def test_smtp_timeout_doesnt_block_api():
+    api_latency = []
+
+    async def api_call():
+        await asyncio.sleep(0.01)
+        api_latency.append("ok")
+        return 200
+
+    async def smtp_timeout():
+        with pytest.raises(asyncio.TimeoutError):
+            await asyncio.wait_for(asyncio.sleep(10), timeout=0.02)
+
+    t0 = asyncio.get_event_loop().time()
+    api, smtp = await asyncio.gather(api_call(), smtp_timeout(), return_exceptions=True)
+    elapsed = asyncio.get_event_loop().time() - t0
+
+    assert elapsed < 0.1, f"SMTP timeout blocked: {elapsed:.3f}s"
+    assert api == 200
+# #endregion test_smtp_timeout_doesnt_block_api
+
+
+# #endregion TestAsyncNotifications