diff --git a/core/frontend/src/components/kraken/KrakenManager.ts b/core/frontend/src/components/kraken/KrakenManager.ts
index 451c4b6d54..7b0f93f106 100644
--- a/core/frontend/src/components/kraken/KrakenManager.ts
+++ b/core/frontend/src/components/kraken/KrakenManager.ts
@@ -1,3 +1,5 @@
+import { QueryTarget, Sample, Subscriber } from '@eclipse-zenoh/zenoh-ts'
+
 import zenoh from '@/libs/zenoh'
 import {
   ExtensionData,
@@ -9,24 +11,14 @@ import {
   UploadProgressEvent,
 } from '@/types/kraken'
 import back_axios from '@/utils/api'
-import { QueryTarget, Sample, Subscriber } from '@eclipse-zenoh/zenoh-ts'
+import { createDeferred } from '@/utils/deferred'
 
 const KRAKEN_BASE_URL = '/kraken'
 const KRAKEN_API_V2_URL = `${KRAKEN_BASE_URL}/v2.0`
+const KRAKEN_BASE_ZENOH = 'kraken'
+const INSTALL_PROGRESS_TOPIC = `${KRAKEN_BASE_ZENOH}/extension/install/progress`
+const ZENOH_QUERY_STANDARD_TIMEOUT = 10000
 
-/**
- * List details of all installed extensions.
- * @returns {Promise<InstalledExtensionData[]>}
- */
-export async function fetchInstalledExtensions(): Promise<InstalledExtensionData[]> {
-  const response = await back_axios({
-    method: 'get',
-    url: `${KRAKEN_API_V2_URL}/extension/`,
-    timeout: 10000,
-  })
-
-  return response.data as InstalledExtensionData[]
-}
 
 /**
  * List all manifest sources from kraken, uses API v2
@@ -188,78 +180,161 @@ export async function setManifestSourceOrder(identifier: string, order: number):
   })
 }
 
+function buildInstallQueryKey(identifier: string, tag: string | undefined, stable: boolean): string {
+  let key = `${KRAKEN_BASE_ZENOH}/extension/install?identifier=${encodeURIComponent(identifier)}`
+  if (tag) key += `;tag=${encodeURIComponent(tag)}`
+  if (!stable) key += ';stable=false'
+  return key
+}
+
+type InstallSample =
+  | { kind: 'error'; message: string }
+  | { kind: 'complete' }
+  | { kind: 'progress'; raw: string }
+  | null
+
+function parseInstallSample(raw: string, identifier: string): InstallSample {
+  let data: { identifier?: string; status?: string; error?: string }
+  try {
+    data = JSON.parse(raw)
+  } catch {
+    return null
+  }
+  if (data.identifier !== identifier) return null
+  if (data.error) return { kind: 'error', message: data.error }
+  if (data.status === 'complete') return { kind: 'complete' }
+  return { kind: 'progress', raw }
+}
+
 /**
- * Install an extension to the latest version available
- * @param {InstalledExtensionData} extension The extension to be installed
+ * Install an extension to the latest version available. 
+ * The backend publishes the pull progress on `INSTALL_PROGRESS_TOPIC`.
+ *
+ * @param {string} identifier The identifier of the extension
  * @param {function} progressHandler The progress handler for the download
+ * @param {string} tag The tag of the extension
+ * @param {boolean} stable If true, will install the latest stable version, default is true
+ * @param {number} timeout The timeout for the install
  */
 export async function installExtension(
-  extension: InstalledExtensionData,
-  progressHandler: (event: any) => void,
+  identifier: string,
+  progressHandler?: (fragment: string) => void,
+  tag?: string,
+  stable = true,
+  timeout = 600000,
 ): Promise<void> {
-  await back_axios({
-    url: `${KRAKEN_API_V2_URL}/extension/install`,
-    method: 'POST',
-    data: {
-      identifier: extension.identifier,
-      name: extension.name,
-      docker: extension.docker,
-      tag: extension.tag,
-      enabled: true,
-      permissions: extension?.permissions ?? '',
-      user_permissions: extension?.user_permissions ?? '',
-    },
-    timeout: 600000,
-    onDownloadProgress: progressHandler,
-  })
+  const deferred = createDeferred<void>()
+  let subscriber: Subscriber | null = null
+  let timer: ReturnType<typeof setTimeout> | null = null
+
+  async function cleanup(): Promise<void> {
+    if (timer !== null) {
+      clearTimeout(timer)
+      timer = null
+    }
+    try {
+      await subscriber?.undeclare()
+    } catch {
+      // The subscriber may already be gone. Ignore cleanup errors.
+    }
+    subscriber = null
+  }
+
+  async function handleSample(sample: Sample): Promise<void> {
+    const result = parseInstallSample(sample.payload().to_string(), identifier)
+    if (result === null) return
+    switch (result.kind) {
+      case 'error':
+        cleanup().finally(() => deferred.reject(new Error(result.message)))
+        break
+      case 'complete':
+        cleanup().finally(() => deferred.resolve())
+        break
+      case 'progress':
+        progressHandler?.(result.raw)
+        break
+      default:
+        break
+    }
+  }
+
+  // Subscribe before triggering the install.
+  subscriber = await zenoh.subscriber(INSTALL_PROGRESS_TOPIC, handleSample)
+  if (!subscriber) {
+    throw new Error('Failed to subscribe to install progress topic')
+  }
+  timer = setTimeout(
+    () => cleanup().finally(() => deferred.reject(new Error(`Install timed out after ${timeout}ms`))),
+    timeout,
+  )
+
+  try {
+    const reply = await zenoh.query(
+      buildInstallQueryKey(identifier, tag, stable),
+      QueryTarget.BestMatching,
+      timeout,
+    )
+    if (!reply || reply.error) {
+      throw new Error(reply?.error ?? 'Install query failed')
+    }
+  } catch (error) {
+    await cleanup()
+    throw error
+  }
+
+  return deferred.promise
 }
 
 /**
- * Enable an extension by its identifier and tag, uses API v2
+ * Enable an extension by its identifier and tag, uses zenoh
  * @param {string} identifier The identifier of the extension
  * @param {string} tag The tag of the extension
  */
 export async function enableExtension(identifier: string, tag: string): Promise<void> {
-  await back_axios({
-    method: 'POST',
-    url: `${KRAKEN_API_V2_URL}/extension/${identifier}/${tag}/enable`,
-    timeout: 10000,
-  })
+  await zenoh.query(
+    `${KRAKEN_BASE_ZENOH}/extension/enable?identifier=${encodeURIComponent(identifier)};tag=${encodeURIComponent(tag)}`,
+    QueryTarget.BestMatching,
+    ZENOH_QUERY_STANDARD_TIMEOUT,
+  )
 }
 
 /**
- * Disable an extension by its identifier, uses API v2
+ * Disable an extension by its identifier, uses zenoh
  * @param {string} identifier The identifier of the extension
  */
 export async function disableExtension(identifier: string): Promise<void> {
-  await back_axios({
-    method: 'POST',
-    url: `${KRAKEN_API_V2_URL}/extension/${identifier}/disable`,
-    timeout: 10000,
-  })
+  await zenoh.query(
+    `${KRAKEN_BASE_ZENOH}/extension/disable?identifier=${encodeURIComponent(identifier)}`,
+    QueryTarget.BestMatching,
+    ZENOH_QUERY_STANDARD_TIMEOUT,
+  )
 }
 
 /**
- * Uninstall an extension by its identifier, uses API v2
+ * Uninstall an extension by its identifier, uses zenoh
  * @param {string} identifier The identifier of the extension
  */
-export async function uninstallExtension(identifier: string): Promise<void> {
-  await back_axios({
-    method: 'DELETE',
-    url: `${KRAKEN_API_V2_URL}/extension/${identifier}`,
-  })
+export async function uninstallExtension(identifier: string, tag?: string): Promise<void> {
+  let queryKey = `${KRAKEN_BASE_ZENOH}/extension/uninstall?identifier=${encodeURIComponent(identifier)}`
+  if (tag) queryKey += `;tag=${encodeURIComponent(tag)}`
+
+  await zenoh.query(
+    queryKey,
+    QueryTarget.BestMatching,
+    ZENOH_QUERY_STANDARD_TIMEOUT,
+  )
 }
 
 /**
- * Restart an extension by its identifier, uses API v2
+ * Restart an extension by its identifier, uses zenoh
  * @param {string} identifier The identifier of the extension
  */
 export async function restartExtension(identifier: string): Promise<void> {
-  await back_axios({
-    method: 'POST',
-    url: `${KRAKEN_API_V2_URL}/extension/${identifier}/restart`,
-    timeout: 10000,
-  })
+  await zenoh.query(
+    `${KRAKEN_BASE_ZENOH}/extension/restart?identifier=${encodeURIComponent(identifier)}`,
+    QueryTarget.BestMatching,
+    ZENOH_QUERY_STANDARD_TIMEOUT,
+  )
 }
 
 /**
@@ -282,42 +357,37 @@ export async function updateExtensionToVersion(
 }
 
 /**
- * List all installed extensions from kraken, uses API v2
+ * List details of all installed extensions.
+ * @returns {Promise<InstalledExtensionData[]> | null}
  */
-export async function getInstalledExtensions(): Promise<InstalledExtensionData[]> {
-  const response = await back_axios({
-    method: 'GET',
-    url: `${KRAKEN_API_V2_URL}/extension/`,
-    timeout: 30000,
-  })
-
-  return response.data as InstalledExtensionData[]
+export async function fetchInstalledExtensions(): Promise<InstalledExtensionData[] | null> {
+  return zenoh.query(
+    `${KRAKEN_BASE_ZENOH}/extension/fetch`, 
+    QueryTarget.BestMatching,
+    30000,
+  )
 }
 
 /**
- * List all running containers from kraken, uses API v2
+ * List all running containers from kraken, uses zenoh.
  */
-export async function listContainers(): Promise<RunningContainer[]> {
-  const response = await back_axios({
-    method: 'GET',
-    url: `${KRAKEN_API_V2_URL}/container/`,
-    timeout: 30000,
-  })
-
-  return response.data as RunningContainer[]
+export async function listContainers(): Promise<RunningContainer[] | null> {
+  return zenoh.query(
+    `${KRAKEN_BASE_ZENOH}/container/fetch`,
+    QueryTarget.BestMatching,
+    ZENOH_QUERY_STANDARD_TIMEOUT,
+  )
 }
 
 /**
- * List all stats of all running containers from kraken, uses API v2
+ * List all stats of all running containers from kraken, uses zenoh.
  */
-export async function getContainersStats(): Promise<any> {
-  const response = await back_axios({
-    method: 'GET',
-    url: `${KRAKEN_API_V2_URL}/container/stats`,
-    timeout: 20000,
-  })
-
-  return response.data
+export async function getContainersStats(): Promise<any | null> {
+  return zenoh.query(
+    `${KRAKEN_BASE_ZENOH}/container/stats`,
+    QueryTarget.BestMatching,
+    ZENOH_QUERY_STANDARD_TIMEOUT,
+  )
 }
 
 /**
@@ -352,11 +422,11 @@ export async function uploadExtensionTarFile(
  * @returns {Promise<void>}
  */
 export async function keepTemporaryExtensionAlive(tempTag: string): Promise<void> {
-  await back_axios({
-    method: 'POST',
-    url: `${KRAKEN_API_V2_URL}/extension/upload/keep-alive?temp_tag=${tempTag}`,
-    timeout: 10000,
-  })
+  await zenoh.query(
+    `${KRAKEN_BASE_ZENOH}/extension/upload/keep-alive?temp_tag=${encodeURIComponent(tempTag)}`,
+    QueryTarget.BestMatching,
+    ZENOH_QUERY_STANDARD_TIMEOUT,
+  )
 }
 
 /**
@@ -395,7 +465,7 @@ export async function finalizeExtension(
  * @returns {Promise<any | null>}
  */
 export async function getHistoricalLogsForExtension(identifier: string, timeout: number): Promise<any | null> {
-  const queryKey = `kraken/extension/logs/request?extension_name=${identifier}`
+  const queryKey = `${KRAKEN_BASE_ZENOH}/container/logs/request?extension_name=${identifier}`
   return await zenoh.query(queryKey, QueryTarget.BestMatching, timeout)
 }
 
@@ -424,9 +494,7 @@ export default {
   disabledManifestSource,
   setManifestSourcesOrders,
   setManifestSourceOrder,
-  updateExtensionToVersion,
   installExtension,
-  getInstalledExtensions,
   enableExtension,
   disableExtension,
   uninstallExtension,
diff --git a/core/frontend/src/utils/deferred.ts b/core/frontend/src/utils/deferred.ts
new file mode 100644
index 0000000000..8fde32480c
--- /dev/null
+++ b/core/frontend/src/utils/deferred.ts
@@ -0,0 +1,15 @@
+interface Deferred<T> {
+  promise: Promise<T>
+  resolve: (value: T) => void
+  reject: (reason: Error) => void
+}
+  
+export function createDeferred<T>(): Deferred<T> {
+  let resolve!: (value: T) => void
+  let reject!: (reason: Error) => void
+  const promise = new Promise<T>((res, rej) => {
+    resolve = res
+    reject = rej
+  })
+  return { promise, resolve, reject }
+}
diff --git a/core/frontend/src/views/ExtensionManagerView.vue b/core/frontend/src/views/ExtensionManagerView.vue
index 91bcc2674f..3b7ea53fc0 100644
--- a/core/frontend/src/views/ExtensionManagerView.vue
+++ b/core/frontend/src/views/ExtensionManagerView.vue
@@ -378,7 +378,6 @@
 </template>
 
 <script lang="ts">
-import { Session } from '@eclipse-zenoh/zenoh-ts'
 import Vue from 'vue'
 
 import NotSafeOverlay from '@/components/common/NotSafeOverlay.vue'
@@ -394,7 +393,6 @@ import ExtensionSettingsModal from '@/components/kraken/modals/ExtensionSettings
 import PullProgress from '@/components/utils/PullProgress.vue'
 import Notifier from '@/libs/notifier'
 import settings from '@/libs/settings'
-import zenoh from '@/libs/zenoh'
 import { OneMoreTime } from '@/one-more-time'
 import { Dictionary } from '@/types/common'
 import { kraken_service } from '@/types/frontend_services'
@@ -509,7 +507,6 @@ export default Vue.extend({
       install_from_file_last_level: -1,
       active_operation_identifier: localStorage.getItem(ACTIVE_OPERATION_KEY) as null | string,
       active_operation_type: (localStorage.getItem(ACTIVE_OPERATION_TYPE_KEY) ?? null) as null | 'install' | 'update',
-      session: null as Session | null,
     }
   },
   computed: {
@@ -612,29 +609,30 @@ export default Vue.extend({
       }
     },
   },
-  async mounted() {
+  mounted() {
     this.fetchManifest()
     this.fetch_installed_ext_task.setAction(this.fetchInstalledExtensions)
     this.fetch_running_containers_task.setAction(this.fetchRunningContainers)
     this.fetch_containers_stats_task.setAction(this.fetchContainersStats)
-    await this.initializeZenohSession()
   },
   destroyed() {
     clearInterval(this.metrics_interval)
     this.stopUploadKeepAlive()
-    this.session = null
   },
   methods: {
-    async initializeZenohSession() {
-      if (this.session) {
-        return
-      }
-
-      try {
-        this.session = await zenoh.getSession()
-      } catch (error) {
-        console.error('[ExtensionManagerView] Failed to open Zenoh session:', error)
+    applyInstalledExtensions(extensions: InstalledExtensionData[]): void {
+      const map: Dictionary<InstalledExtensionData> = {}
+      for (const extension of extensions) {
+        if (!extension?.identifier) continue
+        map[extension.identifier] = extension
       }
+      this.installed_extensions = map
+      this.dockers_fetch_failed = false
+      this.dockers_fetch_done = true
+    },
+    applyRunningContainers(containers: RunningContainer[]): void {
+      this.running_containers = containers
+      this.clearStaleInstallingState()
     },
     clearEditedExtension() {
       this.edited_extension = null
@@ -795,12 +793,8 @@ export default Vue.extend({
     async update(extension: InstalledExtensionData, version: string) {
       this.setInstallingState(extension.identifier, 'update')
       this.show_pull_output = true
-      const tracker = this.getTracker()
-      kraken.updateExtensionToVersion(
-        extension.identifier,
-        version,
-        (progressEvent) => this.handleDownloadProgress(progressEvent.event, tracker),
-      )
+
+      this.installWithTracker(extension.identifier, version)
         .then(() => {
           this.fetchInstalledExtensions()
           notifier.pushSuccess('EXTENSION_UPDATE_SUCCESS', `${extension.name} updated successfully.`, true)
@@ -863,8 +857,11 @@ export default Vue.extend({
     },
     async fetchRunningContainers(): Promise<void> {
       try {
-        this.running_containers = await kraken.listContainers()
-        this.clearStaleInstallingState()
+        const containers = await kraken.listContainers()
+        if (!Array.isArray(containers)) {
+          throw new Error('Unexpected response while fetching running containers')
+        }
+        this.applyRunningContainers(containers)
       } catch (error) {
         notifier.pushBackError('RUNNING_CONTAINERS_FETCH_FAIL', error)
       }
@@ -883,13 +880,15 @@ export default Vue.extend({
       }
     },
     async fetchContainersStats(): Promise<void> {
-      kraken.getContainersStats()
-        .then((response) => {
-          this.metrics = response
-        })
-        .catch((error) => {
-          notifier.pushBackError('EXTENSIONS_METRICS_FETCH_FAIL', error)
-        })
+      try {
+        const stats = await kraken.getContainersStats()
+        if (!stats || typeof stats !== 'object' || stats.error) {
+          throw new Error(stats?.error ?? 'Unexpected response while fetching container stats')
+        }
+        this.metrics = stats
+      } catch (error) {
+        notifier.pushBackError('EXTENSIONS_METRICS_FETCH_FAIL', error)
+      }
     },
     getContainerName(extension: InstalledExtensionData): string | null {
       return this.getContainer(extension)?.name ?? null
@@ -904,21 +903,18 @@ export default Vue.extend({
       }
     },
     async fetchInstalledExtensions(): Promise<void> {
-      kraken.getInstalledExtensions()
-        .then((response) => {
-          this.installed_extensions = {}
-          for (const extension of response) {
-            this.installed_extensions[extension.identifier] = extension
-          }
-          this.dockers_fetch_failed = false
-        })
-        .catch((error) => {
-          notifier.pushBackError('EXTENSIONS_INSTALLED_FETCH_FAIL', error)
-          this.dockers_fetch_failed = true
-        })
-        .finally(() => {
-          this.dockers_fetch_done = true
-        })
+      try {
+        const extensions = await kraken.fetchInstalledExtensions()
+        if (!Array.isArray(extensions)) {
+          throw new Error('Unexpected response while fetching installed extensions')
+        }
+        this.applyInstalledExtensions(extensions)
+      } catch (error) {
+        this.dockers_fetch_failed = true
+        notifier.pushBackError('EXTENSIONS_INSTALLED_FETCH_FAIL', error)
+      } finally {
+        this.dockers_fetch_done = true
+      }
     },
     showLogs(extension: InstalledExtensionData) {
       this.selected_log_extension_identifier = extension.identifier
@@ -933,12 +929,8 @@ export default Vue.extend({
       this.setInstallingState(extension.identifier, 'install')
       this.show_dialog = false
       this.show_pull_output = true
-      const tracker = this.getTracker()
 
-      kraken.installExtension(
-        extension,
-        (progressEvent) => this.handleDownloadProgress(progressEvent.event, tracker),
-      )
+      this.installWithTracker(extension.identifier, extension.tag)
         .then(() => {
           this.fetchInstalledExtensions()
           notifier.pushSuccess('EXTENSION_INSTALL_SUCCESS', `${extension.name} installed successfully.`, true)
@@ -1059,6 +1051,16 @@ export default Vue.extend({
       temp[extension.identifier].loading = loading
       this.installed_extensions = temp
     },
+    installWithTracker(identifier: string, tag?: string): Promise<void> {
+      const tracker = this.getTracker()
+      return kraken.installExtension(identifier, (fragment: string) => {
+        tracker.digestStreamFragment(fragment)
+        this.pull_output = tracker.pull_output
+        this.download_percentage = tracker.download_percentage
+        this.extraction_percentage = tracker.extraction_percentage
+        this.status_text = tracker.overall_status
+      }, tag)
+    },
     getTracker(): PullTracker {
       return new PullTracker(
         () => {
diff --git a/core/libs/commonwealth/src/commonwealth/utils/zenoh_helper.py b/core/libs/commonwealth/src/commonwealth/utils/zenoh_helper.py
index 71914cf469..b3bb9a247a 100644
--- a/core/libs/commonwealth/src/commonwealth/utils/zenoh_helper.py
+++ b/core/libs/commonwealth/src/commonwealth/utils/zenoh_helper.py
@@ -1,8 +1,9 @@
 import asyncio
 import json
 import re
-from concurrent.futures import ThreadPoolExecutor
-from typing import Any, Callable
+import threading
+from concurrent.futures import Future
+from typing import Any, AsyncGenerator, Callable, Coroutine, Optional
 
 import fastapi
 import zenoh
@@ -12,12 +13,17 @@
 from .Singleton import Singleton
 
 PARAM_REGEX = r"{[a-zA-Z0-9_]+}"
+_LOOP_START_TIMEOUT_S = 5.0
+_LOOP_JOIN_TIMEOUT_S = 2.0
 
 
 class ZenohSession(metaclass=Singleton):
     session: zenoh.Session | None = None
     config: zenoh.Config
-    _executor: ThreadPoolExecutor | None = None
+    _loop: asyncio.AbstractEventLoop | None = None
+    _loop_thread: threading.Thread | None = None
+    _loop_ready: threading.Event
+    _loop_start_error: Exception | None = None
 
     def __init__(self, service_name: str) -> None:
         if self.session is not None:
@@ -26,27 +32,83 @@ def __init__(self, service_name: str) -> None:
         self.zenoh_config(service_name)
         self.session = zenoh.open(self.config)
 
-        self._executor = ThreadPoolExecutor(
-            max_workers=4,
-            thread_name_prefix="zenoh-",
+        self._loop_ready = threading.Event()
+        self._loop_start_error = None
+        self._loop_thread = threading.Thread(
+            target=self._run_loop,
+            name="zenoh-loop",
+            daemon=True,
         )
-
-    def submit_to_executor(self, func: Callable[..., Any]) -> None:
-        if self._executor is None:
-            logger.warning("Zenoh session executor is not available, task will not be initialized.")
-            return
+        self._loop_thread.start()
+        if not self._loop_ready.wait(timeout=_LOOP_START_TIMEOUT_S):
+            raise RuntimeError(f"Zenoh event loop did not signal readiness within {_LOOP_START_TIMEOUT_S}s")
+        if self._loop_start_error is not None:
+            raise RuntimeError(
+                f"Zenoh event loop failed to start: {self._loop_start_error}"
+            ) from self._loop_start_error
+
+    def _run_loop(self) -> None:
         try:
-            self._executor.submit(func)
+            loop = asyncio.new_event_loop()
+            asyncio.set_event_loop(loop)
+            self._loop = loop
         except Exception as e:
-            logger.error(f"Error submitting task to zenoh session executor: {e}")
+            self._loop_start_error = e
+            self._loop_ready.set()
+            logger.exception("Failed to initialize Zenoh event loop")
+            return
+        self._loop_ready.set()
+        try:
+            loop.run_forever()
+        finally:
+            try:
+                pending = asyncio.all_tasks(loop)
+                for task in pending:
+                    task.cancel()
+                if pending:
+                    loop.run_until_complete(asyncio.gather(*pending, return_exceptions=True))
+                loop.run_until_complete(loop.shutdown_asyncgens())
+                loop.run_until_complete(loop.shutdown_default_executor())
+            finally:
+                loop.close()
+
+    def submit_coroutine(self, coroutine: Coroutine[Any, Any, Any]) -> Future[Any] | None:
+
+        loop = self._loop
+        if loop is None or not loop.is_running():
+            logger.warning("Zenoh session loop is not available, task will not be scheduled.")
+            coroutine.close()
+            return None
+        try:
+            future = asyncio.run_coroutine_threadsafe(coroutine, loop)
+        except RuntimeError as e:
+            logger.warning(f"Could not schedule coroutine on Zenoh loop: {e}")
+            coroutine.close()
+            return None
+
+        def _log_if_failed(fut: Future[Any]) -> None:
+            if fut.cancelled():
+                return
+            exc = fut.exception()
+            if exc is not None:
+                logger.opt(exception=exc).error("Unhandled error in Zenoh background task")
+
+        future.add_done_callback(_log_if_failed)
+        return future
 
     def close(self) -> None:
         if self.session:
             self.session.close()  # type: ignore[no-untyped-call]
             self.session = None
-        if self._executor:
-            self._executor.shutdown(wait=False, cancel_futures=True)
-            self._executor = None
+        loop = self._loop
+        if loop is not None and loop.is_running():
+            loop.call_soon_threadsafe(loop.stop)
+        if self._loop_thread is not None:
+            self._loop_thread.join(timeout=_LOOP_JOIN_TIMEOUT_S)
+            if self._loop_thread.is_alive():
+                logger.warning(f"Zenoh loop thread did not terminate within {_LOOP_JOIN_TIMEOUT_S}")
+            self._loop_thread = None
+        self._loop = None
 
     def zenoh_config(self, service_name: str) -> None:
         configuration = {
@@ -66,40 +128,90 @@ def zenoh_config(self, service_name: str) -> None:
 class ZenohRouter:
     prefix: str
     zenoh_session: ZenohSession
+    _publishers: dict[str, zenoh.Publisher]
 
     def __init__(self, service_name: str):
         self.prefix = service_name
         self.zenoh_session = ZenohSession(service_name)
+        self._publishers = {}
 
-    def add_queryable(self, path: str, func: Callable[..., Any]) -> None:
-        full_path = self.prefix
-        if path:
-            full_path += f"/{path}"
+    def close(self) -> None:
+        for publisher in self._publishers.values():
+            try:
+                publisher.undeclare()  # type: ignore[no-untyped-call]
+            except Exception:
+                logger.exception(f"Failed to undeclare publisher {publisher.key_expr}.")
+        self._publishers.clear()
+
+    def ensure_publisher(self, path: str) -> None:
+        full_path = self.get_path(path)
+
+        if self.zenoh_session.session is None:
+            logger.warning(f"Zenoh session unavailable, cannot declare publisher for {full_path}")
+            return
 
-        def wrapper(query: zenoh.Query) -> None:
-            params = dict(query.parameters)  # type: ignore
+        if full_path not in self._publishers:
+            self._publishers[full_path] = self.zenoh_session.session.declare_publisher(full_path)
 
-            async def _handle_async() -> None:
-                try:
-                    response = await func(**params)
-                    if response is not None:
-                        query.reply(query.selector.key_expr, json.dumps(response, default=str))
-                except Exception as e:
-                    logger.exception(f"Error in zenoh query handler: {query.selector.key_expr}")
-                    error_response = {
-                        "error": str(e),
-                        "error_type": type(e).__name__,
-                    }
-                    query.reply(query.selector.key_expr, json.dumps(error_response))
+    def get_publisher(self, path: str) -> zenoh.Publisher | None:
+        full_path = self.get_path(path)
+        return self._publishers.get(full_path)
 
-            def run_async() -> None:
-                asyncio.run(_handle_async())
+    def add_queryable(self, path: str, func: Callable[..., Any]) -> None:
+        full_path = self.get_path(path)
 
-            self.zenoh_session.submit_to_executor(run_async)
+        def wrapper(query: zenoh.Query) -> None:
+            params = dict(query.parameters)  # type: ignore
+            key_expr = query.selector.key_expr
+
+            async def _handle_async(q: zenoh.Query) -> None:
+                with q:
+                    try:
+                        response = await func(**params)
+                        if response is not None:
+                            q.reply(key_expr, json.dumps(response, default=str))
+                    except Exception as e:
+                        logger.exception(f"Error in zenoh query handler: {key_expr}")
+                        error_response = {
+                            "error": str(e),
+                            "error_type": type(e).__name__,
+                        }
+                        try:
+                            q.reply(key_expr, json.dumps(error_response))
+                        except Exception:
+                            logger.exception(f"Failed to send error reply for {key_expr}")
+
+            if self.zenoh_session.submit_coroutine(_handle_async(query)) is None:
+                query.drop()  # type: ignore[no-untyped-call]
 
         if self.zenoh_session.session:
             self.zenoh_session.session.declare_queryable(full_path, wrapper)
 
+    def publish_from_generator(
+        self,
+        topic: str,
+        generator: AsyncGenerator[str, None],
+        on_complete: Optional[str] = None,
+    ) -> None:
+        async def _run() -> None:
+            publisher = self.get_publisher(topic)
+            if publisher is None:
+                logger.warning(f"Publisher for {topic} not found, dropping generator")
+                async for _ in generator:
+                    pass
+                return
+
+            try:
+                async for chunk in generator:
+                    publisher.put(chunk)
+            except Exception:
+                logger.exception(f"Error while publishing from generator on {topic}")
+            finally:
+                if on_complete is not None:
+                    publisher.put(on_complete)
+
+        self.zenoh_session.submit_coroutine(_run())
+
     def add_routes_to_zenoh(self, app: fastapi.FastAPI) -> None:
         queryables = []
         for route in app.router.routes:
@@ -115,6 +227,12 @@ def add_routes_to_zenoh(self, app: fastapi.FastAPI) -> None:
         for path, func in queryables:
             self.add_queryable(path, func)
 
+    def get_path(self, path: str) -> str:
+        full_path = self.prefix
+        if path:
+            full_path += f"/{path}"
+        return full_path
+
 
 def clean_path(path: str) -> str:
     path = path.removeprefix("/").removesuffix("/")
diff --git a/core/services/kraken/zenoh_handlers/app.py b/core/services/kraken/zenoh_handlers/app.py
index 67c55d351d..f15230a3d7 100644
--- a/core/services/kraken/zenoh_handlers/app.py
+++ b/core/services/kraken/zenoh_handlers/app.py
@@ -1,5 +1,6 @@
 from commonwealth.utils.zenoh_helper import ZenohRouter, ZenohSession
 from config import SERVICE_NAME
+from zenoh_handlers.container_handler import ContainerHandlers
 from zenoh_handlers.extension_handler import ExtensionHandlers
 
 session = ZenohSession(SERVICE_NAME)
@@ -8,3 +9,7 @@
 # Extension
 extension_handlers = ExtensionHandlers(router)
 extension_handlers.register_queryables()
+
+# Container
+container_handlers = ContainerHandlers(router)
+container_handlers.register_queryables()
diff --git a/core/services/kraken/zenoh_handlers/container_handler.py b/core/services/kraken/zenoh_handlers/container_handler.py
new file mode 100644
index 0000000000..ac4100b191
--- /dev/null
+++ b/core/services/kraken/zenoh_handlers/container_handler.py
@@ -0,0 +1,72 @@
+from typing import Any, List
+
+from commonwealth.utils.zenoh_helper import ZenohRouter
+from extension_logs import ExtensionLogPublisher
+from harbor import ContainerManager
+from loguru import logger
+from settings import get_extension_settings
+
+
+class ContainerHandlers:
+    def __init__(self, router: ZenohRouter) -> None:
+        self.router = router
+
+    async def logs_request_handler(self, extension_name: str) -> dict[str, Any]:
+        if not extension_name:
+            return {"error": "extension_name parameter is required"}
+
+        try:
+            extensions = get_extension_settings()
+            extension = next(
+                (ext for ext in extensions if extension_name in (ext.identifier, ext.name)),
+                None,
+            )
+
+            if not extension:
+                return {"error": f"Extension {extension_name} not found"}
+
+            if not extension.enabled:
+                return {"error": f"Extension {extension_name} is not enabled"}
+
+            topic = ExtensionLogPublisher._topic_for(extension)
+
+            container_name = extension.container_name()
+
+            raw_logs = await ContainerManager.get_container_historical_logs(container_name)
+            formatted_messages = []
+            for raw_line in raw_logs:
+                level, _ = ExtensionLogPublisher._extract_level(raw_line)
+                formatted_messages.append(
+                    {
+                        "level": level,
+                        "message": raw_line,
+                    }
+                )
+            return {
+                "status": "success",
+                "messages": formatted_messages,
+                "total_lines": len(formatted_messages),
+                "topic": topic,
+            }
+        except Exception as e:
+            logger.exception(f"Error handling logs request for {extension_name}")
+            return {"error": str(e), "error_type": type(e).__name__}
+
+    async def list_containers_handler(self) -> List[dict[str, Any]]:
+        """
+        List all running containers.
+        """
+        containers = await ContainerManager.get_running_containers()
+        return [container.dict() for container in containers]
+
+    async def container_stats_handler(self) -> dict[str, dict[str, Any]]:
+        """
+        List stats of all running containers.
+        """
+        stats = await ContainerManager.get_containers_stats()
+        return {name: usage.dict() for name, usage in stats.items()}
+
+    def register_queryables(self) -> None:
+        self.router.add_queryable("container/logs/request", self.logs_request_handler)
+        self.router.add_queryable("container/fetch", self.list_containers_handler)
+        self.router.add_queryable("container/stats", self.container_stats_handler)
diff --git a/core/services/kraken/zenoh_handlers/extension_handler.py b/core/services/kraken/zenoh_handlers/extension_handler.py
index 26a513582b..050c18bbed 100644
--- a/core/services/kraken/zenoh_handlers/extension_handler.py
+++ b/core/services/kraken/zenoh_handlers/extension_handler.py
@@ -1,53 +1,101 @@
-from typing import Any
+import asyncio
+import json
+from typing import Any, AsyncGenerator, List, cast
 
 from commonwealth.utils.zenoh_helper import ZenohRouter
-from extension_logs import ExtensionLogPublisher
-from harbor import ContainerManager
+from extension.extension import Extension
 from loguru import logger
-from settings import get_extension_settings
 
 
 class ExtensionHandlers:
+    INSTALL_PROGRESS_TOPIC = "extension/install/progress"
+
     def __init__(self, router: ZenohRouter) -> None:
         self.router = router
+        self.router.ensure_publisher(self.INSTALL_PROGRESS_TOPIC)
 
-    async def logs_request_handler(self, extension_name: str) -> dict[str, Any]:
-        if not extension_name:
-            return {"error": "extension_name parameter is required"}
-
+    @staticmethod
+    async def _install_progress_stream(identifier: str, extension: Extension) -> AsyncGenerator[str, None]:
         try:
-            extensions = get_extension_settings()
-            extension = next((ext for ext in extensions if extension_name in (ext.identifier, ext.name)), None)
-
-            if not extension:
-                return {"error": f"Extension {extension_name} not found"}
-
-            if not extension.enabled:
-                return {"error": f"Extension {extension_name} is not enabled"}
-
-            topic = ExtensionLogPublisher._topic_for(extension)
-
-            container_name = extension.container_name()
-
-            raw_logs = await ContainerManager.get_container_historical_logs(container_name)
-            formatted_messages = []
-            for raw_line in raw_logs:
-                level, _ = ExtensionLogPublisher._extract_level(raw_line)
-                formatted_messages.append(
-                    {
-                        "level": level,
-                        "message": raw_line,
-                    }
-                )
-            return {
-                "status": "success",
-                "messages": formatted_messages,
-                "total_lines": len(formatted_messages),
-                "topic": topic,
-            }
-        except Exception as e:
-            logger.exception(f"Error handling logs request for {extension_name}")
-            return {"error": str(e), "error_type": type(e).__name__}
+            async for chunk in extension.install():
+                try:
+                    payload = json.loads(chunk)
+                except (TypeError, ValueError) as e:
+                    logger.debug(f"Failed to parse install progress chunk: {e}")
+                    continue
+                payload["identifier"] = identifier
+                yield json.dumps(payload, default=str)
+        except Exception as error:
+            logger.exception(f"Install of {identifier} failed")
+            yield json.dumps({"identifier": identifier, "error": str(error)}, default=str)
+
+    async def install_handler(self, identifier: str, tag: str = "", stable: str = "true") -> dict[str, str]:
+        """
+        Install an extension by its identifier and tag, if tag is not provided it will install the latest stable version.
+        """
+        if tag:
+            extension = cast(Extension, await Extension.from_manifest(identifier, tag))
+        else:
+            extension = await Extension.from_latest(identifier, stable.lower() == "true")
+
+        on_complete = json.dumps({"identifier": identifier, "status": "complete"})
+        self.router.publish_from_generator(
+            self.INSTALL_PROGRESS_TOPIC,
+            self._install_progress_stream(identifier, extension),
+            on_complete=on_complete,
+        )
+        return {"status": "started", "identifier": identifier}
+
+    async def uninstall_handler(self, identifier: str, tag: str = "") -> None:
+        """
+        Uninstall all versions of an extension by its identifier or just a specific version if a tag is provided.
+        """
+        if tag:
+            extension = cast(Extension, await Extension.from_settings(identifier, tag))
+            await extension.uninstall()
+        else:
+            extensions = cast(List[Extension], await Extension.from_settings(identifier))
+            await asyncio.gather(*[ext.uninstall() for ext in extensions])
+
+    async def enable_handler(self, identifier: str, tag: str) -> None:
+        """
+        Enables an extension by its identifier and tag.
+        """
+        extension = cast(Extension, await Extension.from_settings(identifier, tag))
+        await extension.enable()
+
+    async def disable_handler(self, identifier: str) -> None:
+        """
+        Disables current running extension by its identifier.
+        """
+        extension = await Extension.from_running(identifier)
+        await extension.disable()
+
+    async def restart_handler(self, identifier: str) -> None:
+        """
+        Restart current running extension by its identifier.
+        """
+        extension = await Extension.from_running(identifier)
+        await extension.restart()
+
+    async def fetch_handler(self) -> list[dict[str, Any]]:
+        """
+        List details of all installed extensions.
+        """
+        extensions = cast(List[Extension], await Extension.from_settings())
+        return [ext.source.model_dump() for ext in extensions if ext.source.identifier != ""]
+
+    async def keep_uploaded_extension_alive_handler(self, temp_tag: str) -> None:
+        """
+        Refresh the keep-alive timestamp for a temporary extension while the user is editing metadata.
+        """
+        Extension.keep_temporary_extension_alive(temp_tag)
 
     def register_queryables(self) -> None:
-        self.router.add_queryable("extension/logs/request", self.logs_request_handler)
+        self.router.add_queryable("extension/fetch", self.fetch_handler)
+        self.router.add_queryable("extension/install", self.install_handler)
+        self.router.add_queryable("extension/uninstall", self.uninstall_handler)
+        self.router.add_queryable("extension/enable", self.enable_handler)
+        self.router.add_queryable("extension/disable", self.disable_handler)
+        self.router.add_queryable("extension/restart", self.restart_handler)
+        self.router.add_queryable("extension/upload/keep-alive", self.keep_uploaded_extension_alive_handler)