```
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
    ):
        """
        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`.
        """
        if args is None:
            args = ()

        async def task():
            try:
                if delay:
                    await asyncio.sleep(delay)
                if asyncio.iscoroutinefunction(func):
                    result = await func(*args)
                else:
                    # Run synchronous functions in the default executor to avoid blocking
                    loop = asyncio.get_running_loop()
                    result = await loop.run_in_executor(None, func, *args)
                if callback:
                    if asyncio.iscoroutinefunction(callback):
                        await callback(result)
                    else:
                        loop = asyncio.get_running_loop()
                        await loop.run_in_executor(None, callback, result)
                except Exception as e:
                if error_callback:
                    try:
                        if asyncio.iscoroutinefunction(error_callback):
                            await error_callback(e)
                        else:
                            loop = asyncio.get_running_loop()
                            await loop.run_in_executor(None, 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
    ) -> 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`.
        :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:
                loop = asyncio.get_running_loop()
                return await loop.run_in_executor(None, 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()

# Assuming ThreadWorker is already defined as above

def task1(a, b, c):
    print(f"[task1] Executed with arguments: {a}, {b}, {c}")
    return f"Result from task1 with args ({a}, {b}, {c})"


async def task2():
    print("[task2] Started")
    await asyncio.sleep(1)
    print("[task2] Completed")
    return "Result from task2"


async def task3(worker: ThreadWorker):
    print("[task3] Awaiting task1 with 2-second delay")
    result = await worker.after_async(task1, delay=2, args=(1, 2, 3))
    print(f"[task3] Received: {result}")

    await asyncio.sleep(1)
    print("[task3] Printing 'Foo'")
    print("Foo")

    print("[task3] Stopping worker")


def callback(result: Any):
    print(f"[callback] Received: {result}")


with ThreadWorker() as worker:
    # Schedule various tasks
    worker.after(task1, args=(10, 20, 30))  # Synchronous task without callback
    worker.after(task2, callback=callback)  # Asynchronous task with callback
    worker.after(task1, callback=callback, delay=3, args=(40, 50, 60))  # Synchronous task with callback and delay
    worker.after(task1, args=(70, 80, 90))  # Another synchronous task without callback

    # Schedule an asynchronous task that utilizes after_async
    worker.after(task3, args=(worker,))

print("Worker has been stopped.")

```
in this class, when a function/callback is not a corotine, it uses `await loop.run_in_executor(None, callback, e)`
this may create other threads by the default threadexecutor.
why not run the callback synchronously??