diff --git a/agentscope-service/service-dataplane/src/main/java/io/agentscope/builder/web/api/SelfHostedWorkerController.java b/agentscope-service/service-dataplane/src/main/java/io/agentscope/builder/web/api/SelfHostedWorkerController.java
index ec26a4205b..c1cb012b09 100644
--- a/agentscope-service/service-dataplane/src/main/java/io/agentscope/builder/web/api/SelfHostedWorkerController.java
+++ b/agentscope-service/service-dataplane/src/main/java/io/agentscope/builder/web/api/SelfHostedWorkerController.java
@@ -39,6 +39,7 @@
 import org.springframework.web.bind.annotation.RestController;
 import org.springframework.web.server.ResponseStatusException;
 import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Schedulers;
 
 /**
  * Outbound-worker data plane for {@code self_hosted}: pending tool_use listing, tool_result
@@ -74,10 +75,11 @@ public Mono<List<Map<String, Object>>> pendingTools(
             @PathVariable("sessionId") String sessionId,
             Authentication auth) {
         return Mono.fromCallable(
-                () -> {
-                    requireEnvironmentWorker(auth, environmentId, sessionId);
-                    return pendingHandsToolService.listPending(sessionId);
-                });
+                        () -> {
+                            requireEnvironmentWorker(auth, environmentId, sessionId);
+                            return pendingHandsToolService.listPending(sessionId);
+                        })
+                .subscribeOn(Schedulers.boundedElastic());
     }
 
     /** Posts one or more tool results and resumes the suspended turn. */
@@ -88,27 +90,33 @@ public Mono<List<SessionEventDto>> toolResults(
             @RequestBody ToolResultsRequest body,
             Authentication auth) {
         return Mono.fromCallable(
-                () -> {
-                    ManagedSessionDto session =
-                            requireEnvironmentWorker(auth, environmentId, sessionId);
-                    if (body == null || body.results() == null || body.results().isEmpty()) {
-                        throw ApiException.invalidRequest(
-                                "missing_results", "results is required", "results");
-                    }
-                    List<ToolResultBlock> blocks = new ArrayList<>();
-                    List<SessionEventDto> recorded = new ArrayList<>();
-                    for (Map<String, Object> payload : body.results()) {
-                        ToolResultBlock block = SessionTurnRunner.toolResultFromPayload(payload);
-                        blocks.add(block);
-                        Map<String, Object> stored = new LinkedHashMap<>(payload);
-                        stored.putIfAbsent("tool_use_id", block.getId());
-                        recorded.add(
-                                eventLog.append(
-                                        sessionId, SessionEventTypes.USER_TOOL_RESULT, stored));
-                    }
-                    turnRunner.resumeWithToolResults(session, blocks);
-                    return recorded;
-                });
+                        () -> {
+                            ManagedSessionDto session =
+                                    requireEnvironmentWorker(auth, environmentId, sessionId);
+                            if (body == null
+                                    || body.results() == null
+                                    || body.results().isEmpty()) {
+                                throw ApiException.invalidRequest(
+                                        "missing_results", "results is required", "results");
+                            }
+                            List<ToolResultBlock> blocks = new ArrayList<>();
+                            List<SessionEventDto> recorded = new ArrayList<>();
+                            for (Map<String, Object> payload : body.results()) {
+                                ToolResultBlock block =
+                                        SessionTurnRunner.toolResultFromPayload(payload);
+                                blocks.add(block);
+                                Map<String, Object> stored = new LinkedHashMap<>(payload);
+                                stored.putIfAbsent("tool_use_id", block.getId());
+                                recorded.add(
+                                        eventLog.append(
+                                                sessionId,
+                                                SessionEventTypes.USER_TOOL_RESULT,
+                                                stored));
+                            }
+                            turnRunner.resumeWithToolResults(session, blocks);
+                            return recorded;
+                        })
+                .subscribeOn(Schedulers.boundedElastic());
     }
 
     /** Downloads the session agent's skills bundle for local staging on the worker. */
