Merge pull request #902 from cbcoutinho/feat/scan-on-provision

feat(vector-sync): scan provisioned users immediately

nextcloud_mcp_server/api/passwords.py

@@ -407,6 +407,12 @@ async def provision_app_password(request: Request) -> JSONResponse:
             username, app_password, scopes=scopes, username=nc_username
         )
         invalidate_scope_cache(username)
+        # Wake the background sync user manager so this user's scanner starts
+        # now instead of after the next poll. Local import avoids an app <->
+        # api-module import cycle.
+        from nextcloud_mcp_server.app import notify_user_provisioned  # noqa: PLC0415
+
+        notify_user_provisioned()
 
         _record_rate_limit_attempt(path_user_id, success=True)
         logger.info("Provisioned app password for user: %s", username)

nextcloud_mcp_server/app.py

@@ -128,6 +128,7 @@ from nextcloud_mcp_server.server.auth_tools import register_auth_tools
 from nextcloud_mcp_server.server.oauth_tools import register_oauth_tools
 from nextcloud_mcp_server.vector.metrics_publisher import vector_sync_metrics_task
 from nextcloud_mcp_server.vector.oauth_sync import (
+    ProvisionSignal,
     oauth_processor_task,
     user_manager_task,
 )
@@ -344,6 +345,12 @@ class VectorSyncState:
     task_producer: "TaskProducer | None" = None
     shutdown_event: anyio.Event | None = None
     scanner_wake_event: anyio.Event | None = None
+    # Rung by a provisioning request to wake ``user_manager_task`` immediately so
+    # a just-provisioned user's scanner is spawned without waiting out the
+    # ``VECTOR_SYNC_USER_POLL_INTERVAL`` poll. ``None`` when no user manager is
+    # running (single-user mode or vector sync disabled), in which case
+    # ``notify_user_provisioned`` is a no-op.
+    provision_signal: "ProvisionSignal | None" = None
     # Long-lived task group used for fire-and-forget background work spawned
     # from the request path (e.g. ADR-019 verify-on-read eviction). Set by the
     # starlette lifespan after entering its task group; cleared on shutdown.
@@ -354,6 +361,23 @@ class VectorSyncState:
 _vector_sync_state = VectorSyncState()
 
 
