```
import asyncio
from typing import Any

import asyncio
import threading
from typing import Callable, Any, Optional
import sys
import traceback

class ThreadWorker:
    def __init__(self):
        """
        Initializes the ThreadWorker by starting a new thread
        that runs an asyncio event loop.
        """
        self.loop = None
        self.loop_started = threading.Event()  # Event to signal loop initialization
        self.thread = threading.Thread(target=self.loop_task, daemon=True)
        self.thread.start()

        # Wait until the loop is initialized
        loop_initialized = self.loop_started.wait(timeout=5)  # Timeout to prevent indefinite blocking
        if not loop_initialized:
            raise RuntimeError("Failed to initialize the event loop within the timeout period.")

        self.task_counter = 0
        self.stop_requested = False
        self.lock = threading.Lock()  # To protect access to task_counter and stop_requested

    def loop_task(self):
        """Starts the asyncio event loop in a separate thread."""
        self.loop = asyncio.new_event_loop()
        asyncio.set_event_loop(self.loop)
        self.loop_started.set()  # Signal that the loop is initialized
        try:
            self.loop.run_forever()
        finally:
            self.loop.close()

    def after(
        self,
        func: Callable[..., Any],
        callback: Optional[Callable[[Any], None]] = None,
        error_callback: Optional[Callable[[Exception], None]] = None,
        delay: Optional[float] = None,
        args: Optional[Any] = None,
        use_executor: bool = True  # New parameter
    ):
        """
        Schedules a function or coroutine to be executed in the event loop.

        :param func: The function or coroutine to execute.
        :param callback: Optional function to call with the result.
        :param error_callback: Optional function to call with the exception if an error occurs.
        :param delay: Optional delay in seconds before execution.
        :param args: Optional positional arguments to pass to `func`.
        :param use_executor: If True, run synchronous functions via executor;
                                if False, run them directly in the event loop.
        """
        if args is None:
            args = ()

        async def task():
            try:
                if delay:
                    await asyncio.sleep(delay)
                if asyncio.iscoroutinefunction(func):
                    result = await func(*args)
                else:
                    if use_executor:
                        # Run synchronous functions in the executor
                        loop = asyncio.get_running_loop()
                        result = await loop.run_in_executor(None, func, *args)
                    else:
                        # Run synchronous function directly (may block the event loop)
                        result = func(*args)
                if callback:
                    if asyncio.iscoroutinefunction(callback):
                        await callback(result)
                    else:
                        if use_executor:
                            loop = asyncio.get_running_loop()
                            await loop.run_in_executor(None, callback, result)
                        else:
                            callback(result)
            except Exception as e:
                if error_callback:
                    try:
                        if asyncio.iscoroutinefunction(error_callback):
                            await error_callback(e)
                        else:
                            if use_executor:
                                loop = asyncio.get_running_loop()
                                await loop.run_in_executor(None, error_callback, e)
                            else:
                                error_callback(e)
                    except Exception as callback_exc:
                        # Propagate the original exception if error_callback fails
                        print("[ThreadWorker] Exception in error_callback:", file=sys.stderr)
                        traceback.print_exc()
                        raise e from callback_exc
                else:
                    # No error_callback provided; propagate the exception
                    print("[ThreadWorker] Exception in scheduled task:", file=sys.stderr)
                    traceback.print_exc()
                    raise e
            finally:
                with self.lock:
                    self.task_counter -= 1  # Decrement the counter when task completes
                    if self.stop_requested and self.task_counter == 0:
                        self.loop.call_soon_threadsafe(self.loop.stop)  # Safely stop the loop

        # Schedule the coroutine to run in the event loop thread
        with self.lock:
            self.task_counter += 1  # Increment the counter for the new task
            asyncio.run_coroutine_threadsafe(task(), self.loop)

    async def after_async(
        self,
        func: Callable[..., Any],
        delay: Optional[float] = None,
        args: Optional[Any] = None,
        use_executor: bool = True  # New parameter
    ) -> Any:
        """
        Asynchronously schedules a function or coroutine to be executed in the event loop
        and awaits its result.

        :param func: The function or coroutine to execute.
        :param delay: Optional delay in seconds before execution.
        :param args: Optional positional arguments to pass to `func`.
        :param use_executor: If True, run synchronous functions via executor;
                                if False, run them directly in the event loop.
        :return: The result of `func`.
        :raises Exception: Propagates any exception raised by `func`.
        """
        if args is None:
            args = ()

        async def internal_task():
            if delay:
                await asyncio.sleep(delay)
            if asyncio.iscoroutinefunction(func):
                return await func(*args)
            else:
                if use_executor:
                    loop = asyncio.get_running_loop()
                    return await loop.run_in_executor(None, func, *args)
                else:
                    return func(*args)

        # Increment the task counter before scheduling the task
        with self.lock:
            self.task_counter += 1

        # Submit the coroutine to the worker's event loop
        future = asyncio.run_coroutine_threadsafe(internal_task(), self.loop)

        try:
            # Await the result in the caller's event loop
            return await asyncio.wrap_future(future)
        except Exception as e:
            # Propagate exceptions to the caller
            raise e
        finally:
            with self.lock:
                self.task_counter -= 1
                if self.stop_requested and self.task_counter == 0:
                    self.loop.call_soon_threadsafe(self.loop.stop)

    def stop(self):
        """
        Stops the event loop gracefully. It sets a flag to indicate that a stop has been requested.
        The loop will stop once all pending tasks are completed.
        """
        with self.lock:
            if self.stop_requested:
                return  # Stop already requested
            self.stop_requested = True
            if self.task_counter == 0:
                # No pending tasks, stop immediately
                self.loop.call_soon_threadsafe(self.loop.stop)

    def join(self, timeout: Optional[float] = None):
        """
        Waits for the worker thread to finish.

        :param timeout: Optional timeout in seconds.
        """
        self.thread.join(timeout)

    def __enter__(self):
        """
        Enables the use of 'with ThreadWorker() as worker:'
        """
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        """
        Ensures that the worker is stopped and joined upon exiting the context.
        """
        self.stop()
        self.join()

```

should delay be implemented via loop.call_later instead of asyncio.sleep?