@@ -118,10 +126,11 @@ public Mono<Map<String, Object>> skills(
             @PathVariable("sessionId") String sessionId,
             Authentication auth) {
         return Mono.fromCallable(
-                () -> {
-                    requireEnvironmentWorker(auth, environmentId, sessionId);
-                    return skillsBundleService.bundleForSession(sessionId);
-                });
+                        () -> {
+                            requireEnvironmentWorker(auth, environmentId, sessionId);
+                            return skillsBundleService.bundleForSession(sessionId);
+                        })
+                .subscribeOn(Schedulers.boundedElastic());
     }
 
     private ManagedSessionDto requireEnvironmentWorker(
diff --git a/agentscope-service/service-dataplane/src/main/java/io/agentscope/builder/web/api/WorkerEnvironmentController.java b/agentscope-service/service-dataplane/src/main/java/io/agentscope/builder/web/api/WorkerEnvironmentController.java
index dc37a0b9eb..76b45c67d4 100644
--- a/agentscope-service/service-dataplane/src/main/java/io/agentscope/builder/web/api/WorkerEnvironmentController.java
+++ b/agentscope-service/service-dataplane/src/main/java/io/agentscope/builder/web/api/WorkerEnvironmentController.java
@@ -37,6 +37,7 @@
 import org.springframework.web.bind.annotation.RestController;
 import org.springframework.web.server.ResponseStatusException;
 import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Schedulers;
 
 /**
  * REST surface for out-of-process Environment Workers on {@code self_hosted} environments.
@@ -76,19 +77,20 @@ public Mono<ResponseEntity<EnvironmentWorkQueue.WorkItem>> poll(
             @RequestParam("workerId") String workerId,
             @RequestParam(name = "timeoutMs", defaultValue = "25000") long timeoutMs) {
         return Mono.fromCallable(
-                () -> {
-                    try {
-                        Optional<EnvironmentWorkQueue.WorkItem> item =
-                                workQueue.poll(environmentId, workerId, timeoutMs);
-                        return item.map(this::withSessionMetadata)
-                                .map(ResponseEntity::ok)
-                                .orElseGet(() -> ResponseEntity.noContent().build());
-                    } catch (InterruptedException e) {
-                        Thread.currentThread().interrupt();
-                        throw new ResponseStatusException(
-                                HttpStatus.SERVICE_UNAVAILABLE, "Poll interrupted");
-                    }
-                });
+                        () -> {
+                            try {
+                                Optional<EnvironmentWorkQueue.WorkItem> item =
+                                        workQueue.poll(environmentId, workerId, timeoutMs);
+                                return item.map(this::withSessionMetadata)
+                                        .map(ResponseEntity::ok)
+                                        .orElseGet(() -> ResponseEntity.noContent().build());
+                            } catch (InterruptedException e) {
+                                Thread.currentThread().interrupt();
+                                throw new ResponseStatusException(
+                                        HttpStatus.SERVICE_UNAVAILABLE, "Poll interrupted");
+                            }
+                        })
+                .subscribeOn(Schedulers.boundedElastic());
     }
 
     private EnvironmentWorkQueue.WorkItem withSessionMetadata(EnvironmentWorkQueue.WorkItem item) {
@@ -109,7 +111,8 @@ public Mono<List<EnvironmentWorkQueue.WorkItem>> listWork(
             @RequestParam(value = "state", required = false) String state,
             Authentication auth) {
         requireUserAuth(auth);
-        return Mono.fromCallable(() -> workQueue.list(environmentId, state));
+        return Mono.fromCallable(() -> workQueue.list(environmentId, state))
+                .subscribeOn(Schedulers.boundedElastic());
     }
 
     /** Returns per-status counts and oldest queued age for the environment. */
@@ -117,7 +120,8 @@ public Mono<List<EnvironmentWorkQueue.WorkItem>> listWork(
     public Mono<CoordinationStore.WorkStats> workStats(
             @PathVariable("id") String environmentId, Authentication auth) {
         requireUserAuth(auth);
-        return Mono.fromCallable(() -> workQueue.stats(environmentId));
+        return Mono.fromCallable(() -> workQueue.stats(environmentId))
+                .subscribeOn(Schedulers.boundedElastic());
     }
 
     /** Returns a single work item by id. */
@@ -128,14 +132,15 @@ public Mono<EnvironmentWorkQueue.WorkItem> getWork(
             Authentication auth) {
         requireUserAuth(auth);
         return Mono.fromCallable(
-                () ->
-                        workQueue
-                                .get(workId)
-                                .orElseThrow(
-                                        () ->
-                                                new ResponseStatusException(
-                                                        HttpStatus.NOT_FOUND,
-                                                        "Unknown work item: " + workId)));
+                        () ->
+                                workQueue
+                                        .get(workId)
+                                        .orElseThrow(
+                                                () ->
+                                                        new ResponseStatusException(
+                                                                HttpStatus.NOT_FOUND,
+                                                                "Unknown work item: " + workId)))
+                .subscribeOn(Schedulers.boundedElastic());
     }
 
     /**