+def notify_user_provisioned() -> None:
+    """Wake the user manager to discover a just-provisioned user immediately.
+
+    Provisioning call sites invoke this after a successful app-password store so
+    ``user_manager_task`` re-polls at once instead of waiting out
+    ``VECTOR_SYNC_USER_POLL_INTERVAL``. The 60s poll remains the backstop, so a
+    missed signal (e.g. provisioning handled on a different replica than the
+    manager) only delays the scan, never skips it.
+
+    No-op when ``provision_signal`` is ``None`` — single-user mode or vector
+    sync disabled, where no user manager is running.
+    """
+    signal = _vector_sync_state.provision_signal
+    if signal is not None:
+        signal.ring()
+
+
 def _wire_vector_sync_state(
     app: Starlette,
     transport: IngestTransport,
@@ -373,6 +397,10 @@ def _wire_vector_sync_state(
     ``eviction_task_group`` is deliberately not set here: it only exists once the
     lifespan has entered its ``anyio.create_task_group()`` (after this call), so
     the lifespan assigns it on the singleton directly at that point.
+    ``provision_signal`` is likewise excluded on purpose — only ``user_manager_task``
+    consumes it (request handlers reach it via ``notify_user_provisioned``), so the
+    multi-user lifespan sets it on the singleton directly rather than fanning it out
+    to ``app.state``/the browser sub-app.
     """
     send_stream = transport.send_stream
     receive_stream = transport.receive_stream
@@ -416,6 +444,7 @@ def _clear_vector_sync_state() -> None:
     # closed lifespan; the next startup's _wire_vector_sync_state rebinds them.
     _vector_sync_state.shutdown_event = None
     _vector_sync_state.scanner_wake_event = None
+    _vector_sync_state.provision_signal = None
 
 
 # =============================================================================
@@ -2060,6 +2089,12 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None =
                 # that choice — this path is now backend-agnostic.
                 shutdown_event = anyio.Event()
                 scanner_wake_event = anyio.Event()
+                # Doorbell the provisioning request path rings (via
+                # notify_user_provisioned) to wake the user manager immediately
+                # for a newly provisioned user. Held on the singleton only — both
+                # the manager and the signal helper reach it there.
+                provision_signal = ProvisionSignal()
+                _vector_sync_state.provision_signal = provision_signal
 
                 # User state tracking for user manager
                 user_states: dict = {}
@@ -2095,6 +2130,7 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None =
                         nextcloud_host_for_sync,
                         user_states,
                         tg,
+                        provision_signal,
                     )
 
                     # In-process consumer pool. ``run_consumers`` is a no-op for

nextcloud_mcp_server/auth/provision_routes.py

@@ -118,6 +118,14 @@ async def _poll_and_store(provision_id: str) -> None:
                 username=result.login_name,
             )
             invalidate_scope_cache(effective_user_id)
+            # Wake the background sync user manager so this user's scanner
+            # starts now instead of after the next poll. Local import avoids an
+            # app <-> route-module import cycle.
+            from nextcloud_mcp_server.app import (  # noqa: PLC0415
+                notify_user_provisioned,
+            )
+
+            notify_user_provisioned()
             session = _provision_sessions.get(provision_id)
             if session:
                 session["status"] = "completed"

nextcloud_mcp_server/server/auth_tools.py

@@ -288,6 +288,14 @@ def register_auth_tools(mcp: FastMCP) -> None:
                 username=poll_result.login_name,
             )
             invalidate_scope_cache(user_id)
+            # Wake the background sync user manager so this user's scanner
+            # starts now instead of after the next poll. Local import avoids an
+            # app <-> server-module import cycle.
+            from nextcloud_mcp_server.app import (  # noqa: PLC0415
+                notify_user_provisioned,
+            )
+
+            notify_user_provisioned()
 
             # Clean up the flow session
             await storage.delete_login_flow_session(user_id)

nextcloud_mcp_server/vector/oauth_sync.py

@@ -20,6 +20,7 @@ background sync.
 
 import logging
 import time
+from collections.abc import Awaitable, Callable
 from dataclasses import dataclass, field
 
 import anyio
@@ -44,6 +45,45 @@ class NotProvisionedError(Exception):
     pass
 
 
+class ProvisionSignal:
+    """One-shot doorbell that wakes ``user_manager_task`` on demand.
+
+    A provisioning request rings this (``ring()``) right after storing a new
+    user's app password so the manager re-polls immediately instead of waiting
+    out ``VECTOR_SYNC_USER_POLL_INTERVAL``. The manager parks on ``wait()``;
+    each ring releases exactly one wait, after which the underlying event is
+    re-armed for the next cycle.
+
+    The reference is stable for the life of the lifespan (stored once on the
+    ``VectorSyncState`` singleton), so the manager never has to republish a new
+    event back to shared state — avoiding any ``app`` ↔ ``vector`` import cycle.
+
+    Concurrency: ``anyio.Event`` is sticky, so a ``ring()`` that lands before
+    ``wait()`` is still observed. ``wait()`` re-arms in a ``finally`` with no
+    ``await`` before the swap, so under cooperative scheduling a concurrent
+    ``ring()`` cannot slip into that window and be lost — and the re-arm also
+    runs if ``wait()`` is cancelled (e.g. shutdown racing the doorbell), leaving
+    a fresh unset event rather than a stale set-but-consumed one.
+    """
+
+    def __init__(self) -> None:
+        self._event = anyio.Event()
+
+    def ring(self) -> None:
+        """Signal a pending wait (or the next one to arrive)."""
+        self._event.set()
+
+    async def wait(self) -> None:
+        """Block until the next ring, then re-arm for the following cycle."""
+        try:
+            await self._event.wait()
+        finally:
+            # Re-arm even on cancellation. The assignment is not a checkpoint,
+            # so no concurrent ring() can interleave before the swap; a ring
+            # that already arrived lands on the fresh event for the next wait().
+            self._event = anyio.Event()
+
+
 # Process-wide app-password storage for the BasicAuth client path.
 #
 # get_user_client_basic_auth is on the search hot path (Unified Search and the
@@ -408,6 +448,7 @@ async def user_manager_task(
     nextcloud_host: str,
     user_states: dict[str, UserSyncState],
     tg: TaskGroup,
+    provision_signal: "ProvisionSignal",
     *,
     task_status: TaskStatus = anyio.TASK_STATUS_IGNORED,
 ) -> None:
@@ -417,6 +458,12 @@ async def user_manager_task(
     - New users who have provisioned access -> start scanner
     - Users who have revoked access -> cancel their scanner
 
+    Polls every ``VECTOR_SYNC_USER_POLL_INTERVAL`` seconds, but also wakes
+    early whenever ``provision_signal`` is rung (by a provisioning request via
+    ``notify_user_provisioned``) so a just-provisioned user's scanner starts at
+    once rather than after up to a full poll interval. The poll remains the
+    backstop for any missed ring.
+
     Args:
         send_stream: Stream to send documents to processors
         shutdown_event: Event signaling shutdown
@@ -425,6 +472,7 @@ async def user_manager_task(
         nextcloud_host: Nextcloud base URL
         user_states: Shared dict tracking active user scanners
         tg: Task group for spawning scanner tasks
+        provision_signal: Doorbell rung on provisioning to force an early re-poll
         task_status: Status object for signaling task readiness
     """
     settings = get_settings()
@@ -433,6 +481,14 @@ async def user_manager_task(
     logger.info("[BasicAuth] User manager started (poll interval: %ss)", poll_interval)
     task_status.started()
 
+    # Sleep helper: await one of the wakeup events, then end the sleep by
+    # cancelling the shared scope. Defined once (not per loop iteration).
+    async def _wake_on(
+        wait_fn: Callable[[], Awaitable[object]], scope: anyio.CancelScope
+    ) -> None:
+        await wait_fn()
+        scope.cancel()
+
     while not shutdown_event.is_set():
         try:
             # Query the app_passwords table — background sync always
@@ -491,10 +547,19 @@ async def user_manager_task(
                 exc_info=True,
             )
 
-        # Sleep until next poll
+        # Sleep until the next poll tick, but wake early on shutdown or a
+        # provisioning signal so a just-provisioned user is discovered at once.
+        # Watch shutdown concurrently and block here on a provisioning ring;
+        # whichever fires first cancels the shared scope and ends the sleep,
+        # while move_on_after caps the wait at poll_interval. Awaiting one waiter
+        # directly keeps an explicit checkpoint inside the cancellation scope.
         try:
             with anyio.move_on_after(poll_interval):
-                await shutdown_event.wait()
+                async with anyio.create_task_group() as wake_tg:
+                    wake_tg.start_soon(
+                        _wake_on, shutdown_event.wait, wake_tg.cancel_scope
+                    )
+                    await _wake_on(provision_signal.wait, wake_tg.cancel_scope)
         except anyio.get_cancelled_exc_class():
             break
 

tests/integration/test_login_flow_provision_wake.py

@@ -0,0 +1,183 @@
+"""Integration test: Login Flow v2 provisioning wakes the background sync
+user manager immediately.
+
+Wires the real login-flow web-provision path end to end at the component level
+(no browser / container):
+
+    _poll_and_store (provision_routes)
+      -> RefreshTokenStorage.store_app_password_with_scopes  (real, temp DB)
+      -> notify_user_provisioned  ->  ProvisionSignal.ring
+      -> user_manager_task wakes, re-polls the same storage, spawns the scanner
+
+Only the Nextcloud-facing Login Flow v2 poll is mocked (returning "completed");
+everything in between is the real code. With a deliberately long poll interval,
+the new user's scanner must still be spawned promptly — proving the wake came
+from the provisioning signal, not the periodic poll.
+
+This is the Login Flow v2 deployment-mode counterpart to the multi-user
+BasicAuth coverage in ``test_app_password_provisioning.py``.
+"""
+
+import secrets
+import tempfile
+import time
+from pathlib import Path
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import anyio
+import pytest
+from cryptography.fernet import Fernet
+
+from nextcloud_mcp_server.auth.login_flow import LoginFlowPollResult
+from nextcloud_mcp_server.auth.provision_routes import (
+    _poll_and_store,
+    _provision_sessions,
+)
+from nextcloud_mcp_server.auth.storage import RefreshTokenStorage
+from nextcloud_mcp_server.vector.oauth_sync import ProvisionSignal, user_manager_task
+
+pytestmark = pytest.mark.integration
+
+
+@pytest.fixture
+def encryption_key():
+    return Fernet.generate_key().decode()
+
+
+@pytest.fixture
+async def temp_storage(encryption_key):
+    """Real RefreshTokenStorage backed by a temporary SQLite DB."""
+    with tempfile.TemporaryDirectory() as tmpdir:
+        db_path = Path(tmpdir) / "test_login_flow_wake.db"
+        storage = RefreshTokenStorage(
+            db_path=str(db_path), encryption_key=encryption_key
+        )
+        await storage.initialize()
+        yield storage
+
+
+async def test_login_flow_provision_wakes_user_manager(temp_storage, mocker):
+    """A completed Login Flow v2 web provision spawns the user's scanner at
+    once via the provision signal, well inside a long poll interval."""
+    # ── user_manager: long poll interval + stubbed per-user scanner ──────────
+    manager_settings = MagicMock()
+    manager_settings.vector_sync_user_poll_interval = 1000  # never fires here
+    mocker.patch(
+        "nextcloud_mcp_server.vector.oauth_sync.get_settings",
+        return_value=manager_settings,
+    )
+
+    spawned: set[str] = set()
+    alice_spawned = anyio.Event()
+
+    async def fake_scanner(
+        user_id,
+        cancel_scope,
+        send_stream,
+        shutdown_event,
+        wake_event,
+        nextcloud_host,
+        user_states,
+    ):
+        spawned.add(user_id)
+        if user_id == "alice":
+            alice_spawned.set()
+        with cancel_scope:
+            await shutdown_event.wait()
+        user_states.pop(user_id, None)
+
+    mocker.patch(
+        "nextcloud_mcp_server.vector.oauth_sync._run_user_scanner_with_scope",
+        fake_scanner,
+    )
+
+    # ── wire the doorbell exactly as the lifespan does ───────────────────────
+    import nextcloud_mcp_server.app as app_module
+
+    provision_signal = ProvisionSignal()
+    mocker.patch.object(
+        app_module._vector_sync_state, "provision_signal", provision_signal
+    )
+
+    # ── mock only the Nextcloud Login Flow v2 poll ───────────────────────────
+    # Generated, not a hardcoded literal — keeps this a fake token, not a
+    # credential pattern (SonarQube python:S2068).
+    fake_app_password = secrets.token_urlsafe(24)
+    completed = LoginFlowPollResult(
+        status="completed",
+        server="https://cloud.example.com",
+        login_name="alice",
+        app_password=fake_app_password,
+    )
+    flow_client = AsyncMock()
+    flow_client.poll.return_value = completed
+
+    provision_settings = MagicMock()
+    provision_settings.nextcloud_host = "https://cloud.example.com"
+    provision_settings.nextcloud_public_issuer_url = None
+
+    provision_id = "login-flow-wake"
+    _provision_sessions[provision_id] = {
+        "status": "pending",
+        "poll_endpoint": "https://cloud.example.com/login/v2/poll",
+        "poll_token": "secret-token",
+        "user_id": "alice",
+        "created_at": time.time(),
+        "expires_at": time.time() + 1200,
+    }
+
+    try:
+        shutdown_event = anyio.Event()
+        user_states: dict = {}
+
+        async with anyio.create_task_group() as tg:
+            await tg.start(
+                user_manager_task,
+                None,  # send_stream — unused by the stubbed scanner
+                shutdown_event,
+                anyio.Event(),  # scanner wake_event
+                temp_storage,
+                "https://cloud.example.com",
+                user_states,
+                tg,
+                provision_signal,
+            )
+
+            # First poll: no users provisioned yet → no scanner.
+            await anyio.sleep(0.1)
+            assert not spawned
+
+            # Run the real login-flow web-provision background task. It stores
+            # the app password into temp_storage and rings the doorbell.
+            with (
+                patch(
+                    "nextcloud_mcp_server.auth.provision_routes.get_settings",
+                    return_value=provision_settings,
+                ),
+                patch(
+                    "nextcloud_mcp_server.auth.provision_routes.get_nextcloud_ssl_verify",
+                    return_value=False,
+                ),
+                patch(
+                    "nextcloud_mcp_server.auth.provision_routes.LoginFlowV2Client",
+                    return_value=flow_client,
+                ),
+                patch(
+                    "nextcloud_mcp_server.auth.provision_routes.get_shared_storage",
+                    new_callable=AsyncMock,
+                    return_value=temp_storage,
+                ),
+            ):
+                await _poll_and_store(provision_id)
+
+            # The app password was really stored …
+            assert "alice" in await temp_storage.get_all_app_password_user_ids()
+            # … and the manager woke and spawned alice's scanner far inside the
+            # 1000s poll interval (i.e. because of the signal, not the poll).
+            with anyio.fail_after(3):
+                await alice_spawned.wait()
+            assert spawned == {"alice"}
+
+            shutdown_event.set()
+    finally:
+        _provision_sessions.pop(provision_id, None)

tests/unit/test_auth_tools.py

@@ -3,18 +3,45 @@
 Tests the auth tools logic with mocked storage and Login Flow client.
 """
 
+import secrets
 import tempfile
 from pathlib import Path
+from typing import cast
+from unittest.mock import AsyncMock, MagicMock
 
 import pytest
 from cryptography.fernet import Fernet
+from mcp.server.fastmcp import FastMCP
 
+from nextcloud_mcp_server.auth.login_flow import LoginFlowPollResult
 from nextcloud_mcp_server.auth.storage import RefreshTokenStorage
 from nextcloud_mcp_server.models.auth import ALL_SUPPORTED_SCOPES
+from nextcloud_mcp_server.server.auth_tools import register_auth_tools
 
 pytestmark = pytest.mark.unit
 
 
+def _capture_registered_tools() -> dict:
+    """Register the auth tools against a stub MCP and return them by name.
+
+    ``register_auth_tools`` only uses ``@mcp.tool(...)`` decorators, so a stub
+    whose ``tool()`` returns an identity decorator captures the closures without
+    a real FastMCP instance.
+    """
+    captured: dict = {}
+
+    class _StubMCP:
+        def tool(self, *args, **kwargs):
+            def deco(fn):
+                captured[fn.__name__] = fn
+                return fn
+
+            return deco
+
+    register_auth_tools(cast(FastMCP, _StubMCP()))
+    return captured
+
+
 @pytest.fixture
 def encryption_key():
     """Generate a test encryption key."""
@@ -196,3 +223,64 @@ def test_all_supported_scopes():
     read_scopes = [s for s in ALL_SUPPORTED_SCOPES if s.endswith(":read")]
     write_scopes = [s for s in ALL_SUPPORTED_SCOPES if s.endswith(":write")]
     assert len(read_scopes) == len(write_scopes)
+
+
+# ── Background-sync wake on provisioning ──
+
+
+async def test_check_status_completion_wakes_user_manager(mocker):
+    """When nc_auth_check_status polls a completed Login Flow, it stores the app
+    password and rings the background-sync doorbell (the server/auth_tools.py
+    wake path)."""
+    check_status = _capture_registered_tools()["nc_auth_check_status"]
+
+    mocker.patch(
+        "nextcloud_mcp_server.server.auth_tools.extract_user_id_from_token",
+        AsyncMock(return_value="alice"),
+    )
+    storage = MagicMock()
+    storage.get_app_password_with_scopes = AsyncMock(return_value=None)  # not yet
+    storage.get_login_flow_session = AsyncMock(
+        return_value={
+            "poll_endpoint": "https://nc/login/v2/poll",
+            "poll_token": "tok",
+            "requested_scopes": None,
+        }
+    )
+    storage.store_app_password_with_scopes = AsyncMock()
+    storage.delete_login_flow_session = AsyncMock()
+    mocker.patch(
+        "nextcloud_mcp_server.server.auth_tools.get_shared_storage",
+        AsyncMock(return_value=storage),
+    )
+    mocker.patch(
+        "nextcloud_mcp_server.server.auth_tools.get_settings",
+        return_value=MagicMock(
+            nextcloud_host="https://nc", nextcloud_public_issuer_url=None
+        ),
+    )
+    mocker.patch(
+        "nextcloud_mcp_server.server.auth_tools.get_nextcloud_ssl_verify",
+        return_value=False,
+    )
+    mocker.patch("nextcloud_mcp_server.server.auth_tools.invalidate_scope_cache")
+
+    flow_client = AsyncMock()
+    flow_client.poll = AsyncMock(
+        return_value=LoginFlowPollResult(
+            status="completed",
+            login_name="alice",
+            app_password=secrets.token_urlsafe(24),  # generated, not a literal
+        )
+    )
+    mocker.patch(
+        "nextcloud_mcp_server.server.auth_tools.LoginFlowV2Client",
+        return_value=flow_client,
+    )
+    notify = mocker.patch("nextcloud_mcp_server.app.notify_user_provisioned")
+
+    response = await check_status(MagicMock())
+
+    assert response.status == "provisioned"
+    storage.store_app_password_with_scopes.assert_awaited_once()
+    notify.assert_called_once()

tests/unit/test_management_app_password_endpoints.py

@@ -254,6 +254,48 @@ async def test_provision_app_password_success(temp_storage, mocker):
     assert get_kwargs["auth"] == ("testuser", "aaaaa-bbbbb-ccccc-ddddd-eeeee")
 
 
+async def test_provision_app_password_wakes_user_manager(temp_storage, mocker):
+    """A successful provision rings the background-sync doorbell so the user
+    manager re-polls immediately (the api/passwords.py wake path)."""
+    mocker.patch(
+        "nextcloud_mcp_server.api.passwords.get_settings",
+        return_value=MagicMock(
+            nextcloud_host="http://localhost:8080",
+            nextcloud_verify_ssl=True,
+            nextcloud_ca_bundle=None,
+        ),
+    )
+
+    mock_response = MagicMock()
+    mock_response.status_code = 200
+    mock_response.json.return_value = {"ocs": {"data": {"id": "testuser"}}}
+    mock_client = AsyncMock()
+    mock_client.get = AsyncMock(return_value=mock_response)
+    mock_client.__aenter__ = AsyncMock(return_value=mock_client)
+    mock_client.__aexit__ = AsyncMock()
+    mocker.patch(
+        "nextcloud_mcp_server.api.passwords.nextcloud_httpx_client",
+        return_value=mock_client,
+    )
+
+    # Spy on the doorbell helper (imported locally from app at call time).
+    notify = mocker.patch("nextcloud_mcp_server.app.notify_user_provisioned")
+
+    client = TestClient(create_test_app(temp_storage))
+    response = client.post(
+        "/api/v1/users/testuser/app-password",
+        headers={
+            "Authorization": create_basic_auth_header(
+                "testuser", "aaaaa-bbbbb-ccccc-ddddd-eeeee"
+            )
+        },
+    )
+
+    assert response.status_code == 200
+    assert response.json()["success"] is True
+    notify.assert_called_once()
+
+
 async def test_provision_app_password_uses_loginname_not_uid(temp_storage, mocker):
     """Regression: when the Nextcloud UID differs from the loginName (e.g.
     OIDC-provisioned users whose UID is their display name — UID

tests/unit/vector/test_user_manager_provision_wake.py

@@ -0,0 +1,219 @@
+"""Unit tests for immediate-on-provision scanner spawning.
+
+Covers the doorbell that lets a provisioning request wake ``user_manager_task``
+at once instead of waiting out ``VECTOR_SYNC_USER_POLL_INTERVAL``:
+
+- ``ProvisionSignal`` semantics (sticky ring, wake-a-parked-wait, re-arm).
+- ``user_manager_task`` re-polls early when the signal is rung, spawning a
+  newly provisioned user's scanner well before the next poll tick.
+- ``notify_user_provisioned`` is a no-op when no manager is running.
+"""
+
+from unittest.mock import MagicMock
+
+import anyio
+import pytest
+
+from nextcloud_mcp_server.vector.oauth_sync import ProvisionSignal, user_manager_task
+
+pytestmark = pytest.mark.unit
+
+
+# ── ProvisionSignal primitive ────────────────────────────────────────────────
+
+
+async def test_provision_signal_ring_before_wait_is_observed():
+    """A ring that lands before wait() is sticky and returns immediately."""
+    signal = ProvisionSignal()
+    signal.ring()
+    with anyio.fail_after(1):
+        await signal.wait()
+
+
+async def test_provision_signal_wakes_parked_waiter():
+    """ring() releases a wait() that is already parked."""
+    signal = ProvisionSignal()
+    woke = anyio.Event()
+
+    async def waiter():
+        await signal.wait()
+        woke.set()
+
+    async with anyio.create_task_group() as tg:
+        tg.start_soon(waiter)
+        await anyio.sleep(0.05)  # let waiter park
+        assert not woke.is_set()
+        signal.ring()
+        with anyio.fail_after(1):
+            await woke.wait()
+
+
+async def test_provision_signal_rearms_for_next_cycle():
+    """After a ring is consumed, the next wait() blocks until the next ring."""
+    signal = ProvisionSignal()
+    signal.ring()
+    await signal.wait()  # consumes first ring, re-arms
+
+    # Second wait must block (no pending ring) then release on the next ring.
+    second = anyio.Event()
+
+    async def waiter():
+        await signal.wait()
+        second.set()
+
+    async with anyio.create_task_group() as tg:
+        tg.start_soon(waiter)
+        await anyio.sleep(0.05)
+        assert not second.is_set()  # proves the first ring did not carry over
+        signal.ring()
+        with anyio.fail_after(1):
+            await second.wait()
+
+
+# ── user_manager_task wake-on-provision ──────────────────────────────────────
+
+
+class _FakeStorage:
+    """Storage stub whose provisioned-user set the test mutates between polls."""
+
+    def __init__(self, users: set[str]):
+        self.users = users
+
+    async def get_all_app_password_user_ids(self) -> list[str]:
+        return list(self.users)
+
+
+async def test_user_manager_wakes_on_provision_signal(mocker):
+    """Ringing the signal makes the manager re-poll and spawn the new user's
+    scanner well before the (long) poll interval elapses."""
+    # Long poll interval so any prompt spawn proves it was the signal, not poll.
+    settings = MagicMock()
+    settings.vector_sync_user_poll_interval = 1000
+    mocker.patch(
+        "nextcloud_mcp_server.vector.oauth_sync.get_settings", return_value=settings
+    )
+
+    spawned: set[str] = set()
+    spawn_events: dict[str, anyio.Event] = {
+        "alice": anyio.Event(),
+        "bob": anyio.Event(),
+    }
+
+    async def fake_scanner(
+        user_id,
+        cancel_scope,
+        send_stream,
+        shutdown_event,
+        wake_event,
+        nextcloud_host,
+        user_states,
+    ):
+        spawned.add(user_id)
+        if user_id in spawn_events:
+            spawn_events[user_id].set()
+        # Stay alive (keeps user_states populated) until shutdown.
+        with cancel_scope:
+            await shutdown_event.wait()
+        user_states.pop(user_id, None)
+
+    mocker.patch(
+        "nextcloud_mcp_server.vector.oauth_sync._run_user_scanner_with_scope",
+        fake_scanner,
+    )
+
+    storage = _FakeStorage({"alice"})
+    provision_signal = ProvisionSignal()
+    shutdown_event = anyio.Event()
+    scanner_wake_event = anyio.Event()
+    user_states: dict = {}
+
+    async with anyio.create_task_group() as tg:
+        await tg.start(
+            user_manager_task,
+            None,  # send_stream — unused by the stubbed scanner
+            shutdown_event,
+            scanner_wake_event,
+            storage,
+            "https://nextcloud",
+            user_states,
+            tg,
+            provision_signal,
+        )
+
+        # First poll discovers the already-provisioned user.
+        with anyio.fail_after(2):
+            await spawn_events["alice"].wait()
+        assert "bob" not in spawned
+
+        # Provision a new user, then ring — the manager must pick bob up fast.
+        storage.users.add("bob")
+        provision_signal.ring()
+        with anyio.fail_after(2):  # << 1000s poll interval
+            await spawn_events["bob"].wait()
+        assert spawned == {"alice", "bob"}
+
+        shutdown_event.set()
+
+
+async def test_user_manager_shutdown_still_breaks_sleep(mocker):
+    """Setting shutdown wakes the manager out of its sleep promptly even with a
+    long poll interval (the doorbell race must not regress shutdown latency)."""
+    settings = MagicMock()
+    settings.vector_sync_user_poll_interval = 1000
+    mocker.patch(
+        "nextcloud_mcp_server.vector.oauth_sync.get_settings", return_value=settings
+    )
+
+    async def _unused_scanner(*args, **kwargs):
+        # No users provisioned, so this is never called; keep a harmless stub.
+        return None
+
+    mocker.patch(
+        "nextcloud_mcp_server.vector.oauth_sync._run_user_scanner_with_scope",
+        _unused_scanner,
+    )
+
+    storage = _FakeStorage(set())
+    shutdown_event = anyio.Event()
+
+    # fail_after wraps the whole task group: if shutdown_event doesn't break the
+    # 1000s sleep, the task group never exits and the 2s deadline trips.
+    with anyio.fail_after(2):
+        async with anyio.create_task_group() as tg:
+            await tg.start(
+                user_manager_task,
+                None,
+                shutdown_event,
+                anyio.Event(),
+                storage,
+                "https://nextcloud",
+                {},
+                tg,
+                ProvisionSignal(),
+            )
+            await anyio.sleep(0.05)  # let it enter the sleep
+            shutdown_event.set()
+
+
+# ── notify_user_provisioned no-op guard ──────────────────────────────────────
+
+
+def test_notify_user_provisioned_noop_without_manager(mocker):
+    """When no manager is running, the helper must not raise."""
+    import nextcloud_mcp_server.app as app_module
+
+    mocker.patch.object(app_module._vector_sync_state, "provision_signal", None)
+    # Should be a silent no-op.
+    app_module.notify_user_provisioned()
+
+
+async def test_notify_user_provisioned_rings_when_present(mocker):
+    """When a manager is running, the helper rings its signal."""
+    import nextcloud_mcp_server.app as app_module
+
+    signal = ProvisionSignal()
+    mocker.patch.object(app_module._vector_sync_state, "provision_signal", signal)
+    app_module.notify_user_provisioned()
+    # Public contract: after a ring, the next wait() returns without blocking.
+    with anyio.fail_after(1):
+        await signal.wait()