From 74b8c13a7d37a36af46caef0b79a3e52b43ff868 Mon Sep 17 00:00:00 2001
From: Wei Ouyang <oeway007@gmail.com>
Date: Fri, 6 Dec 2024 13:56:01 -0800
Subject: [PATCH] Support browser runner and enhance vector store (#723)

* Use redis as vector search

* Fix publish_to

* Fix syntax

* add runner; bump hypha-rpc

* Enhance vector store, support persistence to s3

* bump version

* rename service

* small fix

* Fix score return

* Enhance vector store, support persistence to s3

* bump version

* rename service

* small fix

* Fix score return

* Fix unload

* Update workspace.py

* Update workspace.py

* refine clean up logic

* Print error

* Improve task scheduling

* add activity tracker

* Fix teardown

* Improve lifetime management

* Fix applications and daemon

* Fix time

* Fix activity monitor

* Fix event bus; add timeout

* dump vector store

* Fix vector dump

* Fix errors

* Fix tests

* wait long enough

* Fix artifact unload and load

* Fix workspaces unload

* Disable schechdule check

* Add update

* Add update

* Add delay

* Fix test

* Long delay

* Default to 60s
---
 CHANGELOG.md                                  |   2 +-
 docs/getting-started.md                       |   2 +-
 docs/hypha-quick-tour.ipynb                   |   6 +-
 docs/migration-guide.md                       |  10 +-
 docs/service-type-annotation.md               |   2 +-
 helm-charts/aks-hypha.md                      |   2 +-
 helm-charts/hypha-server/Chart.yaml           |   2 +-
 helm-charts/hypha-server/values.yaml          |   2 +-
 hypha/VERSION                                 |   2 +-
 hypha/apps.py                                 | 146 ++++-
 hypha/artifact.py                             | 173 +++++-
 hypha/core/__init__.py                        |  39 +-
 hypha/core/activity.py                        | 131 +++++
 hypha/core/auth.py                            |   2 +
 hypha/core/store.py                           | 192 ++++++-
 hypha/core/workspace.py                       | 165 ++++--
 hypha/http.py                                 |   2 +-
 hypha/plugin_parser.py                        |  23 +-
 hypha/{core => }/queue.py                     |   0
 hypha/runner/__init__.py                      |  14 +-
 hypha/runner/browser.py                       |  32 +-
 hypha/s3.py                                   |   7 +-
 hypha/server.py                               |   5 +-
 hypha/taskiq_utils/__init__.py                |   0
 hypha/taskiq_utils/exceptions.py              |  17 +
 hypha/taskiq_utils/redis_backend.py           | 199 +++++++
 hypha/taskiq_utils/redis_broker.py            | 117 ++++
 hypha/taskiq_utils/schedule_source.py         |  92 +++
 hypha/templates/apps/login_template.html      |  36 +-
 .../hypha-core-app/hypha-app-webpython.js     |   2 +-
 hypha/templates/index.html                    |   2 +
 hypha/templates/ws/index.html                 |  14 +-
 hypha/vectors.py                              | 284 ++++++++-
 requirements.txt                              |   6 +-
 setup.py                                      |   6 +-
 tests/conftest.py                             |   8 +-
 tests/testASGIWebPythonPlugin.imjoy.html      |   2 +-
 tests/testFunctionsPlugin.imjoy.html          |   2 +-
 tests/testUnreliablePlugin.imjoy.html         |   2 +-
 tests/testWebPythonPlugin.imjoy.html          |   2 +-
 tests/testWebWorkerPlugin.imjoy.html          |   2 +-
 tests/testWindowPlugin1.imjoy.html            |   2 +-
 tests/test_event_bus.py                       |   2 +-
 tests/test_redis.py                           |  64 ++-
 tests/test_server_apps.py                     | 139 ++++-
 tests/test_services.py                        |   4 +-
 tests/test_vectors.py                         | 543 ++++++++++++++----
 tests/test_workspace.py                       |  14 +-
 48 files changed, 2173 insertions(+), 349 deletions(-)
 create mode 100644 hypha/core/activity.py
 rename hypha/{core => }/queue.py (100%)
 create mode 100644 hypha/taskiq_utils/__init__.py
 create mode 100644 hypha/taskiq_utils/exceptions.py
 create mode 100644 hypha/taskiq_utils/redis_backend.py
 create mode 100644 hypha/taskiq_utils/redis_broker.py
 create mode 100644 hypha/taskiq_utils/schedule_source.py

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 12cae0f9..574f962e 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -2,7 +2,7 @@
 
 ### 0.20.41
 
- - Add `time_limit` for server apps to limit the running time
+ - Add `stop_after_inactive` option for server apps to stop the server app after a period of inactivity.
  - Support launching server apps worker in a separate service
 
 ### 0.20.40
diff --git a/docs/getting-started.md b/docs/getting-started.md
index 99c2a77d..31cbfd51 100644
--- a/docs/getting-started.md
+++ b/docs/getting-started.md
@@ -220,7 +220,7 @@ svc = await get_remote_service("http://localhost:9527/ws-user-scintillating-lawy
 Include the following script in your HTML file to load the `hypha-rpc` client:
 
 ```html
-<script src="https://cdn.jsdelivr.net/npm/hypha-rpc@0.20.41/dist/hypha-rpc-websocket.min.js"></script>
+<script src="https://cdn.jsdelivr.net/npm/hypha-rpc@0.20.42/dist/hypha-rpc-websocket.min.js"></script>
 ```
 
 Use the following code in JavaScript to connect to the server and access an existing service:
diff --git a/docs/hypha-quick-tour.ipynb b/docs/hypha-quick-tour.ipynb
index d82664e4..5a87389b 100644
--- a/docs/hypha-quick-tour.ipynb
+++ b/docs/hypha-quick-tour.ipynb
@@ -2,7 +2,7 @@
  "cells": [
   {
    "attachments": {},
-   "cell_type": "markdown",
+   "close_after_inactive: "markdown",
    "metadata": {},
    "source": [
     "# Hypha Demos\n",
@@ -10,7 +10,7 @@
     "## Install client library"
    ]
   },
-  {
+  {close_after_inactive
    "cell_type": "code",
    "execution_count": null,
    "metadata": {},
@@ -24,7 +24,7 @@
     "    # For native python with pip\n",
     "    import subprocess\n",
     "    subprocess.call(['pip', 'install', 'hypha-rpc'])"
-   ]
+   ]close_after_inactive
   },
   {
    "attachments": {},
diff --git a/docs/migration-guide.md b/docs/migration-guide.md
index 27cfb483..ee3d772b 100644
--- a/docs/migration-guide.md
+++ b/docs/migration-guide.md
@@ -15,7 +15,7 @@ To connect to the server, instead of installing the `imjoy-rpc` module, you will
 pip install -U hypha-rpc # new install
 ```
 
-We also changed our versioning strategy, we use the same version number for the server and client, so it's easier to match the client and server versions. For example, `hypha-rpc` version `0.20.41` is compatible with Hypha server version `0.20.41`.
+We also changed our versioning strategy, we use the same version number for the server and client, so it's easier to match the client and server versions. For example, `hypha-rpc` version `0.20.42` is compatible with Hypha server version `0.20.42`.
 
 #### 2. Change the imports to use `hypha-rpc`
 
@@ -128,10 +128,10 @@ loop.run_forever()
 To connect to the server, instead of using the `imjoy-rpc` module, you will need to use the `hypha-rpc` module. The `hypha-rpc` module is a standalone module that provides the RPC connection to the Hypha server. You can include it in your HTML using a script tag:
 
 ```html
-<script src="https://cdn.jsdelivr.net/npm/hypha-rpc@0.20.41/dist/hypha-rpc-websocket.min.js"></script>
+<script src="https://cdn.jsdelivr.net/npm/hypha-rpc@0.20.42/dist/hypha-rpc-websocket.min.js"></script>
 ```
 
-We also changed our versioning strategy, we use the same version number for the server and client, so it's easier to match the client and server versions. For example, `hypha-rpc` version `0.20.41` is compatible with Hypha server version `0.20.41`.
+We also changed our versioning strategy, we use the same version number for the server and client, so it's easier to match the client and server versions. For example, `hypha-rpc` version `0.20.42` is compatible with Hypha server version `0.20.42`.
 
 #### 2. Change the connection method and use camelCase for service function names
 
@@ -149,7 +149,7 @@ Here is a suggested list of search and replace operations to update your code:
 Here is an example of how the updated code might look:
 
 ```html
-<script src="https://cdn.jsdelivr.net/npm/hypha-rpc@0.20.41/dist/hypha-rpc-websocket.min.js"></script>
+<script src="https://cdn.jsdelivr.net/npm/hypha-rpc@0.20.42/dist/hypha-rpc-websocket.min.js"></script>
 <script>
 async function main(){
     const server = await hyphaWebsocketClient.connectToServer({"server_url": "https://hypha.amun.ai"});
@@ -197,7 +197,7 @@ We created a tutorial to introduce this new feature: [service type annotation](.
 Here is a quick example in JavaScript:
 
 ```html
-<script src="https://cdn.jsdelivr.net/npm/hypha-rpc@0.20.41/dist/hypha-rpc-websocket.min.js"></script>
+<script src="https://cdn.jsdelivr.net/npm/hypha-rpc@0.20.42/dist/hypha-rpc-websocket.min.js"></script>
 
 <script>
 async function main(){
diff --git a/docs/service-type-annotation.md b/docs/service-type-annotation.md
index cbc62871..afcf3c67 100644
--- a/docs/service-type-annotation.md
+++ b/docs/service-type-annotation.md
@@ -134,7 +134,7 @@ if __name__ == "__main__":
 **JavaScript Client: Service Usage**
 
 ```html
-<script src="https://cdn.jsdelivr.net/npm/hypha-rpc@0.20.41/dist/hypha-rpc-websocket.min.js"></script>
+<script src="https://cdn.jsdelivr.net/npm/hypha-rpc@0.20.42/dist/hypha-rpc-websocket.min.js"></script>
 <script>
 async function main() {
     const server = await hyphaWebsocketClient.connectToServer({"server_url": "https://hypha.amun.ai"});
diff --git a/helm-charts/aks-hypha.md b/helm-charts/aks-hypha.md
index 32bd0315..c59ee348 100644
--- a/helm-charts/aks-hypha.md
+++ b/helm-charts/aks-hypha.md
@@ -146,7 +146,7 @@ replicaCount: 1
 image:
   repository: ghcr.io/amun-ai/hypha
   pullPolicy: IfNotPresent
-  tag: "0.20.41"
+  tag: "0.20.42"
 
 serviceAccount:
   create: true
diff --git a/helm-charts/hypha-server/Chart.yaml b/helm-charts/hypha-server/Chart.yaml
index e3238bdb..63337d1a 100644
--- a/helm-charts/hypha-server/Chart.yaml
+++ b/helm-charts/hypha-server/Chart.yaml
@@ -15,7 +15,7 @@ type: application
 # This is the chart version. This version number should be incremented each time you make changes
 # to the chart and its templates, including the app version.
 # Versions are expected to follow Semantic Versioning (https://semver.org/)
-version: 0.20.41
+version: 0.20.42
 
 # This is the version number of the application being deployed. This version number should be
 # incremented each time you make changes to the application. Versions are not expected to
diff --git a/helm-charts/hypha-server/values.yaml b/helm-charts/hypha-server/values.yaml
index 912bdadb..5d9b0450 100644
--- a/helm-charts/hypha-server/values.yaml
+++ b/helm-charts/hypha-server/values.yaml
@@ -8,7 +8,7 @@ image:
   repository: ghcr.io/amun-ai/hypha
   pullPolicy: IfNotPresent
   # Overrides the image tag whose default is the chart appVersion.
-  tag: "0.20.41"
+  tag: "0.20.42"
 
 imagePullSecrets: []
 nameOverride: ""
diff --git a/hypha/VERSION b/hypha/VERSION
index 16f2e1ac..edf84112 100644
--- a/hypha/VERSION
+++ b/hypha/VERSION
@@ -1,3 +1,3 @@
 {
-    "version": "0.20.41"
+    "version": "0.20.42"
 }
diff --git a/hypha/apps.py b/hypha/apps.py
index 3213af0f..f059c9fa 100644
--- a/hypha/apps.py
+++ b/hypha/apps.py
@@ -10,8 +10,7 @@
 from hypha import main_version
 from jinja2 import Environment, PackageLoader, select_autoescape
 from typing import Any, Dict, List, Optional, Union
-from hypha.core import UserInfo, UserPermission, ServiceInfo, ApplicationArtifact
-from hypha.runner.browser import BrowserAppRunner
+from hypha.core import UserInfo, UserPermission, ServiceInfo, ApplicationManifest
 from hypha.utils import (
     random_id,
     PLUGIN_CONFIG_FIELDS,
@@ -20,6 +19,7 @@
 import base58
 import random
 from hypha.plugin_parser import convert_config_to_artifact, parse_imjoy_plugin
+from hypha.core import WorkspaceInfo
 
 logging.basicConfig(stream=sys.stdout)
 logger = logging.getLogger("apps")
@@ -54,27 +54,38 @@ def __init__(
             loader=PackageLoader("hypha"), autoescape=select_autoescape()
         )
         self.templates_dir = Path(__file__).parent / "templates"
-        self._runners = None
 
         def shutdown(_) -> None:
             asyncio.ensure_future(self.shutdown())
 
         self.event_bus.on_local("shutdown", shutdown)
 
+        async def client_disconnected(info: dict) -> None:
+            """Handle client disconnected event."""
+            # {"id": client_id, "workspace": ws}
+            client_id = info["id"]
+            full_client_id = info["workspace"] + "/" + client_id
+            if full_client_id in self._sessions:
+                app_info = self._sessions.pop(full_client_id, None)
+                try:
+                    await app_info["_runner"].stop(full_client_id)
+                except Exception as exp:
+                    logger.warning(f"Failed to stop browser tab: {exp}")
+
+        self.event_bus.on_local("client_disconnected", client_disconnected)
+        store.set_server_app_controller(self)
+
     async def get_runners(self):
         # start the browser runner
         server = await self.store.get_public_api()
         svcs = await server.list_services("public/server-app-worker")
+        if not svcs:
+            return []
         runners = [await server.get_service(svc["id"]) for svc in svcs]
         if runners:
             return runners
-        elif self._runners:
-            return self._runners
-        self._runners = [
-            BrowserAppRunner(in_docker=self.in_docker),
-            BrowserAppRunner(in_docker=self.in_docker),
-        ]
-        return self._runners
+        else:
+            []
 
     async def setup_applications_collection(self, overwrite=True, context=None):
         """Set up the workspace."""
@@ -205,7 +216,7 @@ async def install(
                 "public_url": public_url,
             }
         )
-        ApplicationArtifact.model_validate(artifact_obj)
+        ApplicationManifest.model_validate(artifact_obj)
 
         try:
             artifact = await self.artifact_manager.read("applications", context=context)
@@ -352,7 +363,7 @@ async def start(
         timeout: float = 60,
         version: str = None,
         wait_for_service: Union[str, bool] = None,
-        time_limit: Optional[int] = 600,
+        stop_after_inactive: Optional[int] = None,
         context: Optional[dict] = None,
     ):
         """Start the app and keep it alive."""
@@ -379,10 +390,24 @@ async def start(
         artifact_info = await self.artifact_manager.read(
             f"applications:{app_id}", version=version, context=context
         )
-        artifact = artifact_info.get("manifest", {})
-        artifact = ApplicationArtifact.model_validate(artifact)
-
-        entry_point = artifact.entry_point
+        manifest = artifact_info.get("manifest", {})
+        manifest = ApplicationManifest.model_validate(manifest)
+        if manifest.singleton:
+            # check if the app is already running
+            for session_info in self._sessions.values():
+                if session_info["app_id"] == app_id:
+                    raise RuntimeError(
+                        f"App {app_id} is a singleton app and already running (id: {session_info['id']})"
+                    )
+        if manifest.daemon and stop_after_inactive and stop_after_inactive > 0:
+            raise ValueError("Daemon apps should not have stop_after_inactive set.")
+        if stop_after_inactive is None:
+            stop_after_inactive = (
+                600
+                if manifest.stop_after_inactive is None
+                else manifest.stop_after_inactive
+            )
+        entry_point = manifest.entry_point
         assert entry_point, f"Entry point not found for app {app_id}."
         server_url = self.local_base_url
         local_url = (
@@ -404,14 +429,13 @@ async def start(
             + (f"&version={version}" if version else "")
             + (f"&use_proxy=true")
         )
-
-        runner = random.choice(await self.get_runners())
+        runners = await self.get_runners()
+        if not runners:
+            raise Exception("No server app worker found")
+        runner = random.choice(runners)
 
         full_client_id = workspace + "/" + client_id
-        await runner.start(
-            url=local_url, session_id=full_client_id, time_limit=time_limit
-        )
-        self._sessions[full_client_id] = {
+        metadata = {
             "id": full_client_id,
             "app_id": app_id,
             "workspace": workspace,
@@ -419,6 +443,32 @@ async def start(
             "public_url": public_url,
             "_runner": runner,
         }
+
+        await runner.start(
+            url=local_url,
+            session_id=full_client_id,
+            metadata=metadata,
+        )
+        self._sessions[full_client_id] = metadata
+
+        # test activity tracker
+        tracker = self.store.get_activity_tracker()
+        if not manifest.daemon and stop_after_inactive and stop_after_inactive > 0:
+
+            async def _stop_after_inactive():
+                if full_client_id in self._sessions:
+                    await runner.stop(full_client_id)
+                logger.info(
+                    f"App {full_client_id} stopped because of inactive for {stop_after_inactive}s."
+                )
+
+            tracker.register(
+                full_client_id,
+                inactive_period=stop_after_inactive,
+                on_inactive=_stop_after_inactive,
+                entity_type="client",
+            )
+
         # collecting services registered during the startup of the script
         collected_services: List[ServiceInfo] = []
         app_info = {
@@ -452,14 +502,14 @@ def service_added(info: dict):
                 )
 
             # save the services
-            artifact.services = collected_services
-            artifact = ApplicationArtifact.model_validate(
-                artifact.model_dump(mode="json")
+            manifest.services = collected_services
+            manifest = ApplicationManifest.model_validate(
+                manifest.model_dump(mode="json")
             )
             await self.artifact_manager.edit(
                 f"applications:{app_id}",
                 version=version,
-                manifest=artifact.model_dump(mode="json"),
+                manifest=manifest.model_dump(mode="json"),
                 context=context,
             )
 
@@ -495,7 +545,9 @@ async def stop(
                 f"User {user_info.id} does not have permission"
                 f" to stop app {session_id} in workspace {workspace}."
             )
+        await self._stop(session_id, raise_exception=raise_exception)
 
+    async def _stop(self, session_id: str, raise_exception=True):
         if session_id in self._sessions:
             app_info = self._sessions.pop(session_id, None)
             try:
@@ -543,12 +595,13 @@ async def list_running(self, context: Optional[dict] = None) -> List[str]:
     async def list_apps(self, context: Optional[dict] = None):
         """List applications in the workspace."""
         try:
+            ws = context["ws"]
             apps = await self.artifact_manager.list_children(
-                "applications", context=context
+                f"{ws}/applications", context=context
             )
             return [app["manifest"] for app in apps]
         except KeyError:
-            return []
+            raise KeyError(f"Applications collection not found: {ws}")
         except Exception as exp:
             raise Exception(f"Failed to list apps: {exp}") from exp
 
@@ -556,7 +609,42 @@ async def shutdown(self) -> None:
         """Shutdown the app controller."""
         logger.info("Closing the server app controller...")
         for app in self._sessions.values():
-            await self.stop(app["id"])
+            await self.stop(app["id"], raise_exception=False)
+
+    async def prepare_workspace(self, workspace_info: WorkspaceInfo):
+        """Prepare the workspace."""
+        context = {
+            "ws": workspace_info.id,
+            "user": self.store.get_root_user().model_dump(),
+        }
+        apps = await self.list_apps(context=context)
+        # start daemon apps
+        for app in apps:
+            if app.get("daemon"):
+                try:
+                    await self.start(app["id"], context=context)
+                except Exception as exp:
+                    logger.error(
+                        f"Failed to start daemon app: {app['id']}, error: {exp}"
+                    )
+
+    async def close_workspace(self, workspace_info: WorkspaceInfo):
+        """Archive the workspace."""
+        # Stop all running apps
+        for app in list(self._sessions.values()):
+            if app["workspace"] == workspace_info.id:
+                await self._stop(app["id"], raise_exception=False)
+        # Send to all runners
+        runners = await self.get_runners()
+        if not runners:
+            return
+        for runner in runners:
+            try:
+                await runner.close_workspace(workspace_info.id)
+            except Exception as exp:
+                logger.warning(
+                    f"Worker failed to close workspace: {workspace_info.id}, error: {exp}"
+                )
 
     def get_service_api(self) -> Dict[str, Any]:
         """Get a list of service API endpoints."""
diff --git a/hypha/artifact.py b/hypha/artifact.py
index 36d7c078..3690a695 100644
--- a/hypha/artifact.py
+++ b/hypha/artifact.py
@@ -1,10 +1,12 @@
 import logging
+import traceback
 import time
 import sys
 import uuid_utils as uuid
 import random
 import re
 import json
+import asyncio
 import math
 from io import BytesIO
 import zipfile
@@ -35,6 +37,7 @@
 from hrid import HRID
 from hypha.utils import remove_objects_async, list_objects_async, safe_join
 from hypha.utils.zenodo import ZenodoClient
+from hypha.core import WorkspaceInfo
 from botocore.exceptions import ClientError
 from hypha.s3 import FSFileResponse
 from aiobotocore.session import get_session
@@ -1439,6 +1442,8 @@ async def create(
                         )
 
                 winfo = await self.store.get_workspace_info(workspace, load=True)
+                if not winfo:
+                    raise ValueError(f"Workspace '{workspace}' does not exist.")
                 if not winfo.persistent:
                     raise PermissionError(
                         f"Cannot create artifact in a non-persistent workspace `{workspace}`, you may need to login to a persistent workspace."
@@ -1554,16 +1559,26 @@ async def create(
                     await session.merge(new_artifact)
                 else:
                     session.add(new_artifact)
+                s3_config = self._get_s3_config(new_artifact, parent_artifact)
                 if new_artifact.type == "vector-collection":
-                    await self._vector_engine.create_collection(
-                        f"{new_artifact.workspace}/{new_artifact.alias}",
-                        config.get("vector_fields", []),
-                    )
+                    async with self._create_client_async(s3_config) as s3_client:
+                        prefix = safe_join(
+                            s3_config["prefix"],
+                            f"{new_artifact.id}/v0",
+                        )
+                        await self._vector_engine.create_collection(
+                            f"{new_artifact.workspace}/{new_artifact.alias}",
+                            config.get("vector_fields", []),
+                            overwrite=overwrite,
+                            s3_client=s3_client,
+                            bucket=s3_config["bucket"],
+                            prefix=prefix,
+                        )
                 await session.commit()
                 await self._save_version_to_s3(
                     version_index,
                     new_artifact,
-                    self._get_s3_config(new_artifact, parent_artifact),
+                    s3_config,
                 )
                 if version != "stage":
                     logger.info(
@@ -1914,9 +1929,20 @@ async def delete(
             )
 
             if artifact.type == "vector-collection":
-                await self._vector_engine.delete_collection(
-                    f"{artifact.workspace}/{artifact.alias}"
-                )
+                s3_config = self._get_s3_config(artifact, parent_artifact)
+                async with self._create_client_async(
+                    s3_config,
+                ) as s3_client:
+                    prefix = safe_join(
+                        s3_config["prefix"],
+                        f"{artifact.id}/v0",
+                    )
+                    await self._vector_engine.delete_collection(
+                        f"{artifact.workspace}/{artifact.alias}",
+                        s3_client=s3_client,
+                        bucket=s3_config["bucket"],
+                        prefix=prefix,
+                    )
 
             s3_config = self._get_s3_config(artifact, parent_artifact)
             if version is None:
@@ -1971,6 +1997,7 @@ async def add_vectors(
         self,
         artifact_id: str,
         vectors: list,
+        update: bool = False,
         embedding_models: Optional[Dict[str, str]] = None,
         context: dict = None,
     ):
@@ -1981,7 +2008,7 @@ async def add_vectors(
         session = await self._get_session()
         try:
             async with session.begin():
-                artifact, _ = await self._get_artifact_with_permission(
+                artifact, parent_artifact = await self._get_artifact_with_permission(
                     user_info, artifact_id, "add_vectors", session
                 )
                 assert (
@@ -1989,12 +2016,22 @@ async def add_vectors(
                 ), "Artifact must be a vector collection."
 
                 assert artifact.manifest, "Artifact must be committed before upserting."
-                await self._vector_engine.add_vectors(
-                    f"{artifact.workspace}/{artifact.alias}",
-                    vectors,
-                    embedding_models=embedding_models
-                    or artifact.config.get("embedding_models"),
-                )
+                s3_config = self._get_s3_config(artifact, parent_artifact)
+                async with self._create_client_async(s3_config) as s3_client:
+                    prefix = safe_join(
+                        s3_config["prefix"],
+                        f"{artifact.id}/v0",
+                    )
+                    await self._vector_engine.add_vectors(
+                        f"{artifact.workspace}/{artifact.alias}",
+                        vectors,
+                        update=update,
+                        embedding_models=embedding_models
+                        or artifact.config.get("embedding_models"),
+                        s3_client=s3_client,
+                        bucket=s3_config["bucket"],
+                        prefix=prefix,
+                    )
                 logger.info(f"Added vectors to artifact with ID: {artifact_id}")
         except Exception as e:
             raise e
@@ -2054,15 +2091,27 @@ async def remove_vectors(
         session = await self._get_session()
         try:
             async with session.begin():
-                artifact, _ = await self._get_artifact_with_permission(
+                artifact, parent_artifact = await self._get_artifact_with_permission(
                     user_info, artifact_id, "remove_vectors", session
                 )
                 assert (
                     artifact.type == "vector-collection"
                 ), "Artifact must be a vector collection."
-                await self._vector_engine.remove_vectors(
-                    f"{artifact.workspace}/{artifact.alias}", ids
-                )
+                s3_config = self._get_s3_config(artifact, parent_artifact)
+                async with self._create_client_async(
+                    s3_config,
+                ) as s3_client:
+                    prefix = safe_join(
+                        s3_config["prefix"],
+                        f"{artifact.id}/v0",
+                    )
+                    await self._vector_engine.remove_vectors(
+                        f"{artifact.workspace}/{artifact.alias}",
+                        ids,
+                        s3_client=s3_client,
+                        bucket=s3_config["bucket"],
+                        prefix=prefix,
+                    )
                 logger.info(f"Removed vectors from artifact with ID: {artifact_id}")
         except Exception as e:
             raise e
@@ -2697,6 +2746,92 @@ async def upload_files(dir_path=""):
         finally:
             await session.close()
 
+    async def prepare_workspace(self, workspace_info: WorkspaceInfo):
+        """Prepare all artifacts in the workspace."""
+        # Iterate through all artifacts in the workspace
+        # If it's a vector collection, load the vectors from S3
+        try:
+            session = await self._get_session(read_only=True)
+            async with session.begin():
+                query = select(ArtifactModel).where(
+                    ArtifactModel.workspace == workspace_info.id
+                )
+                result = await session.execute(query)
+                artifacts = result.scalars().all()
+                for artifact in artifacts:
+                    if artifact.type == "vector-collection":
+                        parent_artifact = (
+                            artifact.parent_id
+                            and await self._get_artifact(
+                                session=session, artifact_id=artifact.parent_id
+                            )
+                        )
+                        s3_config = self._get_s3_config(artifact, parent_artifact)
+                        async with self._create_client_async(s3_config) as s3_client:
+                            prefix = safe_join(
+                                s3_config["prefix"],
+                                f"{artifact.id}/v0",
+                            )
+                            await self._vector_engine.load_collection(
+                                f"{artifact.workspace}/{artifact.alias}",
+                                s3_client=s3_client,
+                                bucket=s3_config["bucket"],
+                                prefix=prefix,
+                            )
+            logger.info(
+                f"Artifacts (#{len(artifacts)}) in workspace {workspace_info.id} prepared."
+            )
+        except Exception as e:
+            logger.error(f"Error preparing workspace: {traceback.format_exc()}")
+            raise e
+        finally:
+            await session.close()
+
+    async def close_workspace(self, workspace_info: WorkspaceInfo):
+        """Archive all artifacts in the workspace."""
+        try:
+            session = await self._get_session(read_only=True)
+            async with session.begin():
+                query = select(ArtifactModel).where(
+                    ArtifactModel.workspace == workspace_info.id
+                )
+                result = await session.execute(query)
+                artifacts = result.scalars().all()
+                for artifact in artifacts:
+                    if artifact.type == "vector-collection":
+                        # parent_artifact = (
+                        #     artifact.parent_id
+                        #     and await self._get_artifact(
+                        #         session=session, artifact_id=artifact.parent_id
+                        #     )
+                        # )
+                        # s3_config = self._get_s3_config(artifact, parent_artifact)
+                        # async with self._create_client_async(s3_config) as s3_client:
+                        #     prefix = safe_join(
+                        #         s3_config["prefix"],
+                        #         f"{artifact.id}/v0",
+                        #     )
+                        #     await self._vector_engine.dump_collection(
+                        #         f"{artifact.workspace}/{artifact.alias}",
+                        #         s3_client=s3_client,
+                        #         bucket=s3_config["bucket"],
+                        #         prefix=prefix,
+                        #     )
+                        # Delete the collection
+                        await self._vector_engine.delete_collection(
+                            f"{artifact.workspace}/{artifact.alias}"
+                        )
+            logger.info(
+                f"Artifacts (#{len(artifacts)}) in workspace {workspace_info.id} prepared for closure."
+            )
+        except Exception as e:
+            logger.error(
+                f"Error closing workspace {workspace_info.id}: {traceback.format_exc()}"
+            )
+            raise e
+        finally:
+            await session.close()
+
     def get_artifact_service(self):
         """Return the artifact service definition."""
         return {
diff --git a/hypha/core/__init__.py b/hypha/core/__init__.py
index b1e367d9..117d2c3a 100644
--- a/hypha/core/__init__.py
+++ b/hypha/core/__init__.py
@@ -22,6 +22,7 @@
 
 from hypha.utils import EventBus
 import jsonschema
+from hypha.core.activity import ActivityTracker
 
 from prometheus_client import Counter
 
@@ -68,7 +69,7 @@ class ServiceInfo(BaseModel):
     id: str
     name: str
     type: Optional[str] = "generic"
-    description: Optional[constr(max_length=256)] = None  # type: ignore
+    description: Optional[constr(max_length=1024)] = None  # type: ignore
     docs: Optional[str] = None
     app_id: Optional[str] = None
     service_schema: Optional[Dict[str, Any]] = None
@@ -367,12 +368,21 @@ def model_validate(cls, data):
         return super().model_validate(data)
 
 
-class ApplicationArtifact(Artifact):
+class ApplicationManifest(Artifact):
     """Represent application artifact."""
 
+    passive: Optional[bool] = False
+    dependencies: Optional[List[str]] = []
+    requirements: Optional[List[str]] = []
+    api_version: Optional[str] = "0.1.0"
+    icon: Optional[str] = None
+    env: Optional[Union[str, Dict[str, Any]]] = None
     type: Optional[str] = "application"
     entry_point: Optional[str] = None  # entry point for the application
+    daemon: Optional[bool] = False  # whether the application is a daemon
+    singleton: Optional[bool] = False  # whether the application is a singleton
     services: Optional[List[SerializeAsAny[ServiceInfo]]] = None  # for application
+    stop_after_inactive: Optional[float] = None  # stop the application after inactivity
 
 
 class ServiceTypeInfo(BaseModel):
@@ -401,6 +411,7 @@ class WorkspaceInfo(BaseModel):
     docs: Optional[str] = None
     service_types: Optional[Dict[str, ServiceTypeInfo]] = {}
     config: Optional[Dict[str, Any]] = {}
+    status: Optional[Dict[str, Any]] = None
 
     def __init__(self, **data):
         super().__init__(**data)
@@ -436,6 +447,11 @@ class RedisRPCConnection:
     """Represent a Redis connection for handling RPC-like messaging."""
 
     _counter = Counter("rpc_call", "Counts the RPC calls", ["workspace"])
+    _tracker = None
+
+    @classmethod
+    def set_activity_tracker(cls, tracker: ActivityTracker):
+        cls._tracker = tracker
 
     def __init__(
         self,
@@ -512,6 +528,9 @@ async def emit_message(self, data: Union[dict, bytes]):
         # logger.info(f"Sending message to channel {target_id}:msg")
         await self._event_bus.emit(f"{target_id}:msg", packed_message)
         RedisRPCConnection._counter.labels(workspace=self._workspace).inc()
+        await RedisRPCConnection._tracker.reset_timer(
+            self._workspace + "/" + self._client_id, "client"
+        )
 
     async def disconnect(self, reason=None):
         """Handle disconnection."""
@@ -523,10 +542,14 @@ async def disconnect(self, reason=None):
             self._event_bus.off(f"{self._workspace}/*:msg", self._handle_message)
 
         self._handle_message = None
-        logger.info(f"Redis Connection Disconnected: {reason}")
+        logger.debug(f"Redis Connection Disconnected: {reason}")
         if self._handle_disconnected:
             await self._handle_disconnected(reason)
 
+        RedisRPCConnection._tracker.remove_entity(
+            self._workspace + "/" + self._client_id, "client"
+        )
+
 
 class RedisEventBus:
     """Represent a redis event bus."""
@@ -550,7 +573,7 @@ async def init(self):
         self._loop = loop
 
         # Start the Redis subscription task
-        loop.create_task(self._subscribe_redis())
+        self._subscribe_task = loop.create_task(self._subscribe_redis())
 
         # Wait for readiness signal
         await self._ready
@@ -653,8 +676,14 @@ def emit(self, event_name, data):
             fut.set_result(None)
             return fut
 
-    def stop(self):
+    async def stop(self):
+        """Stop the event bus."""
         self._stop = True
+        self._subscribe_task.cancel()
+        try:
+            await self._subscribe_task
+        except asyncio.CancelledError:
+            pass
 
     async def _subscribe_redis(self):
         cpu_count = os.cpu_count() or 1
diff --git a/hypha/core/activity.py b/hypha/core/activity.py
new file mode 100644
index 00000000..8c29b3e1
--- /dev/null
+++ b/hypha/core/activity.py
@@ -0,0 +1,131 @@
+import asyncio
+import time
+from typing import Callable, Dict, Optional
+import logging
+import sys
+
+logging.basicConfig(stream=sys.stdout)
+logger = logging.getLogger("activity-tracker")
+logger.setLevel(logging.INFO)
+
+
+class ActivityTracker:
+    def __init__(self, check_interval: int = 10):
+        """
+        :param check_interval: Time in seconds to check for active/inactive entities
+        """
+        self._registrations: Dict[str, Dict[str, dict]] = {}
+        self._check_interval = check_interval
+        self._stop = False
+
+    def register(
+        self,
+        entity_id: str,
+        inactive_period: int,
+        on_active: Optional[Callable] = None,
+        on_inactive: Optional[Callable] = None,
+        entity_type: Optional[str] = "default",
+    ) -> str:
+        """Register an entity with activity tracking.
+
+        :param entity_id: Unique ID of the entity (client, workspace, etc.)
+        :param inactive_period: Time in seconds after which the entity is considered inactive
+        :param on_active: Function to call when the entity becomes active
+        :param on_inactive: Function to call when the entity becomes inactive
+        :param entity_type: Type of entity being tracked (e.g., client, workspace)
+        :return: A unique registration ID
+        """
+        full_id = f"{entity_type}:{entity_id}"
+        if full_id not in self._registrations:
+            self._registrations[full_id] = {}
+
+        reg_id = f"{full_id}-{len(self._registrations[full_id])}"
+        if inactive_period < self._check_interval:
+            raise ValueError(
+                f"Inactive period must be greater than the check interval ({self._check_interval} seconds)"
+            )
+        self._registrations[full_id][reg_id] = {
+            "inactive_period": inactive_period,
+            "on_active": on_active,
+            "on_inactive": on_inactive,
+            "last_activity": None,
+            "is_active": None,
+        }
+        return reg_id
+
+    def unregister(
+        self, entity_id: str, reg_id: str, entity_type: Optional[str] = "default"
+    ):
+        """Unregister a specific registration for an entity.
+
+        :param entity_id: The entity's ID
+        :param reg_id: The unique registration ID to remove
+        :param entity_type: Type of entity being tracked
+        """
+        full_id = f"{entity_type}:{entity_id}"
+        if full_id in self._registrations:
+            if reg_id in self._registrations[full_id]:
+                del self._registrations[full_id][reg_id]
+            if not self._registrations[
+                full_id
+            ]:  # Remove entity if no registrations left
+                del self._registrations[full_id]
+
+    def remove_entity(self, entity_id: str, entity_type: Optional[str] = "default"):
+        """Remove all registrations for an entity.
+
+        :param entity_id: The entity's ID to remove
+        :param entity_type: Type of entity being tracked
+        """
+        full_id = f"{entity_type}:{entity_id}"
+        if full_id in self._registrations:
+            del self._registrations[full_id]
+
+    async def reset_timer(self, entity_id: str, entity_type: Optional[str] = "default"):
+        """Reset the activity timer for all registrations of an entity."""
+        full_id = f"{entity_type}:{entity_id}"
+        # logger.info(f"Resetting timer for {full_id}")
+        if full_id in self._registrations:
+            now = time.time()
+            for reg in list(self._registrations[full_id].values()):
+                reg["last_activity"] = now
+                if not reg["is_active"]:
+                    reg["is_active"] = True
+                    if reg["on_active"]:
+                        try:
+                            await reg["on_active"]()
+                            logger.info(f"{full_id} becomes active")
+                        except Exception as e:
+                            logger.error(f"Error calling on_active for {full_id}: {e}")
+
+    async def monitor_entities(self):
+        """Periodically check the activity of registered entities."""
+        while not self._stop:
+            now = time.time()
+            try:
+                # logger.info(f"Checking activity at {now}")
+                for full_id in list(self._registrations.keys()):
+                    for reg in list(self._registrations[full_id].values()):
+                        if (
+                            reg["last_activity"]
+                            and (now - reg["last_activity"] > reg["inactive_period"])
+                            and reg["is_active"]
+                        ):
+                            # Trigger on_inactive event
+                            reg["is_active"] = False
+                            if reg["on_inactive"]:
+                                try:
+                                    await reg["on_inactive"]()
+                                    logger.info(f"{full_id} becomes inactive")
+                                except Exception as e:
+                                    logger.error(
+                                        f"Error calling on_inactive for {full_id}: {e}"
+                                    )
+                        await asyncio.sleep(0)
+            except Exception as e:
+                logger.error(f"Error monitoring activity: {e}")
+            await asyncio.sleep(self._check_interval)
+
+    def stop(self):
+        """Stop the monitoring loop."""
+        self._stop = True
diff --git a/hypha/core/auth.py b/hypha/core/auth.py
index 2d12bfd5..f9c65d91 100644
--- a/hypha/core/auth.py
+++ b/hypha/core/auth.py
@@ -18,6 +18,7 @@
 
 from hypha.core import UserInfo, UserTokenInfo, ScopeInfo, UserPermission, WorkspaceInfo
 from hypha.utils import random_id
+from hypha import __version__
 
 logging.basicConfig(stream=sys.stdout)
 logger = logging.getLogger("auth")
@@ -350,6 +351,7 @@ def create_login_service(store):
     )
     temp = jinja_env.get_template("apps/login_template.html")
     login_page = temp.render(
+        hypha_version=__version__,
         login_service_url=login_service_url,
         generate_token_url=generate_token_url,
         auth0_client_id=AUTH0_CLIENT_ID,
diff --git a/hypha/core/store.py b/hypha/core/store.py
index 540e2c3b..1525629d 100644
--- a/hypha/core/store.py
+++ b/hypha/core/store.py
@@ -2,19 +2,24 @@
 import asyncio
 import json
 import logging
-import time
 import sys
 import datetime
-from typing import List, Union
+from typing import List, Union, Optional
 from pydantic import BaseModel
 from fastapi import Header, Cookie
 
 from hypha_rpc import RPC
 from hypha_rpc.utils.schema import schema_method
 from starlette.routing import Mount
-from pydantic.fields import Field
 from aiocache.backends.redis import RedisCache
 from aiocache.serializers import PickleSerializer
+from taskiq import TaskiqScheduler
+from taskiq.kicker import AsyncKicker
+from taskiq.api import run_receiver_task, run_scheduler_task
+from hypha.taskiq_utils.redis_broker import ListQueueBroker
+from hypha.taskiq_utils.redis_backend import RedisAsyncResultBackend
+from hypha.taskiq_utils.schedule_source import RedisScheduleSource
+
 
 from hypha import __version__
 from hypha.core import (
@@ -25,6 +30,7 @@
     UserInfo,
     WorkspaceInfo,
 )
+from hypha.core.activity import ActivityTracker
 from sqlalchemy.ext.asyncio import (
     create_async_engine,
 )
@@ -101,9 +107,11 @@ def __init__(
         cache_dir=None,
         enable_service_search=False,
         reconnection_token_life_time=2 * 24 * 60 * 60,
+        activity_check_interval=10,
     ):
         """Initialize the redis store."""
         self._s3_controller = None
+        self._server_app_controller = None
         self._artifact_manager = None
         self._app = app
         self._codecs = {}
@@ -122,6 +130,7 @@ def __init__(
         self._manager_id = "manager-" + self._server_id
         self.reconnection_token_life_time = reconnection_token_life_time
         self._enable_service_search = enable_service_search
+        self._activity_check_interval = activity_check_interval
         self._server_info = {
             "server_id": self._server_id,
             "hypha_version": __version__,
@@ -168,17 +177,33 @@ def __init__(
             from redis import asyncio as aioredis
 
             self._redis = aioredis.from_url(redis_uri)
+
         else:  #  Create a redis server with fakeredis
             from fakeredis import aioredis
 
             self._redis = aioredis.FakeRedis.from_url("redis://localhost:9997/11")
 
+        self._result_backend = RedisAsyncResultBackend(self._redis)
+        self._broker = ListQueueBroker(self._redis).with_result_backend(
+            self._result_backend
+        )
+        self._broker_task = None
+        self._source = RedisScheduleSource(self._redis)
+        self._scheduler = TaskiqScheduler(
+            broker=self._broker,
+            sources=[self._source],
+        )
+        self._house_keeping_schedule = None
+        self._first_run = True
+
         self._redis_cache = RedisCache(serializer=PickleSerializer())
         self._redis_cache.client = self._redis
-
         self._root_user = None
         self._event_bus = RedisEventBus(self._redis)
 
+        self._tracker = None
+        self._tracker_task = None
+
     def set_websocket_server(self, websocket_server):
         """Set the websocket server."""
         assert self._websocket_server is None, "Websocket server already set"
@@ -278,23 +303,25 @@ async def _run_startup_functions(self, startup_functions):
 
     async def housekeeping(self):
         """Perform housekeeping tasks."""
-        # Perform housekeeping tasks
-        # Start the housekeeping task after 2 minutes
-        logger.info("Starting housekeeping task in 2 minutes...")
-        await asyncio.sleep(120)
-        while True:
-            try:
-                logger.info("Running housekeeping task...")
-                async with self.get_workspace_interface(
-                    self._root_user, "ws-user-root", client_id="housekeeping"
-                ) as api:
-                    # admin = await api.get_service("admin-utils")
-                    workspaces = await api.list_workspaces()
-                    for workspace in workspaces:
-                        await api.cleanup(workspace.id)
-                await asyncio.sleep(3600)
-            except Exception as e:
-                logger.exception(f"Error in housekeeping: {e}")
+        if self._first_run:
+            logger.info("Skipping housekeeping on first run")
+            self._first_run = False
+            return
+        try:
+            logger.info(f"Running housekeeping task at {datetime.datetime.now()}")
+            async with self.get_workspace_interface(
+                self._root_user, "ws-user-root", client_id="housekeeping"
+            ) as api:
+                # admin = await api.get_service("admin-utils")
+                workspaces = await api.list_workspaces()
+                for workspace in workspaces:
+                    summary = await api.cleanup(workspace.id)
+                    if "removed_clients" in summary:
+                        logger.info(
+                            f"Removed {len(summary['removed_clients'])} clients from workspace {workspace.id}"
+                        )
+        except Exception as e:
+            logger.exception(f"Error in housekeeping: {e}")
 
     async def upgrade(self):
         """Upgrade the store."""
@@ -446,6 +473,10 @@ async def init(self, reset_redis, startup_functions=None):
             logger.warning("RESETTING ALL REDIS DATA!!!")
             await self._redis.flushall()
         await self._event_bus.init()
+        self._tracker = ActivityTracker(check_interval=self._activity_check_interval)
+        self._tracker_task = asyncio.create_task(self._tracker.monitor_entities())
+        RedisRPCConnection.set_activity_tracker(self._tracker)
+
         await self.setup_root_user()
         await self.check_and_cleanup_servers()
         self._workspace_manager = await self.register_workspace_manager()
@@ -521,7 +552,67 @@ async def init(self, reset_redis, startup_functions=None):
         logger.info("Server initialized with server id: %s", self._server_id)
         logger.info("Currently connected hypha servers: %s", servers)
 
-        asyncio.create_task(self.housekeeping())
+        # Setup broker and scheduler
+        await self._broker.startup()
+        self._broker_task = asyncio.create_task(run_receiver_task(self._broker))
+        await self._scheduler.startup()
+        self._scheduler_task = asyncio.create_task(run_scheduler_task(self._scheduler))
+
+        # Do house keeping every 10 minutes
+        self._house_keeping_schedule = await self.schedule_task(
+            self.housekeeping, task_name="housekeeping", corn="*/1 * * * *"
+        )
+        self._first_run = True
+
+    async def schedule_task(
+        self,
+        task,
+        *args,
+        corn: Optional[str] = None,
+        time: Optional[datetime.datetime] = None,
+        task_name: Optional[str] = None,
+        task_id: Optional[str] = None,
+        schedule_id: Optional[str] = None,
+        **kwargs,
+    ):
+        """Schedule a task."""
+        assert not (corn and time), "Only one of corn or time can be provided"
+        assert corn or time, "Either corn or time must be provided"
+        my_task: AsyncKicker = self._broker.register_task(
+            task, task_name=task_name
+        ).kicker()
+        if task_id:
+            my_task = my_task.with_task_id(task_id)
+        if schedule_id:
+            my_task = my_task.with_schedule_id(schedule_id)
+        if corn:
+            return await my_task.schedule_by_cron(self._source, corn, *args, **kwargs)
+        if time:
+            return await my_task.schedule_by_time(self._source, time, *args, **kwargs)
+
+    async def delete_schedule(self, schedule_id: str):
+        """Remove schedule by id."""
+        await self._source.delete_schedule(schedule_id)
+
+    async def run_task(
+        self,
+        task,
+        *args,
+        task_name: Optional[str] = None,
+        task_id: Optional[str] = None,
+        **kwargs,
+    ):
+        """Send a task."""
+        my_task: AsyncKicker = self._broker.register_task(
+            task, task_name=task_name
+        ).kicker()
+        if task_id:
+            my_task = my_task.with_task_id(task_id)
+        return await my_task.kiq(*args, **kwargs)
+
+    async def get_task_result(self, task_id, with_logs: bool = False):
+        """Get the task result."""
+        return await self._result_backend.get_result(task_id, with_logs=with_logs)
 
     async def _register_root_services(self):
         """Register root services."""
@@ -542,9 +633,22 @@ async def _register_root_services(self):
                 "list_servers": self.list_servers,
                 "kickout_client": self.kickout_client,
                 "list_workspaces": self.list_all_workspaces,
+                "unload_workspace": self.unload_workspace,
             }
         )
 
+    @schema_method
+    async def unload_workspace(self, workspace: str, wait: bool = False, timeout=10):
+        """Unload a workspace."""
+        context = {"user": self._root_user.model_dump(), "ws": workspace}
+        await self._workspace_manager.unload(context=context)
+        if wait:
+            if not await self.workspace_exists(workspace):
+                return
+            await self._event_bus.wait_for(
+                "workspace_unloaded", match={"id": workspace}, timeout=timeout
+            )
+
     @schema_method
     async def list_servers(self):
         """List all servers."""
@@ -558,6 +662,9 @@ async def list_servers(self):
             clients.add(client_id)
         return list(clients)
 
+    def get_activity_tracker(self):
+        return self._tracker
+
     async def get_public_api(self):
         """Get the public API."""
         if self._public_workspace_interface is None:
@@ -687,6 +794,7 @@ def get_manager_id(self):
     async def register_workspace_manager(self):
         """Register a workspace manager."""
         manager = WorkspaceManager(
+            self,
             self._redis,
             self._root_user,
             self._event_bus,
@@ -694,6 +802,7 @@ async def register_workspace_manager(self):
             self._manager_id,
             self._sql_engine,
             self._s3_controller,
+            self._server_app_controller,
             self._artifact_manager,
             self._enable_service_search,
             self._cache_dir,
@@ -749,7 +858,7 @@ def create_rpc(
         """Create a rpc object for a workspace."""
         client_id = client_id or "anonymous-client-" + random_id(readable=False)
         assert "/" not in client_id
-        logger.info("Creating RPC for client %s", client_id)
+        logger.debug("Creating RPC for client %s", client_id)
         assert user_info is not None, "User info is required"
         connection = RedisRPCConnection(
             self._event_bus,
@@ -796,6 +905,14 @@ def set_s3_controller(self, controller):
         """Set the s3 controller."""
         self._s3_controller = controller
 
+    def set_server_app_controller(self, controller):
+        """Set the server app controller."""
+        self._server_app_controller = controller
+
+    def get_server_app_controller(self):
+        """Get the server app controller."""
+        return self._server_app_controller
+
     def set_artifact_manager(self, controller):
         """Set the artifact controller."""
         self._artifact_manager = controller
@@ -865,7 +982,24 @@ def unmount_app(self, path):
     async def teardown(self):
         """Teardown the server."""
         self._ready = False
-        logger.info("Tearing down the public workspace...")
+        logger.info("Tearing down the redis store...")
+        if self._house_keeping_schedule:
+            await self._house_keeping_schedule.unschedule()
+
+        await self._broker.shutdown()
+        self._scheduler_task.cancel()
+        try:
+            await self._scheduler_task
+        except asyncio.CancelledError:
+            print("Scheduler successfully exited.")
+        await self._scheduler.shutdown()
+
+        if self._tracker_task:
+            self._tracker_task.cancel()
+            try:
+                await self._tracker_task
+            except asyncio.CancelledError:
+                print("Activity tracker successfully exited.")
         client_id = self._public_workspace_interface.rpc.get_client_info()["id"]
         await self.remove_client(client_id, "public", self._root_user, unload=True)
         client_id = self._root_workspace_interface.rpc.get_client_info()["id"]
@@ -879,4 +1013,14 @@ async def teardown(self):
                     await ws.close()
                 except GeneratorExit:
                     pass
+
+        await self._event_bus.stop()
+
+        # Note: This has to be the last thing to do
+        # Otherwise, the server got stuck in the shutdown process
+        self._broker_task.cancel()
+        try:
+            await self._broker_task
+        except asyncio.CancelledError:
+            print("Broker successfully exited.")
         logger.info("Teardown complete")
diff --git a/hypha/core/workspace.py b/hypha/core/workspace.py
index bf9766a5..219ed895 100644
--- a/hypha/core/workspace.py
+++ b/hypha/core/workspace.py
@@ -2,6 +2,7 @@
 import json
 import asyncio
 import logging
+import traceback
 import time
 import sys
 from typing import Optional, Union, List, Any, Dict
@@ -25,7 +26,7 @@
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
 
 from hypha.core import (
-    ApplicationArtifact,
+    ApplicationManifest,
     RedisRPCConnection,
     UserInfo,
     WorkspaceInfo,
@@ -118,6 +119,7 @@ class GetServiceConfig(BaseModel):
 class WorkspaceManager:
     def __init__(
         self,
+        store: Any,
         redis: aioredis.FakeRedis,
         root_user: UserInfo,
         event_bus: EventBus,
@@ -125,11 +127,13 @@ def __init__(
         client_id: str,
         sql_engine: Optional[str] = None,
         s3_controller: Optional[Any] = None,
+        server_app_controller: Optional[Any] = None,
         artifact_manager: Optional[Any] = None,
         enable_service_search: bool = False,
         cache_dir: str = None,
     ):
         self._redis = redis
+        self._store = store
         self._initialized = False
         self._rpc = None
         self._root_user = root_user
@@ -138,6 +142,7 @@ def __init__(
         self._client_id = client_id
         self._s3_controller = s3_controller
         self._artifact_manager = artifact_manager
+        self._server_app_controller = server_app_controller
         self._sql_engine = sql_engine
         self._cache_dir = cache_dir
         if self._sql_engine:
@@ -150,6 +155,9 @@ def __init__(
         self._active_svc = Gauge(
             "active_services", "Number of active services", ["workspace"]
         )
+        self._active_clients = Gauge(
+            "active_clients", "Number of active clients", ["workspace"]
+        )
         self._enable_service_search = enable_service_search
 
     async def _get_sql_session(self):
@@ -220,6 +228,7 @@ async def setup(
                     {"type": "TAG", "name": "singleton"},
                     {"type": "TEXT", "name": "created_by"},
                 ],
+                overwrite=True,
             )
         self._initialized = True
         return rpc
@@ -529,18 +538,9 @@ async def create_workspace(
             await self.bookmark(
                 {"bookmark_type": "workspace", "id": workspace.id}, context=context
             )
-
-        # clean up the user's workspace
-        summary = await self.cleanup(
-            workspace.id,
-            context={"ws": workspace.id, "user": self._root_user.model_dump()},
-        )
-        if summary:
-            logger.info(
-                "Created workspace %s, clean up summary: %s", workspace.id, summary
-            )
-        else:
-            logger.info("Created workspace %s", workspace.id)
+        logger.info("Created workspace %s", workspace.id)
+        # preare the workspace for the user
+        await self._store.run_task(self._prepare_workspace, workspace)
         return workspace.model_dump()
 
     @schema_method
@@ -1206,9 +1206,7 @@ async def register_service(
                     "client_connected", {"id": client_id, "workspace": ws}
                 )
                 logger.info(f"Adding built-in service: {service.id}")
-                builtins = await self._redis.keys(
-                    f"services:*|*:{client_id}:built-in@*"
-                )
+                self._active_clients.labels(workspace=ws).inc()
             else:
                 # Remove the service embedding from the config
                 if service.config and service.config.service_embedding is not None:
@@ -1333,6 +1331,7 @@ async def unregister_service(
                 await self._event_bus.emit(
                     "client_disconnected", {"id": client_id, "workspace": ws}
                 )
+                self._active_clients.labels(workspace=ws).dec()
             else:
                 await self._event_bus.emit("service_removed", service.model_dump())
                 self._active_svc.labels(workspace=ws).dec()
@@ -1417,9 +1416,11 @@ async def load_workspace_info(self, workspace: str, load=True) -> WorkspaceInfo:
                 )
                 if not workspace_info:
                     raise KeyError(f"Workspace not found: {workspace}")
+                workspace_info.status = None
                 await self._redis.hset(
                     "workspaces", workspace_info.id, workspace_info.model_dump_json()
                 )
+                await self._store.run_task(self._prepare_workspace, workspace_info)
                 self._active_ws.inc()
                 await self._s3_controller.setup_workspace(workspace_info)
                 await self._event_bus.emit(
@@ -1480,10 +1481,11 @@ async def _launch_application_for_service(
         ], f"Invalid service id: {service_id}"
 
         applications = await self._artifact_manager.list_children(
-            "applications", context={"ws": workspace, "user": user_info.model_dump()}
+            f"{workspace}/applications",
+            context={"ws": workspace, "user": user_info.model_dump()},
         )
         applications = {
-            item["manifest"]["id"]: ApplicationArtifact.model_validate(item["manifest"])
+            item["manifest"]["id"]: ApplicationManifest.model_validate(item["manifest"])
             for item in applications
         }
         if app_id not in applications:
@@ -1507,23 +1509,21 @@ async def _launch_application_for_service(
                 f"Service id `{service_id}` not found in application {app_id}"
             )
 
-        async with self._get_service_api(
-            "public/server-apps", context=context
-        ) as controller:
-            if not controller:
-                raise Exception(
-                    "Failed to launch application: server-apps service not found."
-                )
-            client_info = await controller.start(
-                app_id,
-                timeout=timeout * 5,
-                wait_for_service=service_id,
-            )
-            return await self.get_service(
-                f"{client_info['id']}:{service_id}",  # should not contain @app_id
-                dict(timeout=timeout, case_conversion=case_conversion),
-                context=context,
+        if not self._server_app_controller:
+            raise Exception(
+                "Failed to launch application: server apps controller is not configured."
             )
+        client_info = await self._server_app_controller.start(
+            app_id,
+            timeout=timeout * 5,
+            wait_for_service=service_id,
+            context=context,
+        )
+        return await self.get_service(
+            f"{client_info['id']}:{service_id}",  # should not contain @app_id
+            dict(timeout=timeout, case_conversion=case_conversion),
+            context=context,
+        )
 
     @asynccontextmanager
     async def _get_service_api(self, service_id: str, context=None):
@@ -1701,6 +1701,7 @@ async def delete_client(
     ):
         """Delete all services associated with the given client_id in the specified workspace."""
         assert context is not None
+        assert isinstance(user_info, UserInfo)
         self.validate_context(context, permission=UserPermission.admin)
         cws = workspace
         validate_key_part(client_id)
@@ -1720,6 +1721,12 @@ async def delete_client(
         for key in keys:
             await self._redis.delete(key)
 
+        await self._event_bus.emit(
+            "client_disconnected", {"id": client_id, "workspace": cws}
+        )
+        self._active_clients.labels(workspace=cws).dec()
+        self._active_svc.labels(workspace=cws).dec(len(keys) - 1)
+
         if unload:
             if await self._redis.hexists("workspaces", cws):
                 if user_info.is_anonymous and cws == user_info.get_workspace():
@@ -1747,6 +1754,9 @@ async def unload(self, context=None):
         """Unload the workspace."""
         self.validate_context(context, permission=UserPermission.admin)
         ws = context["ws"]
+        if not await self._redis.hexists("workspaces", ws):
+            logger.warning(f"Workspace {ws} has already been unloaded.")
+            return
         winfo = await self.load_workspace_info(ws)
         # list all the clients in the workspace and send a meesage to delete them
         client_keys = await self._list_client_keys(winfo.id)
@@ -1761,23 +1771,87 @@ async def unload(self, context=None):
             )
             await self._event_bus.emit(f"unload:{ws}", "Unloading workspace: " + ws)
 
-        if winfo.persistent and self._s3_controller:
-            # since the workspace will be persisted, we can remove the workspace info from the redis store
-            await self._redis.hdel("workspaces", ws)
-
-        await self._event_bus.emit("workspace_unloaded", winfo.model_dump())
-        logger.info("Workspace %s unloaded.", ws)
+        # Mark the workspace as not ready
+        winfo.status = None
 
         if not winfo.persistent:
             # delete all the items in redis starting with `workspaces_name:`
+            # Including the queue and other associated resources
             keys = await self._redis.keys(f"{ws}:*")
             for key in keys:
                 await self._redis.delete(key)
-            await self._redis.hdel("workspaces", ws)
             if self._s3_controller:
                 await self._s3_controller.cleanup_workspace(winfo)
 
         self._active_ws.dec()
+        try:
+            self._active_clients.remove(ws)
+            self._active_svc.remove(ws)
+        except KeyError:
+            pass
+        await self._close_workspace(winfo)
+        await self._redis.hdel("workspaces", ws)
+
+    async def _prepare_workspace(self, workspace_info: WorkspaceInfo):
+        """Prepare the workspace."""
+        errors = {}
+        if workspace_info.persistent:
+            try:
+                if self._artifact_manager:
+                    await self._artifact_manager.prepare_workspace(workspace_info)
+            except Exception as e:
+                errors["artifact_manager"] = traceback.format_exc()
+            try:
+                if self._server_app_controller:
+                    await self._server_app_controller.prepare_workspace(workspace_info)
+            except Exception as e:
+                errors["server_app_controller"] = traceback.format_exc()
+        # Mark the workspace as ready
+        workspace_info.status = {"ready": True, "errors": errors}
+        await self._redis.hset(
+            "workspaces", workspace_info.id, workspace_info.model_dump_json()
+        )
+        await self._event_bus.emit("workspace_ready", workspace_info.model_dump())
+        logger.info("Workspace %s prepared.", workspace_info.id)
+
+    async def _close_workspace(self, workspace_info: WorkspaceInfo):
+        """Archive the workspace."""
+        assert (
+            workspace_info.status is None
+        ), "Workspace must be unloaded before archiving."
+        if workspace_info.persistent:
+            if self._artifact_manager:
+                try:
+                    await self._artifact_manager.close_workspace(workspace_info)
+                except Exception as e:
+                    logger.error(f"Aritfact manager failed to close workspace: {e}")
+            if self._server_app_controller:
+                try:
+                    await self._server_app_controller.close_workspace(workspace_info)
+                except Exception as e:
+                    logger.error(
+                        f"Server app controller failed to close workspace: {e}"
+                    )
+
+        await self._event_bus.emit("workspace_unloaded", workspace_info.model_dump())
+        logger.info("Workspace %s unloaded.", workspace_info.id)
+
+    @schema_method
+    async def wait_until_ready(self, timeout: Optional[int] = 60, context=None):
+        """Wait for the workspace to be ready."""
+        workspace_info = await self._redis.hget("workspaces", context["ws"])
+        workspace_info = WorkspaceInfo.model_validate(
+            json.loads(workspace_info.decode())
+        )
+        if workspace_info.status:
+            return workspace_info.status
+        try:
+            await self._event_bus.wait_for(
+                "workspace_ready", match={"id": context["ws"]}, timeout=timeout
+            )
+            return workspace_info.status
+        except asyncio.TimeoutError:
+            raise TimeoutError(f"Workspace {context['ws']} is not ready, timeout.")
 
     @schema_method
     async def cleanup(
@@ -1791,6 +1865,7 @@ async def cleanup(
         ws = context["ws"]
         if workspace is None:
             workspace = ws
+        workspace_info = await self.load_workspace_info(workspace)
         user_info = UserInfo.model_validate(context["user"])
         if not user_info.check_permission(workspace, UserPermission.admin):
             raise PermissionError(
@@ -1801,6 +1876,10 @@ async def cleanup(
         client_keys = await self._list_client_keys(workspace)
         removed = []
         summary = {}
+        if not workspace_info.persistent:
+            unload = True
+        else:
+            unload = False
         for client in client_keys:
             try:
                 assert (
@@ -1809,9 +1888,10 @@ async def cleanup(
                 )
             except Exception as e:
                 logger.error(f"Failed to ping client {client}: {e}")
+                user_info = UserInfo.model_validate(context["user"])
                 # Remove dead client
                 await self.delete_client(
-                    client, workspace, context["user"], unload=False, context=context
+                    client, workspace, user_info, unload=unload, context=context
                 )
                 removed.append(client)
         if removed:
@@ -1854,6 +1934,7 @@ def create_service(self, service_id, service_name=None):
             "get_summary": self.get_summary,
             "ping": self.ping_client,
             "cleanup": self.cleanup,
+            "wait_until_ready": self.wait_until_ready,
         }
         interface["config"].update(self._server_info)
         return interface
diff --git a/hypha/http.py b/hypha/http.py
index 74dd62d4..c65581a1 100644
--- a/hypha/http.py
+++ b/hypha/http.py
@@ -432,7 +432,7 @@ def norm_url(url):
         async def hypha_rpc_websocket_mjs(request: Request):
             if not self.rpc_lib_esm_content:
                 _rpc_lib_script = f"https://cdn.jsdelivr.net/npm/hypha-rpc@{main_version}/dist/hypha-rpc-websocket.mjs"
-                async with httpx.AsyncClient() as client:
+                async with httpx.AsyncClient(timeout=10) as client:
                     response = await client.get(_rpc_lib_script)
                     response.raise_for_status()
                     self.rpc_lib_esm_content = response.content
diff --git a/hypha/plugin_parser.py b/hypha/plugin_parser.py
index 76f10e1c..af377ec5 100644
--- a/hypha/plugin_parser.py
+++ b/hypha/plugin_parser.py
@@ -6,6 +6,7 @@
 from lxml import etree
 
 from hypha_rpc.utils import DefaultObjectProxy
+from hypha.core import ApplicationManifest
 
 tag_types = ["config", "script", "link", "window", "style", "docs", "attachment"]
 
@@ -108,28 +109,18 @@ def convert_config_to_artifact(plugin_config, plugin_id, source_url=None):
             "id": plugin_id,
         }
     )
-    if source_url:
-        artifact["source"] = source_url
+
     fields = [
-        "icon",
-        "name",
-        "version",
-        "api_version",
-        "description",
-        "license",
-        "requirements",
-        "dependencies",
-        "env",
-        "passive",
-        "services",
-        "entry_point",
+        field
+        for field in ApplicationManifest.model_fields.keys()
+        if field not in ["id", "type"]
     ]
     for field in fields:
         if field in plugin_config:
             artifact[field] = plugin_config[field]
+    if source_url:
+        artifact["source"] = source_url
     tags = plugin_config.get("labels", []) + plugin_config.get("flags", [])
-    if "bioengine" not in tags:
-        tags.append("bioengine")
     artifact["tags"] = tags
 
     docs = plugin_config.get("docs")
diff --git a/hypha/core/queue.py b/hypha/queue.py
similarity index 100%
rename from hypha/core/queue.py
rename to hypha/queue.py
diff --git a/hypha/runner/__init__.py b/hypha/runner/__init__.py
index ddbf9254..4ea67db7 100644
--- a/hypha/runner/__init__.py
+++ b/hypha/runner/__init__.py
@@ -113,19 +113,7 @@ async def run_server_app_worker(server_config):
 
     bar = BrowserAppRunner()
     await bar.initialize()
-    svc = await server.register_service(
-        {
-            "id": "server-app-worker",
-            "name": "Server App Worker",
-            "description": "A worker for running server apps",
-            "config": {"visibility": "protected"},
-            "start": bar.start,
-            "stop": bar.stop,
-            "list": bar.list,
-            "get_log": bar.get_log,
-            "shutdown": bar.shutdown,
-        }
-    )
+    svc = await server.register_service(bar.get_service())
     logger.info(
         f"Browser worker running at {server.config.public_base_url}/{server.config.workspace}/services/{svc['id'].split('/')[1]}"
     )
diff --git a/hypha/runner/browser.py b/hypha/runner/browser.py
index f921f367..da415f59 100644
--- a/hypha/runner/browser.py
+++ b/hypha/runner/browser.py
@@ -1,5 +1,6 @@
 """Provide a browser runner."""
 import uuid
+import asyncio
 import logging
 import sys
 from typing import Any, Dict, List, Optional, Union
@@ -48,6 +49,7 @@ class BrowserAppRunner:
 
     def __init__(
         self,
+        store,
         in_docker: bool = False,
     ):
         """Initialize the class."""
@@ -55,7 +57,9 @@ def __init__(
         self._browser_sessions = {}
         self.controller_id = str(BrowserAppRunner.instance_counter)
         BrowserAppRunner.instance_counter += 1
+        store.register_public_service(self.get_service())
         self.in_docker = in_docker
+        self._initialized = False
 
     async def initialize(self) -> None:
         """Initialize the app controller."""
@@ -83,7 +87,7 @@ async def start(
         self,
         url: str,
         session_id: Optional[str] = None,
-        time_limit: Optional[int] = None,
+        metadata: Optional[Dict[str, Any]] = None,
     ):
         """Start a browser app instance."""
         if session_id is None:
@@ -98,6 +102,7 @@ async def start(
             "status": "connecting",
             "page": page,
             "logs": logs,
+            "metadata": metadata,
         }
 
         _capture_logs_from_browser_tabs(page, logs)
@@ -115,10 +120,6 @@ async def start(
             await page.close()
             del self._browser_sessions[session_id]
             raise
-        if time_limit and time_limit > 0:
-            await page.evaluate(
-                f"setTimeout(() => {{ window.close() }} , {time_limit*1000})"
-            )
         return {
             "session_id": session_id,
             "status": "connected",
@@ -160,3 +161,24 @@ async def get_log(
                 offset : offset + limit
             ]
         raise Exception(f"browser app instance not found: {session_id}")
+
+    async def close_workspace(self, workspace: str) -> None:
+        """Close all browser app instances for a workspace."""
+        for session_id in list(self._browser_sessions.keys()):
+            if session_id.startswith(workspace + "/"):
+                await self.stop(session_id)
+
+    def get_service(self):
+        """Get the service."""
+        return {
+            "id": "server-app-worker",
+            "name": "Server App Worker",
+            "description": "A worker for running server apps",
+            "config": {"visibility": "protected"},
+            "start": self.start,
+            "stop": self.stop,
+            "list": self.list,
+            "get_log": self.get_log,
+            "shutdown": self.shutdown,
+            "close_workspace": self.close_workspace,
+        }
diff --git a/hypha/s3.py b/hypha/s3.py
index 8f352c8e..711a0b81 100644
--- a/hypha/s3.py
+++ b/hypha/s3.py
@@ -309,7 +309,9 @@ async def get_or_delete_file(
     async def load_workspace_info(self, workspace_name, user_info):
         """Load workspace from s3 record."""
         if not user_info.check_permission(workspace_name, UserPermission.read_write):
-            return None
+            raise PermissionError(
+                f"Permission denied: {workspace_name}, {user_info.id}"
+            )
         config_path = f"{self.workspace_etc_dir}/{workspace_name}/manifest.json"
         try:
             async with self.create_client_async() as s3_client:
@@ -527,6 +529,7 @@ async def cleanup_workspace(self, workspace: WorkspaceInfo, force=False):
                 await remove_objects_async(
                     s3client, self.workspace_bucket, workspace.id + "/"
                 )
+                logger.info("Workspace's S3 storage cleaned up: %s", workspace.id)
 
         if self.minio_client:
             try:
@@ -543,8 +546,6 @@ async def cleanup_workspace(self, workspace: WorkspaceInfo, force=False):
     async def setup_workspace(self, workspace: WorkspaceInfo):
         """Set up workspace."""
         assert isinstance(workspace, WorkspaceInfo)
-        if workspace.read_only:
-            return
         # Save the workspace info
         # workspace_dir = self.local_log_dir / workspace.id
         # os.makedirs(workspace_dir, exist_ok=True)
diff --git a/hypha/server.py b/hypha/server.py
index 45463694..c513fa42 100644
--- a/hypha/server.py
+++ b/hypha/server.py
@@ -14,7 +14,7 @@
 from hypha import __version__
 from hypha.core.auth import create_login_service
 from hypha.core.store import RedisStore
-from hypha.core.queue import create_queue_service
+from hypha.queue import create_queue_service
 from hypha.http import HTTPProxy
 from hypha.triton import TritonProxy
 from hypha.utils import GZipMiddleware, GzipRoute, PatchedCORSMiddleware
@@ -110,7 +110,9 @@ def start_builtin_services(
         assert args.enable_s3, "Server apps require S3 to be enabled"
         # pylint: disable=import-outside-toplevel
         from hypha.apps import ServerAppController
+        from hypha.runner import BrowserAppRunner
 
+        BrowserAppRunner(store, in_docker=args.in_docker)
         ServerAppController(
             store,
             port=args.port,
@@ -231,6 +233,7 @@ async def lifespan(app: FastAPI):
         reconnection_token_life_time=float(
             env.get("RECONNECTION_TOKEN_LIFE_TIME", str(2 * 24 * 60 * 60))
         ),
+        activity_check_interval=float(env.get("ACTIVITY_CHECK_INTERVAL", str(10))),
     )
 
     websocket_server = WebsocketServer(store, path=norm_url(args.base_path, "/ws"))
diff --git a/hypha/taskiq_utils/__init__.py b/hypha/taskiq_utils/__init__.py
new file mode 100644
index 00000000..e69de29b
diff --git a/hypha/taskiq_utils/exceptions.py b/hypha/taskiq_utils/exceptions.py
new file mode 100644
index 00000000..f939f843
--- /dev/null
+++ b/hypha/taskiq_utils/exceptions.py
@@ -0,0 +1,17 @@
+from taskiq.exceptions import ResultBackendError, ResultGetError, TaskiqError
+
+
+class TaskIQRedisError(TaskiqError):
+    """Base error for all taskiq-redis exceptions."""
+
+
+class DuplicateExpireTimeSelectedError(ResultBackendError, TaskIQRedisError):
+    """Error if two lifetimes are selected."""
+
+
+class ExpireTimeMustBeMoreThanZeroError(ResultBackendError, TaskIQRedisError):
+    """Error if two lifetimes are less or equal zero."""
+
+
+class ResultIsMissingError(TaskIQRedisError, ResultGetError):
+    """Error if there is no result when trying to get it."""
diff --git a/hypha/taskiq_utils/redis_backend.py b/hypha/taskiq_utils/redis_backend.py
new file mode 100644
index 00000000..5649cb42
--- /dev/null
+++ b/hypha/taskiq_utils/redis_backend.py
@@ -0,0 +1,199 @@
+"""This file contains modules from the taskiq-redis package
+We need to patch it so we can use fakeredis when redis is not available.
+The library was created by Pavel Kirilin, released under the MIT license.
+"""
+from fakeredis import aioredis
+from typing import (
+    Dict,
+    Optional,
+    TypeVar,
+    Union,
+)
+
+from taskiq import AsyncResultBackend
+from taskiq.abc.result_backend import TaskiqResult
+from taskiq.abc.serializer import TaskiqSerializer
+from taskiq.compat import model_dump, model_validate
+from taskiq.depends.progress_tracker import TaskProgress
+from taskiq.serializers import PickleSerializer
+
+from hypha.taskiq_utils.exceptions import (
+    DuplicateExpireTimeSelectedError,
+    ExpireTimeMustBeMoreThanZeroError,
+    ResultIsMissingError,
+)
+
+
+_ReturnType = TypeVar("_ReturnType")
+
+PROGRESS_KEY_SUFFIX = "__progress"
+
+
+class RedisAsyncResultBackend(AsyncResultBackend[_ReturnType]):
+    """Async result based on redis."""
+
+    def __init__(
+        self,
+        redis: aioredis.FakeRedis,
+        keep_results: bool = True,
+        result_ex_time: Optional[int] = None,
+        result_px_time: Optional[int] = None,
+        serializer: Optional[TaskiqSerializer] = None,
+    ) -> None:
+        """
+        Constructs a new result backend.
+
+        :param redis_url: url to redis.
+        :param keep_results: flag to not remove results from Redis after reading.
+        :param result_ex_time: expire time in seconds for result.
+        :param result_px_time: expire time in milliseconds for result.
+        :param max_connection_pool_size: maximum number of connections in pool.
+        :param connection_kwargs: additional arguments for redis BlockingConnectionPool.
+
+        :raises DuplicateExpireTimeSelectedError: if result_ex_time
+            and result_px_time are selected.
+        :raises ExpireTimeMustBeMoreThanZeroError: if result_ex_time
+            and result_px_time are equal zero.
+        """
+        self.redis = redis
+        self.serializer = serializer or PickleSerializer()
+        self.keep_results = keep_results
+        self.result_ex_time = result_ex_time
+        self.result_px_time = result_px_time
+
+        unavailable_conditions = any(
+            (
+                self.result_ex_time is not None and self.result_ex_time <= 0,
+                self.result_px_time is not None and self.result_px_time <= 0,
+            ),
+        )
+        if unavailable_conditions:
+            raise ExpireTimeMustBeMoreThanZeroError(
+                "You must select one expire time param and it must be more than zero.",
+            )
+
+        if self.result_ex_time and self.result_px_time:
+            raise DuplicateExpireTimeSelectedError(
+                "Choose either result_ex_time or result_px_time.",
+            )
+
+    async def shutdown(self) -> None:
+        """Closes redis connection."""
+        await super().shutdown()
+
+    async def set_result(
+        self,
+        task_id: str,
+        result: TaskiqResult[_ReturnType],
+    ) -> None:
+        """
+        Sets task result in redis.
+
+        Dumps TaskiqResult instance into the bytes and writes
+        it to redis.
+
+        :param task_id: ID of the task.
+        :param result: TaskiqResult instance.
+        """
+        redis_set_params: Dict[str, Union[str, int, bytes]] = {
+            "name": task_id,
+            "value": self.serializer.dumpb(model_dump(result)),
+        }
+        if self.result_ex_time:
+            redis_set_params["ex"] = self.result_ex_time
+        elif self.result_px_time:
+            redis_set_params["px"] = self.result_px_time
+
+        await self.redis.set(**redis_set_params)  # type: ignore
+
+    async def is_result_ready(self, task_id: str) -> bool:
+        """
+        Returns whether the result is ready.
+
+        :param task_id: ID of the task.
+
+        :returns: True if the result is ready else False.
+        """
+        return bool(await self.redis.exists(task_id))
+
+    async def get_result(
+        self,
+        task_id: str,
+        with_logs: bool = False,
+    ) -> TaskiqResult[_ReturnType]:
+        """
+        Gets result from the task.
+
+        :param task_id: task's id.
+        :param with_logs: if True it will download task's logs.
+        :raises ResultIsMissingError: if there is no result when trying to get it.
+        :return: task's return value.
+        """
+        if self.keep_results:
+            result_value = await self.redis.get(
+                name=task_id,
+            )
+        else:
+            result_value = await self.redis.getdel(
+                name=task_id,
+            )
+
+        if result_value is None:
+            raise ResultIsMissingError
+
+        taskiq_result = model_validate(
+            TaskiqResult[_ReturnType],
+            self.serializer.loadb(result_value),
+        )
+
+        if not with_logs:
+            taskiq_result.log = None
+
+        return taskiq_result
+
+    async def set_progress(
+        self,
+        task_id: str,
+        progress: TaskProgress[_ReturnType],
+    ) -> None:
+        """
+        Sets task progress in redis.
+
+        Dumps TaskProgress instance into the bytes and writes
+        it to redis with a standard suffix on the task_id as the key
+
+        :param task_id: ID of the task.
+        :param result: task's TaskProgress instance.
+        """
+        redis_set_params: Dict[str, Union[str, int, bytes]] = {
+            "name": task_id + PROGRESS_KEY_SUFFIX,
+            "value": self.serializer.dumpb(model_dump(progress)),
+        }
+        if self.result_ex_time:
+            redis_set_params["ex"] = self.result_ex_time
+        elif self.result_px_time:
+            redis_set_params["px"] = self.result_px_time
+
+        await self.redis.set(**redis_set_params)  # type: ignore
+
+    async def get_progress(
+        self,
+        task_id: str,
+    ) -> Union[TaskProgress[_ReturnType], None]:
+        """
+        Gets progress results from the task.
+
+        :param task_id: task's id.
+        :return: task's TaskProgress instance.
+        """
+        result_value = await self.redis.get(
+            name=task_id + PROGRESS_KEY_SUFFIX,
+        )
+
+        if result_value is None:
+            return None
+
+        return model_validate(
+            TaskProgress[_ReturnType],
+            self.serializer.loadb(result_value),
+        )
diff --git a/hypha/taskiq_utils/redis_broker.py b/hypha/taskiq_utils/redis_broker.py
new file mode 100644
index 00000000..3c80c273
--- /dev/null
+++ b/hypha/taskiq_utils/redis_broker.py
@@ -0,0 +1,117 @@
+"""This file contains modules from the taskiq-redis package
+We need to patch it so we can use fakeredis when redis is not available.
+The library was created by Pavel Kirilin, released under the MIT license.
+"""
+from fakeredis import aioredis
+from logging import getLogger
+from typing import AsyncGenerator, Callable, Optional, TypeVar
+
+from taskiq.abc.broker import AsyncBroker
+from taskiq.abc.result_backend import AsyncResultBackend
+from taskiq.message import BrokerMessage
+
+_T = TypeVar("_T")
+
+logger = getLogger("taskiq.redis_broker")
+
+
+class BaseRedisBroker(AsyncBroker):
+    """Base broker that works with Redis."""
+
+    def __init__(
+        self,
+        redis: aioredis.FakeRedis,
+        task_id_generator: Optional[Callable[[], str]] = None,
+        result_backend: Optional[AsyncResultBackend[_T]] = None,
+        queue_name: str = "taskiq",
+    ) -> None:
+        """
+        Constructs a new broker.
+
+        :param url: url to redis.
+        :param task_id_generator: custom task_id generator.
+        :param result_backend: custom result backend.
+        :param queue_name: name for a list in redis.
+        :param max_connection_pool_size: maximum number of connections in pool.
+            Each worker opens its own connection. Therefore this value has to be
+            at least number of workers + 1.
+        :param connection_kwargs: additional arguments for redis BlockingConnectionPool.
+        """
+        super().__init__(
+            result_backend=result_backend,
+            task_id_generator=task_id_generator,
+        )
+
+        self.redis = redis
+        self.queue_name = queue_name
+
+    async def shutdown(self) -> None:
+        """Closes redis connection pool."""
+        await super().shutdown()
+
+
+class PubSubBroker(BaseRedisBroker):
+    """Broker that works with Redis and broadcasts tasks to all workers."""
+
+    async def kick(self, message: BrokerMessage) -> None:
+        """
+        Publish message over PUBSUB channel.
+
+        :param message: message to send.
+        """
+        queue_name = message.labels.get("queue_name") or self.queue_name
+        await self.redis.publish(queue_name, message.message)
+
+    async def listen(self) -> AsyncGenerator[bytes, None]:
+        """
+        Listen redis queue for new messages.
+
+        This function listens to the pubsub channel
+        and yields all messages with proper types.
+
+        :yields: broker messages.
+        """
+
+        redis_pubsub_channel = self.redis.pubsub()
+        await redis_pubsub_channel.subscribe(self.queue_name)
+        async for message in redis_pubsub_channel.listen():
+            if not message:
+                continue
+            if message["type"] != "message":
+                logger.debug("Received non-message from redis: %s", message)
+                continue
+            yield message["data"]
+
+
+class ListQueueBroker(BaseRedisBroker):
+    """Broker that works with Redis and distributes tasks between workers."""
+
+    async def kick(self, message: BrokerMessage) -> None:
+        """
+        Put a message in a list.
+
+        This method appends a message to the list of all messages.
+
+        :param message: message to append.
+        """
+        queue_name = message.labels.get("queue_name") or self.queue_name
+        await self.redis.lpush(queue_name, message.message)
+
+    async def listen(self) -> AsyncGenerator[bytes, None]:
+        """
+        Listen redis queue for new messages.
+
+        This function listens to the queue
+        and yields new messages if they have BrokerMessage type.
+
+        :yields: broker messages.
+        """
+        redis_brpop_data_position = 1
+        while True:
+            try:
+                yield (await self.redis.brpop(self.queue_name))[
+                    redis_brpop_data_position
+                ]
+            except ConnectionError as exc:
+                logger.warning("Redis connection error: %s", exc)
+                continue
diff --git a/hypha/taskiq_utils/schedule_source.py b/hypha/taskiq_utils/schedule_source.py
new file mode 100644
index 00000000..379a53bc
--- /dev/null
+++ b/hypha/taskiq_utils/schedule_source.py
@@ -0,0 +1,92 @@
+"""This file contains modules from the taskiq-redis package
+We need to patch it so we can use fakeredis when redis is not available.
+The library was created by Pavel Kirilin, released under the MIT license.
+"""
+from fakeredis import aioredis
+from typing import List, Optional
+
+from taskiq import ScheduleSource
+from taskiq.abc.serializer import TaskiqSerializer
+from taskiq.compat import model_dump, model_validate
+from taskiq.scheduler.scheduled_task import ScheduledTask
+from taskiq.serializers import PickleSerializer
+
+
+class RedisScheduleSource(ScheduleSource):
+    """
+    Source of schedules for redis.
+
+    This class allows you to store schedules in redis.
+    Also it supports dynamic schedules.
+
+    :param url: url to redis.
+    :param prefix: prefix for redis schedule keys.
+    :param buffer_size: buffer size for redis scan.
+        This is how many keys will be fetched at once.
+    :param max_connection_pool_size: maximum number of connections in pool.
+    :param serializer: serializer for data.
+    :param connection_kwargs: additional arguments for redis BlockingConnectionPool.
+    """
+
+    def __init__(
+        self,
+        redis: aioredis.FakeRedis,
+        prefix: str = "schedule",
+        buffer_size: int = 50,
+        serializer: Optional[TaskiqSerializer] = None,
+    ) -> None:
+        self.prefix = prefix
+        self.redis = redis
+        self.buffer_size = buffer_size
+        if serializer is None:
+            serializer = PickleSerializer()
+        self.serializer = serializer
+
+    async def delete_schedule(self, schedule_id: str) -> None:
+        """Remove schedule by id."""
+        await self.redis.delete(f"{self.prefix}:{schedule_id}")
+
+    async def add_schedule(self, schedule: ScheduledTask) -> None:
+        """
+        Add schedule to redis.
+
+        :param schedule: schedule to add.
+        :param schedule_id: schedule id.
+        """
+        await self.redis.set(
+            f"{self.prefix}:{schedule.schedule_id}",
+            self.serializer.dumpb(model_dump(schedule)),
+        )
+
+    async def get_schedules(self) -> List[ScheduledTask]:
+        """
+        Get all schedules from redis.
+
+        This method is used by scheduler to get all schedules.
+
+        :return: list of schedules.
+        """
+        schedules = []
+
+        buffer = []
+        async for key in self.redis.scan_iter(f"{self.prefix}:*"):
+            buffer.append(key)
+            if len(buffer) >= self.buffer_size:
+                schedules.extend(await self.redis.mget(buffer))
+                buffer = []
+        if buffer:
+            schedules.extend(await self.redis.mget(buffer))
+        return [
+            model_validate(ScheduledTask, self.serializer.loadb(schedule))
+            for schedule in schedules
+            if schedule
+        ]
+
+    async def post_send(self, task: ScheduledTask) -> None:
+        """Delete a task after it's completed."""
+        if task.time is not None:
+            await self.delete_schedule(task.schedule_id)
+
+    async def shutdown(self) -> None:
+        """Shut down the schedule source."""
+        pass
diff --git a/hypha/templates/apps/login_template.html b/hypha/templates/apps/login_template.html
index dfb3b0f9..6768a875 100644
--- a/hypha/templates/apps/login_template.html
+++ b/hypha/templates/apps/login_template.html
@@ -79,6 +79,7 @@
         </div>
     </div>
     <script>
+        const HYPHA_VERSION = '{{ hypha_version }}';
         const AUTH0_AUDIENCE = '{{ auth0_audience }}';
         const auth0 = new window.auth0.Auth0Client({
             domain: '{{ auth0_domain }}',
@@ -94,12 +95,13 @@
         document.addEventListener('DOMContentLoaded', async () => {
             try {
                 await window.showUserProfile();
+                if (window.location.search.split('logout=')[1]) {
+                    logout();
+                }
             } catch (e) {
                 console.log("Not logged in, showing login button.");
             }
-            if (window.location.search.split('logout=')[1]) {
-                logout();
-            }
+            
         });
 
         const _generateToken = async (workspace, expires_in) =>{
@@ -172,6 +174,7 @@
             const exp = decodedToken.exp;
             const maxAge = exp - Math.floor(Date.now() / 1000);
             document.cookie = `access_token=${accessToken}; path=/; max-age=${maxAge}; samesite=lax`;
+            document.cookie = `hypha_version=${HYPHA_VERSION}; path=/; max-age=${maxAge}; samesite=lax`;
             document.getElementById("login-btn").style.display = "none";
             document.getElementById("name").innerHTML = user.name;
             document.getElementById("email").innerHTML = user.email;
@@ -191,6 +194,7 @@
             const workspace = urlParams.get('workspace') || ""
             const expires_in = urlParams.get('expires_in') || ""
             const report_url = "{{ login_service_url }}/report";
+            const redirect = urlParams.get('redirect');
 
             if (key && report_url.startsWith('http')) {
                 const url = `${report_url}?key=${key}&token=${accessToken}&workspace=${workspace}&expires_in=${expires_in}&email=${user.email}&email_verified=${user.email_verified}&name=${user.name}&nickname=${user.nickname}&user_id=${user.sub}&picture=${encodeURIComponent(user.picture)}`;
@@ -200,19 +204,33 @@
                     alert(`Failed to report token. Please try again later. (${response.status}: ${response.statusText})`);
                 }
                 else{
-                    setTimeout(() => {
-                        window.close();
-                    }, 500);
+                    if (redirect) {
+                        setTimeout(() => {
+                            window.location.href = redirect;
+                        }, 500);
+                    }
+                    else {
+                        setTimeout(() => {
+                            window.close();
+                        }, 500);
+                    }
                 }
             }
+            else if (redirect) {
+                setTimeout(() => {
+                    window.location.href = redirect;
+                }, 500);
+            }
         };
 
         window.logout = async () => {
-            let href = window.location.href;
-            if(!href.endsWith("/")){
-                href += "/";
+            const urlParams = new URLSearchParams(window.location.search);
+            let href = window.location.href.split('?')[0];
+            if (urlParams.get('redirect')) {
+                href = urlParams.get('redirect');
             }
             document.cookie = 'access_token=; path=/; max-age=0; samesite=lax';
+            document.cookie = 'hypha_version=; path=/; max-age=0; samesite=lax';
             await auth0.logout({ logoutParams: { returnTo: href } });
         };
 
diff --git a/hypha/templates/hypha-core-app/hypha-app-webpython.js b/hypha/templates/hypha-core-app/hypha-app-webpython.js
index 180038a9..42bf0f4b 100644
--- a/hypha/templates/hypha-core-app/hypha-app-webpython.js
+++ b/hypha/templates/hypha-core-app/hypha-app-webpython.js
@@ -27,7 +27,7 @@ loadPyodide().then(async (pyodide) => {
     pyodide.setStderr({ batched: (msg) => console.error(msg) });
     await pyodide.loadPackage("micropip");
     const micropip = pyodide.pyimport("micropip");
-    await micropip.install('hypha-rpc==0.20.41');
+    await micropip.install('hypha-rpc==0.20.42');
     const isWindow = typeof window !== "undefined";
     
     setTimeout(() => {
diff --git a/hypha/templates/index.html b/hypha/templates/index.html
index edbe14cb..f60d1fe3 100644
--- a/hypha/templates/index.html
+++ b/hypha/templates/index.html
@@ -209,6 +209,7 @@ <h3 className="text-xl font-bold">{title}</h3>
                         const exp = decodedToken.exp;
                         const maxAge = exp - Math.floor(Date.now() / 1000);
                         document.cookie = `access_token=${accessToken}; path=/; max-age=${maxAge}; samesite=lax`;
+                        document.cookie = `hypha_version=${config.hypha_version}; path=/; max-age=${maxAge}; samesite=lax`;
                     });
                 }
             }, [isLoggedIn, user]);
@@ -244,6 +245,7 @@ <h3 className="text-xl font-bold">{title}</h3>
 
             const onLogout = async () => {
                 document.cookie = 'access_token=; path=/; max-age=0; samesite=lax';
+                document.cookie = 'hypha_version=; path=/; max-age=0; samesite=lax';
                 await window.auth0Client.logout({
                     logoutParams: { returnTo: window.location.origin }
                 });
diff --git a/hypha/templates/ws/index.html b/hypha/templates/ws/index.html
index 89a5e18e..295e3705 100644
--- a/hypha/templates/ws/index.html
+++ b/hypha/templates/ws/index.html
@@ -160,7 +160,7 @@
 <body class="bg-black text-white font-poppins">
     <div id="app"></div>
     <script type="module">
-        import { HyphaCore } from "https://cdn.jsdelivr.net/npm/hypha-core@0.20.41/dist/hypha-core.mjs";
+        import { HyphaCore } from "https://cdn.jsdelivr.net/npm/hypha-core@0.20.42/dist/hypha-core.mjs";
 
         const defaultService = {
             getServerConfig(context){
@@ -1777,8 +1777,16 @@ <h1 className="text-4xl font-bold capitalize">Artifacts</h1>
                         setWs(remoteServer);
                     }
                     catch(e){
-                        alert(`Failed to connect to the server: ${e.message}`);
-                        window.location.href = '/';
+                        if(e.message === "Permission denied"){
+                            const redirectUrl = '/public/apps/hypha-login/?redirect=' + window.location.href;
+                            alert(`Failed to connect to the server: ${e.message}, redirecting to login: ${redirectUrl}`);
+                            setTimeout(() => {
+                                window.location.href = redirectUrl;
+                            }, 1000);
+                        }
+                        else{
+                            alert(`Failed to connect to the server: ${e.message}`);
+                        }
                     }
                 };
                 connectToServer();
diff --git a/hypha/vectors.py b/hypha/vectors.py
index a0f43ec4..5b635522 100644
--- a/hypha/vectors.py
+++ b/hypha/vectors.py
@@ -4,7 +4,9 @@
 import logging
 import sys
 import re
+import json
 from typing import Any, List, Dict, Optional
+from hypha.utils import list_objects_async, remove_objects_async
 from fakeredis import aioredis
 from redis.commands.search.field import (
     TagField,
@@ -68,6 +70,7 @@ def __init__(self, redis, prefix: str = "vs", cache_dir=None):
         self._redis: aioredis.FakeRedis = redis
         self._index_name_prefix = prefix
         self._cache_dir = cache_dir
+        self._maximum_search_results = None
 
     def _get_index_name(self, collection_name: str) -> str:
         return (
@@ -78,7 +81,10 @@ def _get_index_name(self, collection_name: str) -> str:
 
     async def _get_fields(self, collection_name: str):
         index_name = self._get_index_name(collection_name)
-        info = await self._redis.ft(index_name).info()
+        try:
+            info = await self._redis.ft(index_name).info()
+        except aioredis.ResponseError:
+            raise KeyError(f"Vector collection {collection_name} does not exist.")
         fields = parse_attributes(info["attributes"])
         return fields
 
@@ -86,6 +92,10 @@ async def create_collection(
         self,
         collection_name: str,
         vector_fields: List[Dict[str, Any]],
+        overwrite: bool = False,
+        s3_client=None,
+        bucket=None,
+        prefix=None,
     ):
         """
         Creates a RedisSearch index collection.
@@ -124,6 +134,18 @@ async def create_collection(
                 redis_fields.append(field_class(field_name, algorithm, attributes))
             else:
                 redis_fields.append(field_class(name=field_name))
+        # Check if the collection already exists, remove it if overwrite is True
+        # otherwise raise an error
+        try:
+            await self._redis.ft(index_name).info()
+            if overwrite:
+                # Drop the index
+                await self._redis.ft(index_name).dropindex(delete_documents=False)
+            else:
+                raise ValueError(f"Collection {collection_name} already exists.")
+        # ResponseError is raised if the collection does not exist
+        except aioredis.ResponseError:
+            pass
 
         # Create the index
         await self._redis.ft(index_name).create_index(
@@ -132,7 +154,11 @@ async def create_collection(
                 prefix=[f"{index_name}:"], index_type=IndexType.HASH
             ),
         )
-        fields_info = [f"{field.name} ({type(field)})" for field in redis_fields]
+        if s3_client:
+            await self._dump_index(collection_name, s3_client, bucket, prefix)
+        fields_info = [
+            f"{field.name} ({type(field).__name__})" for field in redis_fields
+        ]
         logger.info(f"Collection {collection_name} created with fields: {fields_info}.")
 
     async def count(self, collection_name: str):
@@ -141,15 +167,187 @@ async def count(self, collection_name: str):
         results = await self._redis.ft(index_name).search(count_query)
         return results.total
 
-    async def delete_collection(self, collection_name: str):
+    async def delete_collection(
+        self, collection_name: str, s3_client=None, bucket=None, prefix=None
+    ):
         index_name = self._get_index_name(collection_name)
-        await self._redis.ft(index_name).dropindex()
-        logger.info(f"Collection {collection_name} deleted.")
+
+        # Drop the index
+        await self._redis.ft(index_name).dropindex(delete_documents=True)
+        if s3_client:
+            if not prefix.endswith("/"):
+                prefix = f"{prefix}/"
+            await remove_objects_async(s3_client, bucket, prefix)
+
+        logger.info(f"Collection {collection_name} and all associated keys deleted.")
+
+    async def _dump_vectors(
+        self,
+        vectors: List[Dict[str, Any]],
+        fields: Dict[str, Any],
+        s3_client,
+        bucket: str,
+        prefix: str,
+    ):
+        if not prefix.endswith("/"):
+            prefix = f"{prefix}/"
+        # Dump vector data for the current page
+        for vector in vectors:
+            vector_id = vector["id"]
+            vector_data = {key: value for key, value in vector.items()}
+
+            # Convert vector bytes to list for JSON serialization
+            for key, value in vector_data.items():
+                if key not in fields:
+                    continue
+                if fields[key]["type"] == "VECTOR":
+                    vector_data[key] = (
+                        np.frombuffer(value, dtype=np.float32).astype(float).tolist()
+                    )
+
+            vector_key = f"{prefix}{vector_id}.json"
+            await s3_client.put_object(
+                Bucket=bucket,
+                Key=vector_key,
+                Body=json.dumps(vector_data),
+            )
+
+    async def _dump_index(
+        self, collection_name: str, s3_client, bucket: str, prefix: str
+    ):
+        if not prefix.endswith("/"):
+            prefix = f"{prefix}/"
+        # Dump index settings
+        fields = await self._get_fields(collection_name)
+        dump_fields = []
+        for field in fields.values():
+            field_data = {
+                "type": field["type"],
+                "name": field["identifier"],
+            }
+            if field["type"] == "VECTOR":
+                field_data["algorithm"] = field["algorithm"]
+                field_data["attributes"] = {
+                    "TYPE": field["data_type"],
+                    "DIM": field["dim"],
+                    "DISTANCE_METRIC": field["distance_metric"],
+                }
+            dump_fields.append(field_data)
+        index_settings = {
+            "collection_name": collection_name,
+            "index_name": self._get_index_name(collection_name),
+            "fields": dump_fields,
+        }
+        index_key = f"{prefix}_index.json"
+        await s3_client.put_object(
+            Bucket=bucket,
+            Key=index_key,
+            Body=json.dumps(index_settings),
+        )
+
+    async def dump_collection(
+        self,
+        collection_name: str,
+        s3_client,
+        bucket: str,
+        prefix: str,
+        page_size: int = 100,
+    ):
+        fields = await self._get_fields(collection_name)
+        if not prefix.endswith("/"):
+            prefix = f"{prefix}/"
+        offset = 0
+        while True:
+            # Fetch vectors in pages
+            vectors = await self.list_vectors(
+                collection_name,
+                offset=offset,
+                limit=page_size,
+                return_fields=list(fields.keys()),
+            )
+            if not vectors:  # Exit loop if no more vectors are found
+                break
+
+            if s3_client:
+                # Dump vectors to S3
+                await self._dump_vectors(vectors, fields, s3_client, bucket, prefix)
+
+            # Move to the next page
+            offset += page_size
+
+        if s3_client:
+            await self._dump_index(collection_name, s3_client, bucket, prefix)
+        logger.info(
+            f"Collection {collection_name} dumped to S3 bucket {bucket} under prefix {prefix}."
+        )
+
+    async def load_collection(
+        self,
+        collection_name: str,
+        s3_client,
+        bucket: str,
+        prefix: str,
+        overwrite: bool = False,
+    ):
+        if not prefix.endswith("/"):
+            prefix = f"{prefix}/"
+        index_key = f"{prefix}_index.json"
+
+        # Load index settings
+        index_response = await s3_client.get_object(Bucket=bucket, Key=index_key)
+        index_settings = json.loads(await index_response["Body"].read())
+        await self.create_collection(
+            collection_name, index_settings["fields"], overwrite=overwrite
+        )
+
+        # List vector files
+
+        vector_prefix = prefix
+        vector_objects = await list_objects_async(s3_client, bucket, vector_prefix)
+
+        # Load vectors
+        for obj in vector_objects:
+            if obj["name"].endswith(".json") and not obj["name"].endswith(
+                "_index.json"
+            ):
+                if obj["type"] != "file":
+                    continue
+                key = f"{vector_prefix}{obj['name']}"
+                vector_response = await s3_client.get_object(Bucket=bucket, Key=key)
+                vector_data = json.loads(await vector_response["Body"].read())
+
+                # Convert vector lists back to bytes
+                for key, value in vector_data.items():
+                    if isinstance(value, list):
+                        vector_data[key] = np.array(value, dtype=np.float32)
+
+                try:
+                    # Add vector to Redis
+                    await self.add_vectors(
+                        collection_name,
+                        [{"id": obj["name"].split(".")[0], **vector_data}],
+                    )
+                except Exception as e:
+                    logger.error(
+                        f"Failed to load vector {obj['name']} from S3 bucket {bucket} under prefix {prefix}: {e}"
+                    )
+                    raise e
+
+        logger.info(
+            f"Collection {collection_name} loaded from S3 bucket {bucket} under prefix {prefix}."
+        )
 
     async def list_collections(self):
         # use redis.ft to list all collections
         collections = await self._redis.execute_command("FT._LIST")
-        return [c.decode("utf-8") for c in collections]
+
+        items = []
+        for c in collections:
+            # remove the prefix from the collection names
+            items.append(
+                {"name": c.decode("utf-8").replace(f"{self._index_name_prefix}:", "")}
+            )
+        return items
 
     async def _embed_texts(
         self,
@@ -175,17 +373,23 @@ async def add_vectors(
         self,
         collection_name: str,
         vectors: List[Dict[str, Any]],
+        update: bool = False,
         embedding_models: Optional[Dict[str, str]] = None,
+        s3_client=None,
+        bucket=None,
+        prefix=None,
     ):
         index_name = self._get_index_name(collection_name)
         fields = await self._get_fields(collection_name)
         ids = []
         for vector in vectors:
-            if "_id" in vector:
-                _id = vector["_id"]
-                del vector["_id"]
+            if "id" in vector:
+                _id = vector["id"]
             else:
+                if update:
+                    raise ValueError("ID is required for update operation.")
                 _id = str(uuid.uuid4())
+                vector["id"] = _id
             ids.append(_id)
             # convert numpy arrays to bytes
             for key, value in vector.items():
@@ -209,6 +413,10 @@ async def add_vectors(
                 mapping=vector,
             )
 
+        if s3_client:
+            assert bucket and prefix, "S3 bucket and prefix must be provided."
+            await self._dump_vectors(vectors, fields, s3_client, bucket, prefix)
+
         logger.info(f"Added {len(vectors)} vectors to collection {collection_name}.")
         return ids
 
@@ -216,6 +424,8 @@ def _format_docs(self, docs, index_name):
         formated = []
         for doc in docs:
             d = vars(doc)
+            if "payload" in d:
+                del d["payload"]
             if "id" in d:
                 d["id"] = d["id"].replace(f"{index_name}:", "")
             formated.append(d)
@@ -304,6 +514,7 @@ async def search_vectors(
         """
         index_name = self._get_index_name(collection_name)
         collection_fields = await self._get_fields(collection_name)
+        fields = {field["identifier"]: field for field in collection_fields.values()}
         all_fields = [field for field in collection_fields.keys()] + ["score"]
         # Generate embedding if text_query is provided
         if query:
@@ -384,12 +595,20 @@ async def search_vectors(
         # Build the RedisSearch query
         query = (
             Query(query_string)
-            .return_fields(*return_fields)
             .sort_by(order_by, asc=True)
             .paging(offset, limit)
             .dialect(2)
         )
-
+        for field in fields.values():
+            if field["identifier"] in return_fields:
+                if field["type"] == "VECTOR":
+                    query.return_field(field["identifier"], decode_field=False)
+                else:
+                    query.return_field(
+                        field["identifier"], decode_field=True, encoding="utf-8"
+                    )
+        if "score" in return_fields:
+            query.return_field("score")
         # Perform the search using the RedisSearch index
         results = await self._redis.ft(index_name).search(
             query, query_params=query_params
@@ -417,10 +636,20 @@ async def search_vectors(
         else:
             return docs
 
-    async def remove_vectors(self, collection_name: str, ids: List[str]):
+    async def remove_vectors(
+        self,
+        collection_name: str,
+        ids: List[str],
+        s3_client=None,
+        bucket=None,
+        prefix=None,
+    ):
         index_name = self._get_index_name(collection_name)
         for id in ids:
             await self._redis.delete(f"{index_name}:{id}")
+            if s3_client:
+                key = f"{prefix}{id}.json"
+                await s3_client.delete_object(Bucket=bucket, Key=key)
         logger.info(f"Removed {len(ids)} vectors from collection {collection_name}.")
 
     async def get_vector(self, collection_name: str, id: str):
@@ -431,6 +660,20 @@ async def get_vector(self, collection_name: str, id: str):
 
         return {key.decode("utf-8"): value for key, value in vector.items()}
 
+    async def _get_max_search_results(self) -> int:
+        try:
+            result = await self._redis.execute_command(
+                "FT.CONFIG", "GET", "MAXSEARCHRESULTS"
+            )
+            result = result[0]
+            config = {
+                result[i].decode(): result[i + 1].decode()
+                for i in range(0, len(result), 2)
+            }
+            return int(config.get("MAXSEARCHRESULTS", 0))  # Default to 0 if not found
+        except Exception as e:
+            raise RuntimeError(f"Error retrieving MAXSEARCHRESULTS: {e}")
+
     async def list_vectors(
         self,
         collection_name: str,
@@ -455,16 +698,27 @@ async def list_vectors(
                 raise ValueError(f"Invalid order_by field: {order_by}")
 
         query_string = "*"
+        if limit is None:
+            if not self._maximum_search_results:
+                self._maximum_search_results = await self._get_max_search_results()
+            limit = self._maximum_search_results
+        if offset is None:
+            offset = 0
 
         query = (
             Query(query_string)
-            .return_fields(*return_fields)
             .paging(offset, limit)
             .dialect(2)
             .sort_by(order_by, asc=True)
         )
-
-        # Perform the search using the RedisSearch index
+        for field in fields.values():
+            if field["identifier"] in return_fields:
+                if field["type"] == "VECTOR":
+                    query.return_field(field["identifier"], decode_field=False)
+                else:
+                    query.return_field(
+                        field["identifier"], decode_field=True, encoding="utf-8"
+                    )
         results = await self._redis.ft(index_name).search(query)
         docs = self._format_docs(results.docs, index_name)
 
diff --git a/requirements.txt b/requirements.txt
index 9263779c..cd23b7ad 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -3,7 +3,7 @@ aiofiles==23.2.1
 websockets==13.1
 base58==2.1.1
 fastapi==0.106.0
-hypha-rpc==0.20.41
+hypha-rpc==0.20.42
 jinja2==3.1.4
 lxml==4.9.3
 msgpack==1.0.8
@@ -21,7 +21,7 @@ uvicorn==0.23.2
 httpx==0.27.0
 pyotritonclient==0.2.6
 fakeredis==2.24.1
-redis==4.6.0
+redis==5.2.0
 friendlywords==1.1.3
 aiocache==0.12.2
 jsonschema==3.2.0
@@ -38,3 +38,5 @@ ollama==0.3.3
 fastembed==0.4.2
 asgiproxy==0.1.1
 stream-zip==0.0.83
+greenlet==3.0.3
+taskiq==0.11.9
diff --git a/setup.py b/setup.py
index e608a474..420f21b6 100644
--- a/setup.py
+++ b/setup.py
@@ -12,7 +12,7 @@
 REQUIREMENTS = [
     "aiofiles",
     "fastapi>=0.70.0,<=0.106.0",
-    "hypha-rpc>=0.20.41",
+    "hypha-rpc>=0.20.42",
     "msgpack>=1.0.2",
     "numpy",
     "pydantic[email]>=2.6.1",
@@ -37,6 +37,7 @@
     "aiocache>=0.12.2",
     "jsonschema>=3.2.0",
     "sqlalchemy>=2.0.35",
+    "greenlet>=3.0.3",
     "aiosqlite>=0.20.0",
     "prometheus-client>=0.21.0",
     "uuid-utils>=0.9.0",
@@ -45,6 +46,7 @@
     "hrid>=0.2.4",
     "asgiproxy>=0.1.1",
     "stream-zip>=0.0.83",
+    "taskiq>=0.11.9",
 ]
 
 ROOT_DIR = Path(__file__).parent.resolve()
@@ -73,7 +75,7 @@
             "aiobotocore>=2.1.0",
         ],
         "server-apps": [
-            "redis>=4.5.5",
+            "redis>=5.2.0",
             "aiobotocore>=2.1.0",
             "aiortc>=1.9.0",
             "requests>=2.26.0",
diff --git a/tests/conftest.py b/tests/conftest.py
index 445cf87c..4d17e50a 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -9,7 +9,6 @@
 import uuid
 from threading import Thread
 
-import pytest_asyncio
 import requests
 from requests import RequestException
 import pytest_asyncio
@@ -42,6 +41,7 @@
 
 JWT_SECRET = str(uuid.uuid4())
 os.environ["JWT_SECRET"] = JWT_SECRET
+os.environ["ACTIVITY_CHECK_INTERVAL"] = "0.3"
 test_env = os.environ.copy()
 
 
@@ -228,7 +228,7 @@ def redis_server():
     try:
         r = Redis(host="localhost", port=REDIS_PORT)
         r.ping()
-        yield
+        yield f"redis://127.0.0.1:{REDIS_PORT}/0"
     except Exception:
         # Pull the Redis image
         subprocess.run(["docker", "pull", "redis/redis-stack:7.2.0-v13"], check=True)
@@ -253,7 +253,7 @@ def redis_server():
                 pass
             timeout -= 0.1
             time.sleep(0.1)
-        yield
+        yield f"redis://127.0.0.1:{REDIS_PORT}/0"
         subprocess.Popen(["docker", "stop", "redis"])
         subprocess.Popen(["docker", "rm", "redis"])
         time.sleep(1)
@@ -505,7 +505,7 @@ def minio_server_fixture():
             timeout -= 0.1
             time.sleep(0.1)
         print("\nMinio server started.")
-        yield
+        yield MINIO_SERVER_URL
 
         proc.terminate()
         # shutil.rmtree(dirpath, ignore_errors=True)
diff --git a/tests/testASGIWebPythonPlugin.imjoy.html b/tests/testASGIWebPythonPlugin.imjoy.html
index c8de2b93..67256c3d 100644
--- a/tests/testASGIWebPythonPlugin.imjoy.html
+++ b/tests/testASGIWebPythonPlugin.imjoy.html
@@ -16,7 +16,7 @@
     "flags": [],
     "icon": "extension",
     "api_version": "0.1.7",
-    "env": "",
+    "env": {},
     "requirements": ["fastapi==0.112.1"],
     "dependencies": []
 }
diff --git a/tests/testFunctionsPlugin.imjoy.html b/tests/testFunctionsPlugin.imjoy.html
index a9aa8a85..57e5a31b 100644
--- a/tests/testFunctionsPlugin.imjoy.html
+++ b/tests/testFunctionsPlugin.imjoy.html
@@ -15,7 +15,7 @@
     "inputs": null,
     "outputs": null,
     "api_version": "0.1.7",
-    "env": "",
+    "env": {},
     "requirements": [],
     "dependencies": []
 }
diff --git a/tests/testUnreliablePlugin.imjoy.html b/tests/testUnreliablePlugin.imjoy.html
index 688fb547..5c69d080 100644
--- a/tests/testUnreliablePlugin.imjoy.html
+++ b/tests/testUnreliablePlugin.imjoy.html
@@ -14,7 +14,7 @@
     "icon": "extension",
     "inputs": null,
     "outputs": null,
-    "env": "",
+    "env": {},
     "requirements": [],
     "dependencies": [],
     "defaults": {"w": 7, "h": 7}
diff --git a/tests/testWebPythonPlugin.imjoy.html b/tests/testWebPythonPlugin.imjoy.html
index c20d7b62..4dbc9267 100644
--- a/tests/testWebPythonPlugin.imjoy.html
+++ b/tests/testWebPythonPlugin.imjoy.html
@@ -16,7 +16,7 @@
     "flags": [],
     "icon": "extension",
     "api_version": "0.1.7",
-    "env": "",
+    "env": {},
     "requirements": [],
     "dependencies": []
 }
diff --git a/tests/testWebWorkerPlugin.imjoy.html b/tests/testWebWorkerPlugin.imjoy.html
index b1fa8caa..509b7aa6 100644
--- a/tests/testWebWorkerPlugin.imjoy.html
+++ b/tests/testWebWorkerPlugin.imjoy.html
@@ -15,7 +15,7 @@
     "inputs": null,
     "outputs": null,
     "api_version": "0.1.7",
-    "env": "",
+    "env": {},
     "requirements": [],
     "dependencies": []
 }
diff --git a/tests/testWindowPlugin1.imjoy.html b/tests/testWindowPlugin1.imjoy.html
index 9da5e690..0380cdb6 100644
--- a/tests/testWindowPlugin1.imjoy.html
+++ b/tests/testWindowPlugin1.imjoy.html
@@ -14,7 +14,7 @@
     "icon": "extension",
     "inputs": null,
     "outputs": null,
-    "env": "",
+    "env": {},
     "requirements": [],
     "dependencies": [],
     "defaults": {"w": 7, "h": 7}
diff --git a/tests/test_event_bus.py b/tests/test_event_bus.py
index 822ef3e9..3fdc65b1 100644
--- a/tests/test_event_bus.py
+++ b/tests/test_event_bus.py
@@ -145,7 +145,7 @@ def handler(data):
 
 
 async def test_stop(event_bus):
-    event_bus.stop()
+    await event_bus.stop()
     assert event_bus._stop is True
 
 
diff --git a/tests/test_redis.py b/tests/test_redis.py
index 1f7ea659..e9e32137 100644
--- a/tests/test_redis.py
+++ b/tests/test_redis.py
@@ -4,6 +4,7 @@
 import numpy as np
 import pytest
 import pytest_asyncio
+from datetime import datetime, timedelta
 from hypha_rpc import connect_to_server
 
 from hypha.core.store import RedisStore
@@ -13,7 +14,7 @@
 pytestmark = pytest.mark.asyncio
 
 
-@pytest_asyncio.fixture()
+@pytest_asyncio.fixture(name="redis_store")
 async def redis_store():
     """Represent the redis store."""
     store = RedisStore(None, redis_uri=None)
@@ -87,6 +88,67 @@ def echo(data):
     assert await service.echo("hello") == "hello"
     assert await service.echo("hello") == "hello"
 
+    # External state to track task execution
+    execution_tracker = {"executed": False, "data": None}
+
+    # Define a dummy task that updates the tracker
+    async def dummy_task(data):
+        execution_tracker["executed"] = True
+        execution_tracker["data"] = data
+        return f"Processed {data}"
+
+    # Schedule the task to run immediately
+    task = await redis_store.run_task(dummy_task, "test-data")
+
+    # Assert that the task ID is valid
+    assert task.task_id is not None
+
+    # Allow enough time for the task to execute
+    await asyncio.sleep(0.5)
+
+    # Fetch the result of the task
+    result = await redis_store.get_task_result(task.task_id)
+    assert result.return_value == "Processed test-data"
+    assert execution_tracker["executed"] is True
+    assert execution_tracker["data"] == "test-data"
+
+    # Test scheduling a task using a cron expression
+    execution_tracker["executed"] = False  # Reset the tracker
+    cron_expr = "*/1 * * * *"
+    schedule = await redis_store.schedule_task(
+        dummy_task, "cron-test-data", corn=cron_expr, schedule_id="cron-test"
+    )
+
+    # Assert that the schedule ID is valid
+    assert schedule.schedule_id == "cron-test"
+
+    # Allow some time for the cron job to execute
+    await asyncio.sleep(65)  # Wait for at least one minute
+
+    # Check that the cron job executed
+    assert execution_tracker["executed"] is True
+    assert execution_tracker["data"] == "cron-test-data"
+
+    # Clean up the scheduled task
+    await redis_store.delete_schedule("cron-test")
+
+    # Test scheduling a task with a specific time
+    execution_tracker["executed"] = False  # Reset the tracker
+    scheduled_time = datetime.now() + timedelta(seconds=2)
+    schedule = await redis_store.schedule_task(
+        dummy_task, "scheduled-test-data", time=scheduled_time, schedule_id="scheduled"
+    )
+
+    # Assert that the schedule ID is valid
+    assert schedule.schedule_id == "scheduled"
+
+    # Disable schedule check for now due to an error
+    # Allow enough time for the task to execute
+    # await asyncio.sleep(10
+    # Verify that the task executed
+    # assert execution_tracker["executed"] is True
+    # assert execution_tracker["data"] == "scheduled-test-data"
+
 
 async def test_websocket_server(fastapi_server, test_user_token_2):
     """Test the websocket server."""
diff --git a/tests/test_server_apps.py b/tests/test_server_apps.py
index 7c9cf633..085732a9 100644
--- a/tests/test_server_apps.py
+++ b/tests/test_server_apps.py
@@ -37,7 +37,7 @@
 """
 
 
-async def test_server_apps_unauthorized(
+async def test_server_apps_workspace_removal(
     fastapi_server, test_user_token_5, root_user_token
 ):
     """Test the server apps."""
@@ -147,6 +147,100 @@ async def test_server_apps(fastapi_server, test_user_token):
     await controller.stop(config.id)
 
 
+async def test_singleton_apps(fastapi_server, test_user_token):
+    """Test the singleton apps."""
+    api = await connect_to_server(
+        {
+            "name": "test client",
+            "server_url": WS_SERVER_URL,
+            "method_timeout": 30,
+            "token": test_user_token,
+        }
+    )
+    controller = await api.get_service("public/server-apps")
+
+    # Launch the first instance of the singleton app
+    config1 = await controller.launch(
+        source=TEST_APP_CODE,
+        config={"type": "window", "singleton": True},
+        overwrite=True,
+        wait_for_service="default",
+    )
+    assert "id" in config1.keys()
+
+    # Try launching another instance of the same singleton app
+    with pytest.raises(Exception, match="is a singleton app and already running"):
+        await controller.launch(
+            source=TEST_APP_CODE,
+            config={"type": "window", "singleton": True},
+            overwrite=True,
+            wait_for_service="default",
+        )
+
+    # Stop the singleton app
+    await controller.stop(config1.id)
+
+    # Verify the singleton app is no longer running
+    running_apps = await controller.list_running()
+    assert not any(app["id"] == config1.id for app in running_apps)
+
+
+async def test_daemon_apps(fastapi_server, test_user_token, root_user_token):
+    """Test the daemon apps."""
+    async with connect_to_server(
+        {
+            "name": "test client",
+            "server_url": WS_SERVER_URL,
+            "method_timeout": 30,
+            "token": test_user_token,
+        }
+    ) as api:
+        controller = await api.get_service("public/server-apps")
+
+        # Launch a daemon app
+        config = await controller.launch(
+            source=TEST_APP_CODE,
+            config={"type": "window", "daemon": True},
+            overwrite=True,
+            wait_for_service="default",
+        )
+        assert "id" in config.keys()
+
+        # Verify the daemon app is running
+        running_apps = await controller.list_running()
+        assert any(app["id"] == config.id for app in running_apps)
+        # await controller.stop(config.id)
+
+        apps = await controller.list_apps()
+        assert find_item(apps, "id", config.app_id)
+
+    await asyncio.sleep(1)
+
+    async with connect_to_server(
+        {"server_url": WS_SERVER_URL, "client_id": "admin", "token": root_user_token}
+    ) as root:
+        admin = await root.get_service("admin-utils")
+        await admin.unload_workspace(api.config["workspace"], wait=True, timeout=60)
+        workspaces = await admin.list_workspaces()
+        assert not find_item(workspaces, "id", api.config["workspace"])
+
+    async with connect_to_server(
+        {
+            "name": "test client",
+            "server_url": WS_SERVER_URL,
+            "method_timeout": 30,
+            "token": test_user_token,
+        }
+    ) as api:
+        controller = await api.get_service("public/server-apps")
+        apps = await controller.list_apps()
+        await api.wait_until_ready()
+        controller = await api.get_service("public/server-apps")
+        running_apps = await controller.list_running()
+        # The daemon app should be running
+        assert any(app["app_id"] == config.app_id for app in running_apps)
+
+
 async def test_web_python_apps(fastapi_server, test_user_token):
     """Test webpython app."""
     api = await connect_to_server(
@@ -279,7 +373,7 @@ async def test_lazy_plugin(fastapi_server, test_user_token):
     apps = await controller.list_apps()
     assert find_item(apps, "id", app_info.id)
 
-    app = await api.get_service(f"echo@{app_info.id}")
+    app = await api.get_service(f"echo@{app_info.id}", mode="first")
     assert app is not None
 
     await controller.uninstall(app_info.id)
@@ -287,6 +381,45 @@ async def test_lazy_plugin(fastapi_server, test_user_token):
     await api.disconnect()
 
 
+async def test_stop_after_inactive(fastapi_server, test_user_token):
+    """Test app with time limit."""
+    api = await connect_to_server(
+        {
+            "name": "test client",
+            "server_url": WS_SERVER_URL,
+            "method_timeout": 30,
+            "token": test_user_token,
+        }
+    )
+
+    # Test app with custom template
+    controller = await api.get_service("public/server-apps")
+
+    source = (
+        (Path(__file__).parent / "testWebWorkerPlugin.imjoy.html")
+        .open(encoding="utf-8")
+        .read()
+    )
+
+    app_info = await controller.install(
+        source=source,
+        overwrite=True,
+    )
+
+    apps = await controller.list_apps()
+    assert find_item(apps, "id", app_info.id)
+
+    # start the app
+    app = await controller.start(app_info.id, stop_after_inactive=1)
+    apps = await controller.list_running()
+    assert find_item(apps, "id", app.id) is not None
+    await asyncio.sleep(2)
+    apps = await controller.list_running()
+    assert find_item(apps, "id", app.id) is None
+    await controller.uninstall(app_info.id)
+    await api.disconnect()
+
+
 async def test_lazy_service(fastapi_server, test_user_token):
     """Test lazy service loading."""
     api = await connect_to_server(
@@ -313,7 +446,7 @@ async def test_lazy_service(fastapi_server, test_user_token):
         overwrite=True,
     )
 
-    service = await api.get_service("echo@" + app_info.id)
+    service = await api.get_service("echo@" + app_info.id, mode="first")
     assert service.echo is not None
     assert await service.echo("hello") == "hello"
 
diff --git a/tests/test_services.py b/tests/test_services.py
index 4da6e060..a97ad361 100644
--- a/tests/test_services.py
+++ b/tests/test_services.py
@@ -21,7 +21,7 @@ async def test_singleton_service(fastapi_server):
     ) as api:
         await api.register_service(
             {
-                "id": "test-service",
+                "id": "test-service-1",
                 "name": "Test Service",
                 "config": {"singleton": True},
                 "description": "A test service",
@@ -36,7 +36,7 @@ async def test_singleton_service(fastapi_server):
         with pytest.raises(Exception, match=".*Failed to notify workspace manager.*"):
             await api.register_service(
                 {
-                    "id": "test-service",
+                    "id": "test-service-1",
                     "name": "Test Service",
                     "config": {"singleton": True},
                     "description": "A test service",
diff --git a/tests/test_vectors.py b/tests/test_vectors.py
index a1c1fee0..8cfd4ccf 100644
--- a/tests/test_vectors.py
+++ b/tests/test_vectors.py
@@ -1,166 +1,457 @@
 """Test Vector Search."""
+import asyncio
 import pytest
 import numpy as np
 import random
 from hypha_rpc import connect_to_server
+from redis import asyncio as aioredis
+import pytest_asyncio
+from aiobotocore.session import get_session
 
+from hypha.vectors import VectorSearchEngine
 
-from . import SERVER_URL_REDIS_1
+from . import SERVER_URL_REDIS_1, find_item, MINIO_ROOT_PASSWORD, MINIO_ROOT_USER
 
 # All test coroutines will be treated as marked.
 pytestmark = pytest.mark.asyncio
 
 
-async def test_artifact_vector_collection(
-    minio_server, fastapi_server_redis_1, test_user_token
-):
-    """Test vector-related functions within a vector-collection artifact."""
+@pytest_asyncio.fixture(name="vector_search_engine")
+async def vector_search_engine(redis_server):
+    """Fixture to set up VectorSearchEngine instance."""
+    redis = aioredis.from_url(redis_server)
+    yield VectorSearchEngine(redis=redis)
 
-    # Connect to the server and set up the artifact manager
-    api = await connect_to_server(
+
+async def test_create_and_list_collection(vector_search_engine):
+    """Test creating a collection and listing collections."""
+    # Create a collection
+    collection_name = "test_collection"
+    vector_fields = [
+        {"type": "TAG", "name": "id"},
         {
-            "name": "test deploy client",
-            "server_url": SERVER_URL_REDIS_1,
-            "token": test_user_token,
-        }
+            "type": "VECTOR",
+            "name": "vector",
+            "algorithm": "FLAT",
+            "attributes": {"TYPE": "FLOAT32", "DIM": 128, "DISTANCE_METRIC": "COSINE"},
+        },
+    ]
+
+    await vector_search_engine.create_collection(
+        collection_name, vector_fields, overwrite=True
+    )
+
+    # Verify the collection exists
+    collections = await vector_search_engine.list_collections()
+    assert find_item(collections, "name", collection_name)
+
+
+@pytest.mark.asyncio
+async def test_add_and_search_vectors(vector_search_engine):
+    """Test adding and searching vectors in a collection."""
+    # Setup
+    collection_name = "vector_test"
+    vector_fields = [
+        {"type": "TAG", "name": "id"},
+        {
+            "type": "VECTOR",
+            "name": "vector",
+            "algorithm": "FLAT",
+            "attributes": {"TYPE": "FLOAT32", "DIM": 128, "DISTANCE_METRIC": "COSINE"},
+        },
+    ]
+    await vector_search_engine.create_collection(
+        collection_name, vector_fields, overwrite=True
+    )
+
+    # Add vectors
+    vectors = [
+        {"id": "vec1", "vector": np.random.rand(128).tolist()},
+        {"id": "vec2", "vector": np.random.rand(128).tolist()},
+    ]
+    await vector_search_engine.add_vectors(collection_name, vectors)
+
+    # Search for a vector
+    query_vector = np.random.rand(128).tolist()
+    results = await vector_search_engine.search_vectors(
+        collection_name,
+        query={"vector": query_vector},
+        limit=1,
     )
-    artifact_manager = await api.get_service("public/artifact-manager")
-
-    # Create a vector-collection artifact
-    vector_collection_manifest = {
-        "name": "vector-collection",
-        "description": "A test vector collection",
-    }
-    vector_collection_config = {
-        "vector_fields": [
+
+    assert len(results) == 1
+    assert "id" in results[0]
+
+
+@pytest.mark.asyncio
+async def test_dump_and_load_collection(vector_search_engine, minio_server):
+    """Test dumping and loading a collection to/from S3."""
+    # Setup
+    async with get_session().create_client(
+        "s3",
+        endpoint_url=minio_server,
+        aws_access_key_id=MINIO_ROOT_USER,
+        aws_secret_access_key=MINIO_ROOT_PASSWORD,
+        region_name="us-east-1",
+    ) as s3_client:
+        bucket = "test-bucket"
+        prefix = "test-prefix"
+
+        # Create a bucket if it doesn't exist
+        try:
+            await s3_client.create_bucket(Bucket=bucket)
+        except s3_client.meta.client.exceptions.BucketAlreadyOwnedByYou:
+            pass
+
+        collection_name = "s3_test_collection"
+        vector_fields = [
+            {"type": "TAG", "name": "id"},
             {
                 "type": "VECTOR",
                 "name": "vector",
                 "algorithm": "FLAT",
                 "attributes": {
                     "TYPE": "FLOAT32",
-                    "DIM": 384,
+                    "DIM": 128,
                     "DISTANCE_METRIC": "COSINE",
                 },
             },
-            {"type": "TEXT", "name": "text"},
-            {"type": "TAG", "name": "label"},
-            {"type": "NUMERIC", "name": "rand_number"},
-        ],
-        "embedding_models": {
-            "vector": "fastembed:BAAI/bge-small-en-v1.5",
+        ]
+        await vector_search_engine.create_collection(
+            collection_name, vector_fields, overwrite=True
+        )
+
+        # Add vectors
+        vectors = [
+            {"id": "vec1", "vector": np.random.rand(128).tolist()},
+            {"id": "vec2", "vector": np.random.rand(128).tolist()},
+        ]
+        await vector_search_engine.add_vectors(collection_name, vectors)
+
+        # Dump collection
+        await vector_search_engine.dump_collection(
+            collection_name, s3_client, bucket, prefix
+        )
+
+        # Verify collection was dumped
+        objects = await s3_client.list_objects(Bucket=bucket, Prefix=prefix)
+        assert find_item(objects["Contents"], "Key", f"{prefix}/vec1.json")
+        assert find_item(objects["Contents"], "Key", f"{prefix}/vec2.json")
+
+        # Load collection
+        await vector_search_engine.load_collection(
+            collection_name, s3_client, bucket, prefix, overwrite=True
+        )
+
+        # Verify vectors and index were loaded correctly
+        collections = await vector_search_engine.list_collections()
+        assert find_item(collections, "name", collection_name)
+
+
+@pytest.mark.asyncio
+async def test_delete_collection(vector_search_engine):
+    """Test deleting a collection."""
+    collection_name = "delete_test"
+    vector_fields = [
+        {"type": "TAG", "name": "id"},
+        {
+            "type": "VECTOR",
+            "name": "vector",
+            "algorithm": "FLAT",
+            "attributes": {"TYPE": "FLOAT32", "DIM": 128, "DISTANCE_METRIC": "COSINE"},
         },
-    }
-    vector_collection = await artifact_manager.create(
-        type="vector-collection",
-        manifest=vector_collection_manifest,
-        config=vector_collection_config,
+    ]
+    await vector_search_engine.create_collection(
+        collection_name, vector_fields, overwrite=True
     )
-    # Add vectors to the collection
+
+    # Add vectors
     vectors = [
+        {"id": "vec1", "vector": np.random.rand(128).tolist()},
+        {"id": "vec2", "vector": np.random.rand(128).tolist()},
+    ]
+    await vector_search_engine.add_vectors(collection_name, vectors)
+
+    # Delete collection
+    await vector_search_engine.delete_collection(collection_name)
+
+    # Verify deletion
+    collections = await vector_search_engine.list_collections()
+    assert not find_item(collections, "name", collection_name)
+
+
+async def test_artifact_vector_collection(
+    minio_server, fastapi_server_redis_1, test_user_token
+):
+    """Test vector-related functions within a vector-collection artifact."""
+
+    # Connect to the server and set up the artifact manager
+    async with connect_to_server(
         {
-            "vector": [random.random() for _ in range(384)],
-            "text": "This is a test document.",
-            "label": "doc1",
-            "rand_number": random.randint(0, 10),
-        },
-        {
-            "vector": np.random.rand(384),
-            "text": "Another document.",
-            "label": "doc2",
-            "rand_number": random.randint(0, 10),
-        },
+            "name": "test deploy client",
+            "server_url": SERVER_URL_REDIS_1,
+            "token": test_user_token,
+        }
+    ) as api:
+        await api.create_workspace(
+            {
+                "name": "my-vector-test-workspace",
+                "description": "This is a test workspace",
+                "persistent": True,
+            },
+            overwrite=True,
+        )
+
+    async with connect_to_server(
         {
-            "vector": np.random.rand(384),
-            "text": "Yet another document.",
-            "label": "doc3",
-            "rand_number": random.randint(0, 10),
-        },
-    ]
-    await artifact_manager.add_vectors(
-        artifact_id=vector_collection.id,
-        vectors=vectors,
-    )
+            "name": "test deploy client",
+            "server_url": SERVER_URL_REDIS_1,
+            "token": test_user_token,
+            "workspace": "my-vector-test-workspace",
+        }
+    ) as api:
+        artifact_manager = await api.get_service("public/artifact-manager")
 
-    vc = await artifact_manager.read(artifact_id=vector_collection.id)
-    assert vc["config"]["vector_count"] == 3
+        # Create a vector-collection artifact
+        vector_collection_manifest = {
+            "name": "vector-collection",
+            "description": "A test vector collection",
+        }
+        vector_collection_config = {
+            "vector_fields": [
+                {
+                    "type": "VECTOR",
+                    "name": "vector",
+                    "algorithm": "FLAT",
+                    "attributes": {
+                        "TYPE": "FLOAT32",
+                        "DIM": 384,
+                        "DISTANCE_METRIC": "COSINE",
+                    },
+                },
+                {"type": "TEXT", "name": "text"},
+                {"type": "TAG", "name": "label"},
+                {"type": "NUMERIC", "name": "rand_number"},
+            ],
+            "embedding_models": {
+                "vector": "fastembed:BAAI/bge-small-en-v1.5",
+            },
+        }
+        vector_collection = await artifact_manager.create(
+            type="vector-collection",
+            manifest=vector_collection_manifest,
+            config=vector_collection_config,
+        )
+        # Add vectors to the collection
+        vectors = [
+            {
+                "vector": [random.random() for _ in range(384)],
+                "text": "This is a test document.",
+                "label": "doc1",
+                "rand_number": random.randint(0, 10),
+            },
+            {
+                "vector": np.random.rand(384),
+                "text": "Another document.",
+                "label": "doc2",
+                "rand_number": random.randint(0, 10),
+            },
+            {
+                "vector": np.random.rand(384),
+                "text": "Yet another document.",
+                "label": "doc3",
+                "rand_number": random.randint(0, 10),
+            },
+        ]
+        await artifact_manager.add_vectors(
+            artifact_id=vector_collection.id,
+            vectors=vectors,
+        )
 
-    # Search for vectors by query vector
-    query_vector = [random.random() for _ in range(384)]
-    search_results = await artifact_manager.search_vectors(
-        artifact_id=vector_collection.id,
-        query={"vector": query_vector},
-        limit=2,
-    )
-    assert len(search_results) <= 2
+        vc = await artifact_manager.read(artifact_id=vector_collection.id)
+        assert vc["config"]["vector_count"] == 3
 
-    results = await artifact_manager.search_vectors(
-        artifact_id=vector_collection.id,
-        query={"vector": query_vector},
-        limit=3,
-        pagination=True,
-    )
-    assert results["total"] == 3
+        # Search for vectors by query vector
+        query_vector = [random.random() for _ in range(384)]
+        search_results = await artifact_manager.search_vectors(
+            artifact_id=vector_collection.id,
+            query={"vector": query_vector},
+            limit=2,
+        )
+        assert len(search_results) <= 2
 
-    search_results = await artifact_manager.search_vectors(
-        artifact_id=vector_collection.id,
-        filters={"rand_number": [-2, -1]},
-        query={"vector": np.random.rand(384)},
-        limit=2,
-    )
-    assert len(search_results) == 0
+        results = await artifact_manager.search_vectors(
+            artifact_id=vector_collection.id,
+            query={"vector": query_vector},
+            limit=3,
+            pagination=True,
+        )
+        assert results["total"] == 3
 
-    search_results = await artifact_manager.search_vectors(
-        artifact_id=vector_collection.id,
-        filters={"rand_number": [0, 10]},
-        query={"vector": np.random.rand(384)},
-        limit=2,
-    )
-    assert len(search_results) > 0
+        search_results = await artifact_manager.search_vectors(
+            artifact_id=vector_collection.id,
+            filters={"rand_number": [-2, -1]},
+            query={"vector": np.random.rand(384)},
+            limit=2,
+        )
+        assert len(search_results) == 0
 
-    # Search for vectors by text
-    vectors = [
-        {"vector": "This is a test document.", "label": "doc1"},
-        {"vector": "Another test document.", "label": "doc2"},
-    ]
-    await artifact_manager.add_vectors(
-        artifact_id=vector_collection.id,
-        vectors=vectors,
-    )
+        search_results = await artifact_manager.search_vectors(
+            artifact_id=vector_collection.id,
+            filters={"rand_number": [0, 10]},
+            query={"vector": np.random.rand(384)},
+            limit=2,
+        )
+        assert len(search_results) > 0
 
-    text_search_results = await artifact_manager.search_vectors(
-        artifact_id=vector_collection.id,
-        query={"vector": "test document"},
-        limit=2,
-    )
-    assert len(text_search_results) <= 2
+        # Search for vectors by text
+        vectors = [
+            {"vector": "This is a test document.", "label": "doc1"},
+            {"vector": "Another test document.", "label": "doc2"},
+        ]
+        await artifact_manager.add_vectors(
+            artifact_id=vector_collection.id,
+            vectors=vectors,
+        )
 
-    # Retrieve a specific vector
-    retrieved_vector = await artifact_manager.get_vector(
-        artifact_id=vector_collection.id,
-        id=text_search_results[0]["id"],
-    )
-    assert "label" in retrieved_vector
+        text_search_results = await artifact_manager.search_vectors(
+            artifact_id=vector_collection.id,
+            query={"vector": "test document"},
+            limit=2,
+        )
+        assert len(text_search_results) <= 2
 
-    # List vectors in the collection
-    vector_list = await artifact_manager.list_vectors(
-        artifact_id=vector_collection.id,
-        offset=0,
-        limit=10,
-    )
-    assert len(vector_list) > 0
+        # Retrieve a specific vector
+        retrieved_vector = await artifact_manager.get_vector(
+            artifact_id=vector_collection.id,
+            id=text_search_results[0]["id"],
+        )
+        assert "label" in retrieved_vector
+
+        # List vectors in the collection
+        vector_list = await artifact_manager.list_vectors(
+            artifact_id=vector_collection.id,
+            offset=0,
+            limit=10,
+        )
+        assert len(vector_list) > 0
+
+        # Remove a vector from the collection
+        await artifact_manager.remove_vectors(
+            artifact_id=vector_collection.id,
+            ids=[vector_list[0]["id"]],
+        )
+        remaining_vectors = await artifact_manager.list_vectors(
+            artifact_id=vector_collection.id,
+            offset=0,
+            limit=10,
+        )
+        assert all(v["id"] != vector_list[0]["id"] for v in remaining_vectors)
+
+        # Clean up by deleting the vector collection
+        await artifact_manager.delete(artifact_id=vector_collection.id)
 
-    # Remove a vector from the collection
-    await artifact_manager.remove_vectors(
-        artifact_id=vector_collection.id,
-        ids=[vector_list[0]["id"]],
-    )
-    remaining_vectors = await artifact_manager.list_vectors(
-        artifact_id=vector_collection.id,
-        offset=0,
-        limit=10,
-    )
-    assert all(v["id"] != vector_list[0]["id"] for v in remaining_vectors)
 
-    # Clean up by deleting the vector collection
-    await artifact_manager.delete(artifact_id=vector_collection.id)
+async def test_load_dump_vector_collections(
+    minio_server, fastapi_server_redis_1, test_user_token, root_user_token
+):
+    """Test loading and dumping vector collections."""
+    # Connect to the server and set up the artifact manager
+    async with connect_to_server(
+        {
+            "name": "test deploy client",
+            "server_url": SERVER_URL_REDIS_1,
+            "token": test_user_token,
+        }
+    ) as api:
+        await api.create_workspace(
+            {
+                "name": "my-vector-dump-workspace",
+                "description": "This is a test workspace for dumping vector collections",
+                "persistent": True,
+            },
+            overwrite=True,
+        )
+    async with connect_to_server(
+        {
+            "name": "test deploy client",
+            "server_url": SERVER_URL_REDIS_1,
+            "token": test_user_token,
+            "workspace": "my-vector-dump-workspace",
+        }
+    ) as api:
+        artifact_manager = await api.get_service("public/artifact-manager")
+
+        # Create a vector-collection artifact
+        vector_collection_manifest = {
+            "name": "vector-collection",
+            "description": "A test vector collection",
+        }
+        vector_collection_config = {
+            "vector_fields": [
+                {
+                    "type": "VECTOR",
+                    "name": "vector",
+                    "algorithm": "FLAT",
+                    "attributes": {
+                        "TYPE": "FLOAT32",
+                        "DIM": 384,
+                        "DISTANCE_METRIC": "COSINE",
+                    },
+                },
+                {"type": "TEXT", "name": "text"},
+                {"type": "TAG", "name": "label"},
+                {"type": "NUMERIC", "name": "rand_number"},
+            ],
+            "embedding_models": {
+                "vector": "fastembed:BAAI/bge-small-en-v1.5",
+            },
+        }
+        vector_collection = await artifact_manager.create(
+            type="vector-collection",
+            manifest=vector_collection_manifest,
+            config=vector_collection_config,
+        )
+
+        # Add vectors to the collection
+        vectors = [
+            {
+                "vector": [random.random() for _ in range(384)],
+                "text": "This is a test document.",
+                "label": "doc1",
+                "rand_number": random.randint(0, 10),
+            },
+            {
+                "vector": np.random.rand(384),
+                "text": "Another document.",
+                "label": "doc2",
+                "rand_number": random.randint(0, 10),
+            },
+            {
+                "vector": np.random.rand(384),
+                "text": "Yet another document.",
+                "label": "doc3",
+                "rand_number": random.randint(0, 10),
+            },
+        ]
+        await artifact_manager.add_vectors(
+            artifact_id=vector_collection.id,
+            vectors=vectors,
+        )
+    await asyncio.sleep(10)
+    async with connect_to_server(
+        {
+            "name": "test deploy client",
+            "server_url": SERVER_URL_REDIS_1,
+            "token": test_user_token,
+            "workspace": "my-vector-dump-workspace",
+        }
+    ) as api:
+        await api.wait_until_ready(timeout=60)
+        artifact_manager = await api.get_service("public/artifact-manager")
+        vc = await artifact_manager.read(artifact_id=vector_collection.id)
+        assert vc["config"]["vector_count"] == 3
+        assert (
+            vc["config"]["embedding_models"]["vector"]
+            == "fastembed:BAAI/bge-small-en-v1.5"
+        )
diff --git a/tests/test_workspace.py b/tests/test_workspace.py
index 15aa5515..7276f7b0 100644
--- a/tests/test_workspace.py
+++ b/tests/test_workspace.py
@@ -131,7 +131,7 @@ def test(context=None):
     await api.disconnect()
 
 
-async def test_create_workspace_token(test_user_token):
+async def test_create_workspace_token(fastapi_server, test_user_token):
     server_url = WS_SERVER_URL
     # Connect to the Hypha server
     user = await connect_to_server(
@@ -162,3 +162,15 @@ async def test_create_workspace_token(test_user_token):
         }
     )
     print(f"Token generated: {token}")
+
+
+async def test_workspace_ready(fastapi_server):
+    """Test workspace."""
+    server_url = WS_SERVER_URL
+    server = await connect_to_server(
+        {
+            "server_url": server_url,
+        }
+    )
+    result = await server.wait_until_ready(timeout=1)
+    assert result and result.ready is True and not result.errors