diff --git a/chronis/core/base/scheduler.py b/chronis/core/base/scheduler.py
index 9264e32..1aa708c 100644
--- a/chronis/core/base/scheduler.py
+++ b/chronis/core/base/scheduler.py
@@ -13,6 +13,7 @@
 )
 from chronis.core.execution.callbacks import OnFailureCallback, OnSuccessCallback
 from chronis.core.execution.coordinator import ExecutionCoordinator
+from chronis.core.execution.worker_pool import WorkerPool
 from chronis.core.jobs.definition import JobDefinition, JobInfo
 from chronis.core.schedulers.next_run_calculator import NextRunTimeCalculator
 from chronis.core.state import JobStatus
@@ -87,23 +88,19 @@ def __init__(
         self._init_execution_coordinator()
 
     def _init_execution_coordinator(self) -> None:
-        """Build a fresh thread pool and coordinator.
+        """Build a fresh worker pool and coordinator.
 
-        Called again on restart: a ThreadPoolExecutor cannot be reused after shutdown, so
+        Called again on restart: a pool that has been shut down stops accepting work, so
         start() must rebuild it. The registries are shared by reference, so registered
         functions and handlers survive.
         """
-        from concurrent.futures import ThreadPoolExecutor
-
-        self._executor = ThreadPoolExecutor(
-            max_workers=self.max_workers, thread_name_prefix="chronis-worker-"
-        )
+        self._workers = WorkerPool(max_workers=self.max_workers, logger=self.logger)
         self._executor_shutdown = False
 
         self._execution_coordinator = ExecutionCoordinator(
             storage=self.storage,
             lock=self.lock,
-            executor=self._executor,
+            workers=self._workers,
             function_registry=self._job_registry,
             failure_handler_registry=self._failure_handler_registry,
             success_handler_registry=self._success_handler_registry,
diff --git a/chronis/core/execution/coordinator.py b/chronis/core/execution/coordinator.py
index 486f67a..733db63 100644
--- a/chronis/core/execution/coordinator.py
+++ b/chronis/core/execution/coordinator.py
@@ -3,7 +3,6 @@
 import asyncio
 import threading
 from collections.abc import Callable, Generator
-from concurrent.futures import ThreadPoolExecutor
 from contextlib import contextmanager
 from dataclasses import dataclass
 from datetime import timedelta
@@ -20,6 +19,7 @@
 )
 from chronis.core.execution.callbacks import OnFailureCallback, OnSuccessCallback
 from chronis.core.execution.job_executor import JobExecutor
+from chronis.core.execution.worker_pool import Slot, WorkerPool
 from chronis.core.jobs.definition import JobInfo
 from chronis.core.schedulers.next_run_calculator import NextRunTimeCalculator
 from chronis.core.state import JobStatus
@@ -120,7 +120,7 @@ def __init__(
         self,
         storage: JobStorageAdapter,
         lock: LockAdapter | None,
-        executor: ThreadPoolExecutor,
+        workers: WorkerPool,
         function_registry: dict[str, Callable],
         failure_handler_registry: dict[str, OnFailureCallback],
         success_handler_registry: dict[str, OnSuccessCallback],
@@ -132,7 +132,7 @@ def __init__(
     ) -> None:
         self.storage = storage
         self.lock = lock
-        self.executor = executor
+        self.workers = workers
         self.function_registry = function_registry
         self.failure_handler_registry = failure_handler_registry
         self.success_handler_registry = success_handler_registry
@@ -149,7 +149,11 @@ def __init__(
 
     def try_execute(self, job_data: dict[str, Any], on_complete: Callable[[str], None]) -> bool:
         """
-        Try to execute a job with distributed lock.
+        Try to execute a job: reserve capacity, claim it, run it.
+
+        Capacity is reserved *before* the job is claimed. Claiming a job we have no room to
+        run would strand it in RUNNING until its liveness lease expired; leaving it
+        SCHEDULED lets the next poll — or a node with spare capacity — take it instead.
 
         Args:
             job_data: Job data from storage
@@ -164,27 +168,38 @@ def try_execute(self, job_data: dict[str, Any], on_complete: Callable[[str], Non
 
         job_logger = self.logger.with_context(job_id=job_id, job_name=job_name)
 
-        with self._acquire_lock_context(lock_key) as lock_acquired:
-            if not lock_acquired:
-                return False
+        slot = self.workers.reserve()
+        if slot is None:
+            return False
 
-            try:
-                claimed = self._try_claim_job_with_cas(job_id, job_data)
-
-                if claimed is None:
+        started = False
+        try:
+            with self._acquire_lock_context(lock_key) as lock_acquired:
+                if not lock_acquired:
                     return False
 
-                self._track_running(job_id)
                 try:
-                    self._trigger_execution(claimed, job_logger, on_complete)
-                except Exception:
-                    self._untrack_running(job_id)
-                    raise
-                return True
+                    claimed = self._try_claim_job_with_cas(job_id, job_data)
 
-            except Exception as e:
-                job_logger.error("Job execution failed", error=str(e), exc_info=True)
-                return False
+                    if claimed is None:
+                        return False
+
+                    self._track_running(job_id)
+                    try:
+                        self._trigger_execution(claimed, job_logger, on_complete, slot)
+                    except Exception:
+                        self._untrack_running(job_id)
+                        raise
+                    started = True
+                    return True
+
+                except Exception as e:
+                    job_logger.error("Job execution failed", error=str(e), exc_info=True)
+                    return False
+        finally:
+            # Every path that did not hand the slot to a running job gives it straight back.
+            if not started:
+                slot.release()
 
     def _track_running(self, job_id: str) -> None:
         """Mark a job as executing on this node so heartbeat() keeps it alive."""
@@ -248,43 +263,46 @@ def _trigger_execution(
         claimed: ClaimedJob,
         job_logger: ContextLogger,
         on_complete: Callable[[str], None],
+        slot: Slot,
     ) -> None:
         """
-        Trigger job execution with automatic rollback on failure.
+        Start the job on its slot, rolling the job back if it cannot be started at all.
 
-        Routes async functions to the shared event loop and sync functions
-        to the thread pool executor.
+        Async functions run on the shared event loop, sync functions on a worker thread.
+        Both hold the same kind of slot, so ``max_workers`` bounds them together.
 
         A sync job that times out cannot be killed — Python has no safe way to stop a
-        thread — so the timeout marks the job failed and abandons the thread. The job is
-        released from this node's bookkeeping at that moment rather than when the thread
-        eventually finishes: a thread that never finishes would otherwise hold its queue
-        slot forever and, because the queue rejects duplicates, block its own retry from
-        ever being picked up here again.
+        thread — so the timeout marks the job failed and abandons the thread. Everything the
+        job was holding is handed back at that moment rather than when the thread eventually
+        returns (it may never): its slot, its place in the queue, and its heartbeat. Waiting
+        for the thread instead would let a handful of hung jobs consume every slot and stop
+        the node executing anything at all.
         """
         job_id = claimed.job_id
         func_name = claimed.func_name
         timeout_seconds = claimed.timeout_seconds
 
-        def _on_done(_future: Any) -> None:
+        def _finish() -> None:
             self._untrack_running(job_id)
             on_complete(job_id)
 
+        timer: threading.Timer | None = None
+
         try:
             if self._job_executor.is_async(func_name):
                 loop = self._job_executor.ensure_async_loop()
                 future = asyncio.run_coroutine_threadsafe(
                     self._execute_async(claimed, job_logger), loop
                 )
-                future.add_done_callback(_on_done)
+
+                def _on_async_done(_future: Any) -> None:
+                    slot.release()
+                    _finish()
+
+                future.add_done_callback(_on_async_done)
             else:
                 gate = _OutcomeGate() if timeout_seconds else None
 
-                future = self.executor.submit(
-                    self._execute_in_background, claimed, job_logger, gate
-                )
-                future.add_done_callback(_on_done)
-
                 if timeout_seconds and gate is not None:
                     timeout_gate = gate
 
@@ -299,11 +317,11 @@ def _on_timeout() -> None:
                             timeout_seconds=timeout_seconds,
                         )
 
-                        # Hand the slot back now: the abandoned thread may never return, and
-                        # _on_done would never fire. Both calls are idempotent, so the late
-                        # _on_done (if it ever comes) is harmless.
-                        self._untrack_running(job_id)
-                        on_complete(job_id)
+                        # Give back the capacity now. The thread is written off; its slot,
+                        # queue place and heartbeat must not be. abandon() and _finish() are
+                        # idempotent, so the thread returning later is harmless.
+                        slot.abandon()
+                        _finish()
                         self._handle_job_failure(
                             claimed,
                             JobTimeoutError(f"Job '{job_id}' timed out after {timeout_seconds}s"),
@@ -314,24 +332,56 @@ def _on_timeout() -> None:
                     timer.daemon = True
                     timer.start()
 
-                    # Cancel timer when the future completes normally
-                    future.add_done_callback(lambda f: timer.cancel())
+                armed_timer = timer
+
+                def _run() -> None:
+                    try:
+                        self._execute_in_background(claimed, job_logger, gate)
+                    finally:
+                        if armed_timer is not None:
+                            armed_timer.cancel()
+                        _finish()
+
+                # Slot.run releases the slot when the thread returns — including a thread
+                # that was abandoned and came back later, where release() is a no-op.
+                slot.run(_run, name=f"chronis-job-{job_id}")
 
         except Exception as submit_error:
+            if timer is not None:
+                timer.cancel()  # the job never started; don't let its timeout fire
+
             job_logger.warning(
-                "Executor submit failed, rolling back to SCHEDULED",
+                "Could not start job, rolling back to SCHEDULED",
                 error=str(submit_error),
                 error_type=type(submit_error).__name__,
             )
-            try:
-                self._update_job_status(job_id, JobStatus.SCHEDULED)
-            except Exception as rollback_error:
-                job_logger.error(
-                    "Failed to rollback job status after submit failure",
-                    error=str(rollback_error),
-                )
+            self._rollback_unstarted_claim(claimed, job_logger)
             raise submit_error
 
+    def _rollback_unstarted_claim(self, claimed: ClaimedJob, job_logger: ContextLogger) -> None:
+        """Undo a claim for a job that never actually started.
+
+        The claim already moved next_run_time on to the *next* occurrence, so restoring only
+        the status would quietly drop the run we were about to perform: the job sits in
+        SCHEDULED with a future run time, invisible to both the poller and misfire detection,
+        and simply misses this firing. Put the scheduled time back so the job is still due
+        and the next poll picks it up.
+        """
+        updates: JobUpdateData = {
+            "status": JobStatus.SCHEDULED.value,
+            "updated_at": canonical_utc_now(),
+        }
+        if claimed.original_scheduled_time:
+            updates["next_run_time"] = claimed.original_scheduled_time
+
+        try:
+            self.storage.update_job(claimed.job_id, updates)
+        except Exception as rollback_error:
+            job_logger.error(
+                "Failed to rollback job after it could not be started",
+                error=str(rollback_error),
+            )
+
     def _execute_in_background(
         self,
         claimed: ClaimedJob,
diff --git a/chronis/core/execution/job_executor.py b/chronis/core/execution/job_executor.py
index 8194068..b43e708 100644
--- a/chronis/core/execution/job_executor.py
+++ b/chronis/core/execution/job_executor.py
@@ -4,6 +4,7 @@
 import inspect
 import threading
 from collections.abc import Callable
+from concurrent.futures import ThreadPoolExecutor
 from typing import Any
 
 from chronis.core.common.exceptions import FunctionNotRegisteredError
@@ -22,6 +23,7 @@ def __init__(
         self,
         function_registry: dict[str, Callable],
         logger: ContextLogger,
+        async_io_workers: int = 4,
     ) -> None:
         self.function_registry = function_registry
         self.logger = logger
@@ -29,6 +31,8 @@ def __init__(
         self._async_loop: asyncio.AbstractEventLoop | None = None
         self._async_thread: threading.Thread | None = None
         self._async_lock = threading.Lock()
+        self._async_io_workers = async_io_workers
+        self._async_io_executor: ThreadPoolExecutor | None = None
 
     def is_async(self, func_name: str) -> bool:
         """Check if a registered function is async."""
@@ -85,11 +89,25 @@ async def execute_async(self, job_data: dict[str, Any], job_logger: ContextLogge
             await coro
 
     def ensure_async_loop(self) -> asyncio.AbstractEventLoop:
-        """Get or create the shared event loop for async jobs (thread-safe)."""
+        """Get or create the shared event loop for async jobs (thread-safe).
+
+        The loop gets an executor we own and name. An async job's post-execution work
+        (storage writes, callbacks) is blocking, so it is pushed off the loop with
+        run_in_executor — which, left to itself, would silently spin up asyncio's *default*
+        executor: an unnamed pool, sized by a rule of thumb, that nothing in Chronis
+        configures or shuts down.
+        """
         if self._async_loop is None or self._async_loop.is_closed():
             with self._async_lock:
                 if self._async_loop is None or self._async_loop.is_closed():
                     self._async_loop = asyncio.new_event_loop()
+
+                    self._async_io_executor = ThreadPoolExecutor(
+                        max_workers=self._async_io_workers,
+                        thread_name_prefix="chronis-async-io-",
+                    )
+                    self._async_loop.set_default_executor(self._async_io_executor)
+
                     self._async_thread = threading.Thread(
                         target=self._async_loop.run_forever,
                         daemon=True,
@@ -139,4 +157,8 @@ async def _drain() -> None:
         self._async_loop = None
         self._async_thread = None
 
+        if self._async_io_executor is not None:
+            self._async_io_executor.shutdown(wait=wait)
+            self._async_io_executor = None
+
         return drained
diff --git a/chronis/core/execution/worker_pool.py b/chronis/core/execution/worker_pool.py
new file mode 100644
index 0000000..8cbd726
--- /dev/null
+++ b/chronis/core/execution/worker_pool.py
@@ -0,0 +1,178 @@
+"""Bounded execution capacity for jobs, counted in permits rather than threads."""
+
+import threading
+from collections.abc import Callable
+from typing import Any
+
+from chronis.utils.logging import ContextLogger
+
+
+class Slot:
+    """One unit of the pool's capacity, held by exactly one job execution.
+
+    Released when the job finishes — or abandoned, if the job timed out and its thread had
+    to be left behind. Both are idempotent: an abandoned thread that does eventually return
+    will try to release the slot a second time, and must not hand back capacity twice.
+    """
+
+    def __init__(self, pool: "WorkerPool") -> None:
+        self._pool = pool
+        self._lock = threading.Lock()
+        self._settled = False
+        self._thread: threading.Thread | None = None
+
+    def run(self, func: Callable[[], Any], *, name: str) -> None:
+        """Run func on a daemon thread, releasing this slot when it returns."""
+
+        def _body() -> None:
+            try:
+                func()
+            finally:
+                self.release()
+
+        thread = threading.Thread(target=_body, daemon=True, name=name)
+        self._thread = thread
+        self._pool._register(thread)
+        thread.start()
+
+    def release(self) -> None:
+        """Hand this unit of capacity back. Safe to call more than once."""
+        if self._settle():
+            self._pool._release()
+
+    def abandon(self) -> None:
+        """Hand the capacity back, but count the thread as leaked.
+
+        Used when a job times out: Python cannot kill the thread running it, so the thread
+        is left to finish on its own — possibly never. The *capacity* must come back anyway,
+        or enough timeouts would leave the scheduler unable to run anything at all.
+
+        The thread also stops being something shutdown waits for. Joining a thread we have
+        given up on would hang shutdown for exactly as long as the job hangs.
+        """
+        if self._settle():
+            if self._thread is not None:
+                self._pool._forget(self._thread)
+            self._pool._release(abandoned=True)
+
+    def _settle(self) -> bool:
+        """Claim the right to settle this slot. True for the first caller only."""
+        with self._lock:
+            if self._settled:
+                return False
+            self._settled = True
+            return True
+
+
+class WorkerPool:
+    """Caps how many jobs run at once — sync and async alike.
+
+    A ThreadPoolExecutor cannot do this job. A job killed by a timeout keeps its worker
+    thread forever (Python has no safe way to stop a thread) and the pool has no way to
+    replace that worker, so ``max_workers`` hung jobs permanently starve the node: it stops
+    executing anything, silently. And async jobs never touched the pool at all, so
+    ``max_workers`` never bounded them.
+
+    So capacity is counted in *permits*, not threads. A job takes one permit whether it runs
+    on a thread or on the event loop, and a timed-out job returns its permit immediately
+    even though its thread lives on. The stranded threads are counted and reported, so the
+    leak is visible instead of presenting as a mysterious stall.
+    """
+
+    def __init__(self, max_workers: int, logger: ContextLogger | None = None) -> None:
+        self.max_workers = max_workers
+        self.logger = logger
+
+        self._permits = threading.BoundedSemaphore(max_workers)
+        self._lock = threading.Lock()
+        self._in_use = 0
+        self._abandoned = 0
+        self._threads: list[threading.Thread] = []
+        self._shutdown = False
+
+    def reserve(self) -> Slot | None:
+        """Take one unit of capacity, or None if the pool is full.
+
+        A caller that gets None must not claim the job: leaving it SCHEDULED lets the next
+        poll — or another node with spare capacity — pick it up.
+        """
+        if self._shutdown:
+            return None
+
+        if not self._permits.acquire(blocking=False):
+            return None
+
+        with self._lock:
+            self._in_use += 1
+
+        return Slot(self)
+
+    def stats(self) -> dict[str, int]:
+        """Capacity accounting, including threads stranded by timeouts."""
+        with self._lock:
+            in_use = self._in_use
+            abandoned = self._abandoned
+
+        return {
+            "max_workers": self.max_workers,
+            "in_use": in_use,
+            "available": self.max_workers - in_use,
+            "abandoned_threads": abandoned,
+        }
+
+    def shutdown(self, wait: bool = True, timeout: float = 30.0) -> bool:
+        """Stop accepting work and wait for live jobs to finish.
+
+        Threads abandoned by a timeout are not waited on — that is the whole point of
+        abandoning them, and joining one would hang shutdown for as long as the job hangs.
+
+        Returns:
+            True if every live thread finished within the timeout.
+        """
+        self._shutdown = True
+
+        if not wait:
+            return False
+
+        with self._lock:
+            threads = list(self._threads)
+
+        deadline_all_done = True
+        for thread in threads:
+            thread.join(timeout=timeout)
+            if thread.is_alive():
+                deadline_all_done = False
+
+        return deadline_all_done
+
+    def _register(self, thread: threading.Thread) -> None:
+        with self._lock:
+            self._threads = [t for t in self._threads if t.is_alive()]
+            self._threads.append(thread)
+
+    def _forget(self, thread: threading.Thread) -> None:
+        """Stop waiting on a thread we have abandoned."""
+        with self._lock:
+            self._threads = [t for t in self._threads if t is not thread]
+
+    def _release(self, abandoned: bool = False) -> None:
+        with self._lock:
+            self._in_use -= 1
+            if abandoned:
+                self._abandoned += 1
+                abandoned_total = self._abandoned
+            else:
+                abandoned_total = 0
+
+        self._permits.release()
+
+        if abandoned and self.logger:
+            self.logger.warning(
+                "Abandoned a timed-out job's thread; its capacity was returned but the "
+                "thread cannot be killed and may run indefinitely",
+                abandoned_threads=abandoned_total,
+                max_workers=self.max_workers,
+            )
+
+
+__all__ = ["WorkerPool", "Slot"]
diff --git a/chronis/core/schedulers/fluent_builders.py b/chronis/core/schedulers/fluent_builders.py
index f4a34da..995b59a 100644
--- a/chronis/core/schedulers/fluent_builders.py
+++ b/chronis/core/schedulers/fluent_builders.py
@@ -225,7 +225,13 @@ def config(
             timezone: Timezone for scheduling
             retry: Maximum retry attempts
             retry_delay: Delay between retries in seconds
-            timeout: Execution timeout in seconds
+            timeout: Execution timeout in seconds. What a timeout *does* depends on the
+                function: an async job is genuinely cancelled, while a sync job cannot be
+                — Python has no safe way to stop a thread — so it is marked failed and its
+                thread is left running. Either way the job raises JobTimeoutError and takes
+                the normal retry / on_failure path. Abandoned threads are counted by
+                scheduler.get_worker_status()['abandoned_threads']; a number that keeps
+                climbing means jobs are hanging well past their timeout.
             priority: Job priority (0-10, higher = more urgent)
             metadata: Custom metadata dict
             on_success: Success callback
diff --git a/chronis/core/schedulers/polling_scheduler.py b/chronis/core/schedulers/polling_scheduler.py
index 5277770..d91e0c9 100644
--- a/chronis/core/schedulers/polling_scheduler.py
+++ b/chronis/core/schedulers/polling_scheduler.py
@@ -236,9 +236,9 @@ def stop(self) -> dict[str, Any]:
 
         Returns:
             Dictionary with shutdown status:
-            - sync_jobs_completed: Whether the worker pool drained. A job abandoned by a
-              timeout may still be running — its thread cannot be killed — so this reports
-              that the pool shut down, not that every thread is gone.
+            - sync_jobs_completed: Whether every live worker thread finished. Threads
+              abandoned by a timeout are not waited on — they cannot be killed and may run
+              forever — so this says nothing about them. get_worker_status() counts them.
             - async_jobs_completed: Whether every in-flight async job finished. False if the
               60s drain timed out and jobs were abandoned.
             - was_running: Whether scheduler was running before stop
@@ -256,17 +256,32 @@ def stop(self) -> dict[str, Any]:
 
         self._runner.shutdown(wait=True)
         async_drained = self._execution_coordinator.shutdown_async(wait=True)
-        self._executor.shutdown(wait=True, cancel_futures=False)
+        sync_drained = self._workers.shutdown(wait=True)
         self._executor_shutdown = True
 
         self._running = False
 
         return {
-            "sync_jobs_completed": True,
+            "sync_jobs_completed": sync_drained,
             "async_jobs_completed": async_drained,
             "was_running": True,
         }
 
+    def get_worker_status(self) -> dict[str, int]:
+        """
+        Execution capacity, for monitoring.
+
+        Returns:
+            - max_workers: The configured cap on jobs running at once (sync and async)
+            - in_use: Jobs currently holding capacity
+            - available: Capacity left
+            - abandoned_threads: Threads left behind by timed-out sync jobs. Their capacity
+              was returned, but the threads themselves cannot be killed and may still be
+              running. A number that keeps climbing means jobs are hanging past their
+              timeout — the leak is in the job, not the scheduler.
+        """
+        return self._workers.stats()
+
     def get_queue_status(self) -> dict[str, Any]:
         """
         Get job queue status for monitoring.
@@ -821,7 +836,13 @@ def config(
             timezone: Timezone for scheduling
             retry: Maximum retry attempts
             retry_delay: Delay between retries in seconds
-            timeout: Execution timeout in seconds
+            timeout: Execution timeout in seconds. What a timeout *does* depends on the
+                function: an async job is genuinely cancelled, while a sync job cannot be
+                — Python has no safe way to stop a thread — so it is marked failed and its
+                thread is left running. Either way the job raises JobTimeoutError and takes
+                the normal retry / on_failure path. Abandoned threads are counted by
+                scheduler.get_worker_status()['abandoned_threads']; a number that keeps
+                climbing means jobs are hanging well past their timeout.
             priority: Job priority (0-10, higher = more urgent)
             metadata: Custom metadata dict
 
diff --git a/docs/ARCHITECTURE_NOTES.md b/docs/ARCHITECTURE_NOTES.md
index ba6ebe0..1e2afab 100644
--- a/docs/ARCHITECTURE_NOTES.md
+++ b/docs/ARCHITECTURE_NOTES.md
@@ -33,30 +33,87 @@ to quietly cause duplicate execution.
 This makes `updated_at` core's field. Adapters must store it verbatim and never stamp their
 own clock on it — see `chronis.testing`, which checks this.
 
+## How execution capacity is counted
+
+**`max_workers` caps jobs, not threads** — sync and async alike. Capacity is counted in
+*permits* (`WorkerPool`), and a job takes one whether it runs on a worker thread or on the
+event loop. A job with no permit available is simply not claimed; it stays SCHEDULED for the
+next poll, or for a node with room.
+
+The reason capacity is not just "a thread in a pool" is the timeout. A sync job that times
+out cannot be killed — Python has no safe way to stop a thread — so it is abandoned. Held on
+a `ThreadPoolExecutor`, that abandoned thread would occupy a worker forever and the pool has
+no way to replace it: `max_workers` hung jobs and the node silently stops executing anything
+at all. Separating the permit from the thread means the *capacity* comes back immediately
+even though the thread does not.
+
+What that costs, honestly: the OS thread is still leaked. It is counted
+(`scheduler.get_worker_status()["abandoned_threads"]`) and logged, so it shows up as a
+number that climbs rather than as a scheduler that mysteriously stalls — but a job that
+hangs forever still holds a thread forever. The fix for that lives in the job, not here.
+
+### What abandoned threads actually cost
+
+It depends entirely on what the job is stuck on, and the two cases are not close (measured
+on CPython 3.12, macOS):
+
+| Abandoned threads | Cost |
+|---|---|
+| Blocked on I/O, a lock, a DB call, `sleep` — the usual case | **~37 KB RSS each**, and no CPU: a blocked thread releases the GIL. 200 of them slowed a Python workload by 1.25x, i.e. noise. A thousand of them is ~37 MB. |
+| Spinning in Python (a runaway loop) | **Ruinous.** They hold the GIL. *Twenty* of them slowed the same workload by **17.6x** — the whole scheduler crawls. No counter can save you from this; it only tells you where to look. |
+
+The terminal state is the OS thread limit (`ulimit -u`; 4000+ on a typical dev machine).
+Reaching it does not crash anything: thread creation fails, the job is rolled back to
+SCHEDULED with its scheduled time intact, its permit is returned, and the scheduler picks it
+up again once threads are free. It simply cannot start new work in the meantime.
+
+So: leaking blocked threads is survivable and visible long before it is fatal. Leaking
+spinning threads is not survivable, and the counter is the fastest way to find out which
+job is doing it.
+
+Timeouts therefore mean different things by function colour, and this is documented on
+`timeout` / `timeout_seconds`: an async job is genuinely cancelled (`asyncio.wait_for`), a
+sync job is marked failed and abandoned. Both raise `JobTimeoutError` and take the same
+retry / `on_failure` path.
+
+### Beyond threads: execution models considered
+
+The thread model's residual problems are really two independent axes, and the candidate
+models each fix a different one:
+
+- **Axis A — scheduler machinery.** Three daemon loops plus a `threading.Timer` per timed
+  job is inelegant. An asyncio-native core would collapse all of it into one event loop and
+  make async cancellation first-class. But it does nothing for axis B — sync jobs still have
+  to run *somewhere* killable — so it is a large rewrite with no safety payoff. Long-term
+  track, not scheduled.
+
+- **Axis B — job isolation.** Only a process can make `timeout` a real guarantee. Measured
+  (CPython 3.12, macOS): `terminate()` kills a sleep-hung child in ~20 ms and a
+  **CPU-spinning one in ~1 ms** — the exact case that poisons the GIL in the thread model
+  (17.6x slowdown from twenty spinners). Dispatch costs 1.5 ms/job (fork) to 25 ms/job
+  (spawn) versus 0.03 ms for a thread — acceptable at Chronis's polling cadence. The real
+  constraint is function resolution: a child process has no registry, so names registered
+  against lambdas/closures cannot be resolved there; a worker-initializer hook is needed.
+  Notably, job `args`/`kwargs` already round-trip through storage as JSON, so the data side
+  of the serialization constraint is already met by contract.
+
+  Subinterpreters (PEP 734) and free-threaded 3.13t would fix the GIL-poisoning but still
+  cannot kill a hung job, and their ecosystems are young — not candidates yet.
+
+The path we would take, when this is picked up: **per-job opt-in process isolation** on top
+of the permit model — `config(isolation="process")`. `WorkerPool` already decoupled capacity
+from threads, so a `ProcessSlot` beside the thread slot is an extension, not a rewrite.
+Threads stay the default (cheap, API unchanged); jobs whose timeout must be a hard guarantee
+opt into a process where timeout means SIGKILL — no zombie, no GIL poisoning. Celery's
+pluggable pools reached the same conclusion.
+
+Deployment-level backstop regardless of model: alert on `abandoned_threads` (or wire it into
+a liveness probe) so a node accumulating zombies gets recycled instead of limping.
+
 ---
 
 ## Known weaknesses (not yet addressed)
 
-### The concurrency model is four models
-
-There are, concurrently: a daemon thread per periodic task (`_PeriodicRunner`), a
-`ThreadPoolExecutor` for sync jobs, a dedicated event-loop thread for async jobs, a
-`threading.Timer` per timed sync job, and the event loop's own default executor for
-post-execution storage writes. Consequences we know about and have not fixed:
-
-- **A timed-out sync job leaks a worker.** Python cannot safely kill a thread, so a timeout
-  marks the job failed and abandons the thread — which keeps occupying a pool slot until it
-  returns, possibly never. (We do release its *queue* slot immediately, so the job is not
-  also blocked from being re-enqueued. The pool slot is still gone.) Enough hung jobs will
-  starve the pool.
-- **`max_workers` does not bound async jobs.** They run on the event loop, not the pool. For
-  an async-heavy workload the only bound is `max_queue_size`.
-- **Timeout semantics differ by function colour.** An async job's timeout is a real
-  cancellation (`asyncio.wait_for`). A sync job's is mark-and-abandon.
-
-An asyncio-native core would collapse most of this. That is a large change and it is not
-scheduled.
-
 ### Misfire policy lives in three places
 
 Classification is in `core/misfire.py`, the SKIP branch in `PollingScheduler._get_ready_jobs`
diff --git a/tests/integration/test_execution_capacity.py b/tests/integration/test_execution_capacity.py
new file mode 100644
index 0000000..3a11111
--- /dev/null
+++ b/tests/integration/test_execution_capacity.py
@@ -0,0 +1,192 @@
+"""max_workers must be a real bound, and a hung job must not take the scheduler down with it."""
+
+import asyncio
+import threading
+import time
+
+import pytest
+
+from chronis import InMemoryStorage, PollingScheduler
+from chronis.core.state import JobStatus
+from chronis.utils.time import utc_now
+
+
+def _due_now(scheduler, job_id):
+    scheduler.storage.update_job(job_id, {"next_run_time": utc_now().isoformat()})
+    scheduler._job_queue.add_job(job_id)
+
+
+@pytest.mark.integration
+class TestTimedOutJobsDoNotStarveTheScheduler:
+    """A timed-out sync job's thread cannot be killed, so it is abandoned.
+
+    Its *capacity* must come back regardless. Held on a ThreadPoolExecutor, max_workers
+    hung jobs would permanently consume every worker and the node would stop executing
+    anything at all — silently.
+    """
+
+    def test_a_full_pool_of_hung_jobs_still_runs_the_next_job(self):
+        storage = InMemoryStorage()
+        scheduler = PollingScheduler(
+            storage_adapter=storage, polling_interval_seconds=1, max_workers=2
+        )
+
+        hang = threading.Event()
+        ran = []
+        scheduler.register_job_function("hang", lambda **kwargs: hang.wait(30))
+        scheduler.register_job_function("normal", lambda **kwargs: ran.append(1))
+
+        # fill every worker with a job that will hang past its timeout
+        for i in range(2):
+            job = scheduler.create_interval_job(
+                func="hang", job_id=f"hang-{i}", seconds=300, timeout_seconds=1
+            )
+            _due_now(scheduler, job.job_id)
+        scheduler._execute_queued_jobs()
+
+        time.sleep(2.5)  # both time out; their threads are still stuck in wait()
+
+        status = scheduler.get_worker_status()
+        assert status["in_use"] == 0  # capacity handed back
+        assert status["abandoned_threads"] == 2  # ...and the leak is visible, not silent
+
+        scheduler.create_interval_job(func="normal", job_id="normal-1", seconds=300)
+        _due_now(scheduler, "normal-1")
+        scheduler._execute_queued_jobs()
+        time.sleep(1.5)
+
+        assert ran == [1]  # the scheduler is still alive
+
+        hang.set()
+        scheduler.stop()
+
+    def test_timed_out_job_is_still_marked_failed(self):
+        """Handing the capacity back must not skip the job's own outcome."""
+        storage = InMemoryStorage()
+        scheduler = PollingScheduler(
+            storage_adapter=storage, polling_interval_seconds=1, max_workers=2
+        )
+
+        hang = threading.Event()
+        scheduler.register_job_function("hang", lambda **kwargs: hang.wait(30))
+
+        job = scheduler.create_interval_job(func="hang", seconds=300, timeout_seconds=1)
+        _due_now(scheduler, job.job_id)
+        scheduler._execute_queued_jobs()
+        time.sleep(2.5)
+
+        assert scheduler.get_job(job.job_id).status == JobStatus.FAILED
+
+        hang.set()
+        scheduler.stop()
+
+
+@pytest.mark.integration
+class TestMaxWorkersBoundsAsyncJobs:
+    """Async jobs run on the event loop, never touching the worker threads.
+
+    They used to be bounded by nothing: max_workers said one thing and the scheduler did
+    another. Capacity is now counted in permits, which async jobs hold too.
+    """
+
+    def test_async_jobs_respect_max_workers(self):
+        storage = InMemoryStorage()
+        scheduler = PollingScheduler(
+            storage_adapter=storage, polling_interval_seconds=1, max_workers=2
+        )
+
+        running = []
+        peak = []
+        counter_lock = threading.Lock()
+
+        async def slow(**kwargs):
+            with counter_lock:
+                running.append(1)
+                peak.append(len(running))
+            await asyncio.sleep(0.4)
+            with counter_lock:
+                running.pop()
+
+        scheduler.register_job_function("slow", slow)
+
+        for i in range(8):
+            job = scheduler.create_interval_job(func="slow", job_id=f"async-{i}", seconds=300)
+            _due_now(scheduler, job.job_id)
+
+        scheduler._execute_queued_jobs()
+        time.sleep(1.5)
+
+        assert max(peak) <= 2  # max_workers, not "however many were queued"
+
+        scheduler.stop()
+
+    def test_saturated_pool_leaves_the_job_scheduled_for_the_next_poll(self):
+        """A job we have no room for must not be claimed — it would strand in RUNNING."""
+        storage = InMemoryStorage()
+        scheduler = PollingScheduler(
+            storage_adapter=storage, polling_interval_seconds=1, max_workers=1
+        )
+
+        hang = threading.Event()
+        scheduler.register_job_function("hang", lambda **kwargs: hang.wait(30))
+
+        scheduler.create_interval_job(func="hang", job_id="busy", seconds=300)
+        _due_now(scheduler, "busy")
+        scheduler._execute_queued_jobs()
+        time.sleep(0.3)
+
+        assert scheduler.get_worker_status()["available"] == 0
+
+        scheduler.create_interval_job(func="hang", job_id="waiting", seconds=300)
+        _due_now(scheduler, "waiting")
+        scheduler._execute_queued_jobs()
+
+        # Not claimed: still SCHEDULED, so the next poll (or another node) can take it
+        assert scheduler.get_job("waiting").status == JobStatus.SCHEDULED
+        assert scheduler.get_job("busy").status == JobStatus.RUNNING
+
+        hang.set()
+        scheduler.stop()
+
+
+@pytest.mark.integration
+class TestUnstartableJobKeepsItsFiring:
+    """A job that cannot be started must not lose the run it was claimed for.
+
+    The claim has already moved next_run_time on to the next occurrence. Rolling back only
+    the status leaves the job SCHEDULED with a future run time — invisible to the poller and
+    to misfire detection — so the firing is silently dropped. Under thread exhaustion that
+    happens to every job that fails to start.
+    """
+
+    def test_rollback_restores_the_scheduled_time(self):
+        from unittest.mock import patch
+
+        storage = InMemoryStorage()
+        scheduler = PollingScheduler(
+            storage_adapter=storage, polling_interval_seconds=1, max_workers=4
+        )
+        ran = []
+        scheduler.register_job_function("f", lambda **kwargs: ran.append(1))
+
+        scheduler.create_interval_job(func="f", job_id="J", seconds=300)
+        _due_now(scheduler, "J")
+        was_due_at = storage.get_job("J")["next_run_time"]
+
+        with patch.object(
+            threading.Thread, "start", side_effect=RuntimeError("can't start new thread")
+        ):
+            scheduler._execute_queued_jobs()
+
+        job = storage.get_job("J")
+        assert job["status"] == JobStatus.SCHEDULED.value
+        assert job["next_run_time"] == was_due_at  # still due; the run was not dropped
+
+        # once threads are available again, the job runs — it did not skip to +300s
+        scheduler._job_queue.add_job("J")
+        scheduler._execute_queued_jobs()
+        time.sleep(0.5)
+
+        assert ran == [1]
+
+        scheduler.stop()
diff --git a/tests/unit/core/test_execute_queued_jobs.py b/tests/unit/core/test_execute_queued_jobs.py
index 5695e89..38777d3 100644
--- a/tests/unit/core/test_execute_queued_jobs.py
+++ b/tests/unit/core/test_execute_queued_jobs.py
@@ -190,7 +190,7 @@ def test_lock_release_error_does_not_abort_batch(self):
         scheduler.lock.release = Mock(side_effect=ConnectionError("lock backend down"))
 
         scheduler._execute_queued_jobs()
-        scheduler._executor.shutdown(wait=True)
+        scheduler._workers.shutdown(wait=True)
 
         assert len(executed) == 3
         assert scheduler._job_queue.get_available_slots() == 5
diff --git a/tests/unit/core/test_worker_pool.py b/tests/unit/core/test_worker_pool.py
new file mode 100644
index 0000000..83bfb5d
--- /dev/null
+++ b/tests/unit/core/test_worker_pool.py
@@ -0,0 +1,149 @@
+"""Unit tests for WorkerPool — capacity accounting that survives abandoned threads."""
+
+import threading
+import time
+
+from chronis.core.execution.worker_pool import WorkerPool
+
+
+class TestReserve:
+    def test_reserve_yields_slots_up_to_max_workers(self):
+        pool = WorkerPool(max_workers=2)
+
+        assert pool.reserve() is not None
+        assert pool.reserve() is not None
+        assert pool.reserve() is None  # saturated
+
+    def test_stats_track_usage(self):
+        pool = WorkerPool(max_workers=3)
+        pool.reserve()
+
+        assert pool.stats() == {
+            "max_workers": 3,
+            "in_use": 1,
+            "available": 2,
+            "abandoned_threads": 0,
+        }
+
+    def test_reserve_returns_none_after_shutdown(self):
+        pool = WorkerPool(max_workers=2)
+        pool.shutdown(wait=False)
+
+        assert pool.reserve() is None
+
+
+class TestRelease:
+    def test_release_returns_capacity(self):
+        pool = WorkerPool(max_workers=1)
+        slot = pool.reserve()
+        assert pool.reserve() is None
+
+        slot.release()
+
+        assert pool.reserve() is not None
+
+    def test_release_is_idempotent(self):
+        """A thread abandoned by a timeout may return later and release a second time.
+
+        If that gave the capacity back twice, the pool would slowly inflate past
+        max_workers and stop bounding anything.
+        """
+        pool = WorkerPool(max_workers=2)
+        slot = pool.reserve()
+
+        slot.release()
+        slot.release()
+        slot.release()
+
+        assert pool.stats()["in_use"] == 0
+        assert pool.reserve() is not None
+        assert pool.reserve() is not None
+        assert pool.reserve() is None  # still capped at 2, not inflated
+
+
+class TestAbandon:
+    def test_abandon_returns_capacity_but_counts_the_thread(self):
+        pool = WorkerPool(max_workers=1)
+        slot = pool.reserve()
+
+        slot.abandon()
+
+        assert pool.stats()["abandoned_threads"] == 1
+        assert pool.reserve() is not None  # capacity came back
+
+    def test_abandon_is_idempotent(self):
+        pool = WorkerPool(max_workers=2)
+        slot = pool.reserve()
+
+        slot.abandon()
+        slot.abandon()
+
+        assert pool.stats()["abandoned_threads"] == 1
+        assert pool.stats()["in_use"] == 0
+
+    def test_release_after_abandon_does_not_double_count(self):
+        """The abandoned thread finishes eventually and releases its slot on the way out."""
+        pool = WorkerPool(max_workers=2)
+        slot = pool.reserve()
+
+        slot.abandon()
+        slot.release()  # the late-returning thread
+
+        assert pool.stats()["in_use"] == 0
+        assert pool.stats()["abandoned_threads"] == 1
+
+
+class TestRun:
+    def test_run_releases_the_slot_when_the_thread_returns(self):
+        pool = WorkerPool(max_workers=1)
+        slot = pool.reserve()
+        done = threading.Event()
+
+        slot.run(done.set, name="job")
+        done.wait(2)
+        time.sleep(0.05)  # let the thread unwind
+
+        assert pool.stats()["in_use"] == 0
+
+    def test_run_releases_the_slot_even_when_the_job_raises(self):
+        pool = WorkerPool(max_workers=1)
+        slot = pool.reserve()
+
+        def boom():
+            raise RuntimeError("job exploded")
+
+        slot.run(boom, name="job")
+        time.sleep(0.1)
+
+        assert pool.stats()["in_use"] == 0
+        assert pool.reserve() is not None
+
+
+class TestShutdown:
+    def test_shutdown_waits_for_live_threads(self):
+        pool = WorkerPool(max_workers=2)
+        finished = []
+
+        slot = pool.reserve()
+        slot.run(lambda: (time.sleep(0.2), finished.append(1)), name="slow")
+
+        assert pool.shutdown(wait=True, timeout=5) is True
+        assert finished == [1]
+
+    def test_shutdown_does_not_wait_for_abandoned_threads(self):
+        """Joining a thread we gave up on would hang shutdown for as long as the job hangs."""
+        pool = WorkerPool(max_workers=2)
+        hang = threading.Event()
+
+        slot = pool.reserve()
+        slot.run(lambda: hang.wait(30), name="zombie")
+        time.sleep(0.05)
+        slot.abandon()
+
+        started = time.time()
+        all_done = pool.shutdown(wait=True, timeout=5)
+        elapsed = time.time() - started
+
+        assert elapsed < 1.0  # did not wait on the zombie
+        assert all_done is True
+        hang.set()
diff --git a/tests/unit/services/test_execution_coordinator.py b/tests/unit/services/test_execution_coordinator.py
index 347e480..83ae3bd 100644
--- a/tests/unit/services/test_execution_coordinator.py
+++ b/tests/unit/services/test_execution_coordinator.py
@@ -9,6 +9,7 @@
 
 from chronis.core.execution.coordinator import ClaimedJob, ExecutionCoordinator, _OutcomeGate
 from chronis.core.execution.job_executor import JobExecutor
+from chronis.core.execution.worker_pool import Slot, WorkerPool
 from chronis.core.state import JobStatus
 
 
@@ -17,6 +18,18 @@ def claimed(job_data, original_scheduled_time=None):
     return ClaimedJob(data=job_data, original_scheduled_time=original_scheduled_time)
 
 
+def a_slot(pool=None):
+    """Reserve real capacity, as try_execute does before handing a job to _trigger_execution."""
+    slot = (pool or WorkerPool(max_workers=4)).reserve()
+    assert slot is not None
+    return slot
+
+
+def mock_slot():
+    """A stand-in slot, so a test can see whether the job was handed to a worker thread."""
+    return Mock(spec=Slot)
+
+
 class TestTriggerExecutionLogging:
     """Test _trigger_execution logging behavior."""
 
@@ -25,7 +38,7 @@ def test_trigger_execution_does_not_log_info(self):
         coordinator = ExecutionCoordinator(
             storage=Mock(),
             lock=Mock(),
-            executor=Mock(),
+            workers=WorkerPool(max_workers=4),
             function_registry={},
             failure_handler_registry={},
             success_handler_registry={},
@@ -36,23 +49,22 @@ def test_trigger_execution_does_not_log_info(self):
         )
 
         coordinator._update_job_status = Mock()
-        coordinator.executor.submit = Mock(return_value=Mock(add_done_callback=Mock()))
 
         job_logger = Mock()
-        coordinator._trigger_execution(claimed({"job_id": "test-1"}), job_logger, Mock())
+        coordinator._trigger_execution(claimed({"job_id": "test-1"}), job_logger, Mock(), a_slot())
 
         # Verify info was NOT called
         job_logger.info.assert_not_called()
 
     def test_exception_during_trigger_is_logged(self):
         """Test that exception during execution trigger is logged."""
-        executor_mock = Mock()
-        executor_mock.submit.side_effect = Exception("Submit failed")
+        failing_slot = mock_slot()
+        failing_slot.run.side_effect = Exception("Submit failed")
 
         coordinator = ExecutionCoordinator(
             storage=Mock(),
             lock=Mock(),
-            executor=executor_mock,
+            workers=WorkerPool(max_workers=4),
             function_registry={},
             failure_handler_registry={},
             success_handler_registry={},
@@ -65,7 +77,9 @@ def test_exception_during_trigger_is_logged(self):
 
         # Should raise exception and rollback
         with pytest.raises(Exception, match="Submit failed"):
-            coordinator._trigger_execution(claimed({"job_id": "test-1"}), job_logger, Mock())
+            coordinator._trigger_execution(
+                claimed({"job_id": "test-1"}), job_logger, Mock(), failing_slot
+            )
 
         # Verify warning was logged (rollback message)
         job_logger.warning.assert_called_once()
@@ -85,7 +99,7 @@ def test_successful_execution_with_previous_retries_resets_count(self, mock_utc_
         coordinator = ExecutionCoordinator(
             storage=storage_mock,
             lock=Mock(),
-            executor=Mock(),
+            workers=WorkerPool(max_workers=4),
             function_registry={"test_func": lambda: None},
             failure_handler_registry={},
             success_handler_registry={},
@@ -145,7 +159,7 @@ def test_storage_error_during_retry_scheduling_is_logged(self, mock_utc_now, moc
         coordinator = ExecutionCoordinator(
             storage=storage_mock,
             lock=Mock(),
-            executor=Mock(),
+            workers=WorkerPool(max_workers=4),
             function_registry={},
             failure_handler_registry={},
             success_handler_registry={},
@@ -183,7 +197,7 @@ def failing_callback(job_id, job_info):
         coordinator = ExecutionCoordinator(
             storage=Mock(),
             lock=Mock(),
-            executor=Mock(),
+            workers=WorkerPool(max_workers=4),
             function_registry={},
             failure_handler_registry={},
             success_handler_registry={"test-1": failing_callback},
@@ -224,7 +238,7 @@ def failing_global(job_id, job_info):
         coordinator = ExecutionCoordinator(
             storage=Mock(),
             lock=Mock(),
-            executor=Mock(),
+            workers=WorkerPool(max_workers=4),
             function_registry={},
             failure_handler_registry={},
             success_handler_registry={},
@@ -260,7 +274,7 @@ def failing_callback(job_id, error, job_info):
         coordinator = ExecutionCoordinator(
             storage=Mock(),
             lock=Mock(),
-            executor=Mock(),
+            workers=WorkerPool(max_workers=4),
             function_registry={},
             failure_handler_registry={"test-1": failing_callback},
             success_handler_registry={},
@@ -299,7 +313,7 @@ def failing_global(job_id, error, job_info):
         coordinator = ExecutionCoordinator(
             storage=Mock(),
             lock=Mock(),
-            executor=Mock(),
+            workers=WorkerPool(max_workers=4),
             function_registry={},
             failure_handler_registry={},
             success_handler_registry={},
@@ -330,19 +344,17 @@ def failing_global(job_id, error, job_info):
 class TestExecutorSubmitRollback:
     """Test executor submit failure rollback behavior."""
 
-    def test_trigger_execution_rolls_back_on_submit_failure(self):
-        """ThreadPool submit 실패 시 SCHEDULED로 복구."""
-        # Mock storage
+    def test_trigger_execution_rolls_back_on_start_failure(self):
+        """job을 시작조차 못 하면 SCHEDULED로 복구."""
         storage_mock = Mock()
 
-        # Mock executor that raises on submit
-        executor_mock = Mock()
-        executor_mock.submit.side_effect = RuntimeError("ThreadPool is shutting down")
+        failing_slot = mock_slot()
+        failing_slot.run.side_effect = RuntimeError("ThreadPool is shutting down")
 
         coordinator = ExecutionCoordinator(
             storage=storage_mock,
             lock=Mock(),
-            executor=executor_mock,
+            workers=WorkerPool(max_workers=4),
             function_registry={},
             failure_handler_registry={},
             success_handler_registry={},
@@ -360,7 +372,7 @@ def test_trigger_execution_rolls_back_on_submit_failure(self):
 
         # Should raise RuntimeError
         with pytest.raises(RuntimeError, match="ThreadPool is shutting down"):
-            coordinator._trigger_execution(claimed(job_data), job_logger, Mock())
+            coordinator._trigger_execution(claimed(job_data), job_logger, Mock(), failing_slot)
 
         # Verify status update - only one call for rollback
         # (RUNNING update now happens in try_execute before calling _trigger_execution)
@@ -382,14 +394,11 @@ def test_trigger_execution_rolls_back_on_submit_failure(self):
     def test_trigger_execution_succeeds_normally(self):
         """정상적인 경우 rollback 발생 안 함."""
         storage_mock = Mock()
-        executor_mock = Mock()
-        future_mock = Mock()
-        executor_mock.submit.return_value = future_mock
 
         coordinator = ExecutionCoordinator(
             storage=storage_mock,
             lock=Mock(),
-            executor=executor_mock,
+            workers=WorkerPool(max_workers=4),
             function_registry={},
             failure_handler_registry={},
             success_handler_registry={},
@@ -401,20 +410,16 @@ def test_trigger_execution_succeeds_normally(self):
 
         job_data = {"job_id": "test-1", "status": "scheduled"}
         job_logger = Mock()
+        slot = mock_slot()
 
         # Should not raise
-        coordinator._trigger_execution(claimed(job_data), job_logger, Mock())
+        coordinator._trigger_execution(claimed(job_data), job_logger, Mock(), slot)
 
+        slot.run.assert_called_once()  # handed to a worker thread
         # RUNNING update는 이제 try_execute에서 발생하므로, _trigger_execution에서는 없어야 함
         assert storage_mock.update_job.call_count == 0
 
-        # Verify submit was called
-        executor_mock.submit.assert_called_once()
-
-        # Verify callback registered
-        future_mock.add_done_callback.assert_called_once()
-
-        # Verify no warning logged
+        # Verify no warning logged (i.e. no rollback)
         job_logger.warning.assert_not_called()
 
 
@@ -430,7 +435,7 @@ async def async_func():
         coordinator = ExecutionCoordinator(
             storage=Mock(),
             lock=Mock(),
-            executor=Mock(),
+            workers=WorkerPool(max_workers=4),
             function_registry={"async_func": async_func},
             failure_handler_registry={},
             success_handler_registry={},
@@ -450,11 +455,10 @@ async def async_func():
         coordinator._job_executor._async_thread = Mock()
 
         with patch("asyncio.run_coroutine_threadsafe", return_value=mock_future) as mock_rcts:
-            coordinator._trigger_execution(claimed(job_data), Mock(), on_complete)
+            coordinator._trigger_execution(claimed(job_data), Mock(), on_complete, a_slot())
 
-        # Async path should be used, not executor.submit
+        # Async path should be used; no worker thread is spawned for it
         mock_rcts.assert_called_once()
-        coordinator.executor.submit.assert_not_called()
         mock_future.add_done_callback.assert_called_once()
 
     def test_sync_job_dispatches_to_thread_pool(self):
@@ -463,14 +467,12 @@ def test_sync_job_dispatches_to_thread_pool(self):
         def sync_func():
             pass
 
-        executor_mock = Mock()
-        future_mock = Mock()
-        executor_mock.submit.return_value = future_mock
+        slot = mock_slot()
 
         coordinator = ExecutionCoordinator(
             storage=Mock(),
             lock=Mock(),
-            executor=executor_mock,
+            workers=WorkerPool(max_workers=4),
             function_registry={"sync_func": sync_func},
             failure_handler_registry={},
             success_handler_registry={},
@@ -482,10 +484,10 @@ def sync_func():
         job_data = {"job_id": "test-1", "func_name": "sync_func"}
 
         with patch("asyncio.run_coroutine_threadsafe") as mock_rcts:
-            coordinator._trigger_execution(claimed(job_data), Mock(), Mock())
+            coordinator._trigger_execution(claimed(job_data), Mock(), Mock(), slot)
 
-        executor_mock.submit.assert_called_once()
-        mock_rcts.assert_not_called()
+        slot.run.assert_called_once()  # handed to a worker thread
+        mock_rcts.assert_not_called()  # not to the event loop
 
 
 class TestEnsureAsyncLoopThreadSafety:
@@ -526,13 +528,13 @@ def test_rollback_failure_does_not_mask_original_error(self):
         storage_mock = Mock()
         storage_mock.update_job.side_effect = RuntimeError("Storage down")
 
-        executor_mock = Mock()
-        executor_mock.submit.side_effect = RuntimeError("ThreadPool is shutting down")
+        failing_slot = mock_slot()
+        failing_slot.run.side_effect = RuntimeError("ThreadPool is shutting down")
 
         coordinator = ExecutionCoordinator(
             storage=storage_mock,
             lock=Mock(),
-            executor=executor_mock,
+            workers=WorkerPool(max_workers=4),
             function_registry={},
             failure_handler_registry={},
             success_handler_registry={},
@@ -545,7 +547,9 @@ def test_rollback_failure_does_not_mask_original_error(self):
 
         # Original error should propagate, not the rollback error
         with pytest.raises(RuntimeError, match="ThreadPool is shutting down"):
-            coordinator._trigger_execution(claimed({"job_id": "test-1"}), job_logger, Mock())
+            coordinator._trigger_execution(
+                claimed({"job_id": "test-1"}), job_logger, Mock(), failing_slot
+            )
 
         # Rollback was attempted
         storage_mock.update_job.assert_called_once()
@@ -590,7 +594,7 @@ def _make_coordinator(self, **overrides):
         defaults = {
             "storage": Mock(),
             "lock": Mock(),
-            "executor": Mock(),
+            "workers": WorkerPool(max_workers=4),
             "function_registry": {},
             "failure_handler_registry": {},
             "success_handler_registry": {},
@@ -743,7 +747,7 @@ def global_callback(job_id, job_info):
         coordinator = ExecutionCoordinator(
             storage=Mock(),
             lock=Mock(),
-            executor=Mock(),
+            workers=WorkerPool(max_workers=4),
             function_registry={},
             failure_handler_registry={},
             success_handler_registry={"test-1": job_callback},
@@ -784,7 +788,7 @@ def global_callback(job_id, error, job_info):
         coordinator = ExecutionCoordinator(
             storage=Mock(),
             lock=Mock(),
-            executor=Mock(),
+            workers=WorkerPool(max_workers=4),
             function_registry={},
             failure_handler_registry={"test-1": job_callback},
             success_handler_registry={},
@@ -824,7 +828,7 @@ def callback(job_id, job_info):
         coordinator = ExecutionCoordinator(
             storage=Mock(),
             lock=Mock(),
-            executor=Mock(),
+            workers=WorkerPool(max_workers=4),
             function_registry={},
             failure_handler_registry={},
             success_handler_registry={"test-1": callback},
@@ -866,7 +870,7 @@ def on_fail(job_id, error, job_info):
         coordinator = ExecutionCoordinator(
             storage=Mock(),
             lock=Mock(),
-            executor=Mock(),
+            workers=WorkerPool(max_workers=4),
             function_registry={"test_func": lambda: None},
             failure_handler_registry={"test-1": on_fail},
             success_handler_registry={},
@@ -905,7 +909,7 @@ def _make_coordinator(self):
         return ExecutionCoordinator(
             storage=Mock(),
             lock=Mock(),
-            executor=Mock(),
+            workers=WorkerPool(max_workers=4),
             function_registry={},
             failure_handler_registry={},
             success_handler_registry={},