feat(scheduling): pluggable task executor (asyncio|thread) (v26.06.67)

CHANGELOG.md

@@ -6,6 +6,16 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 
 ---
 
+## v26.06.67 (2026-06-07)
+
+### Added (scheduling — pluggable task executor; fixes a weak default)
+
+- The `TaskScheduler`'s executor is now selectable via `pyfly.scheduling.executor.type`
+  (`asyncio`, default — in-loop tasks; or `thread` — offload blocking jobs to a pool of
+  `pyfly.scheduling.executor.max-workers`, default 4). The `ThreadPoolTaskExecutor` already
+  existed and was tested but was never reachable through configuration — the auto-config
+  hardcoded `AsyncIOTaskExecutor`. Found by the ports/adapters audit (weak default).
+
 ## v26.06.66 (2026-06-07)
 
 ### Added (distributed lock — Postgres advisory-lock adapter; Postgres parity)

README.md

@@ -11,7 +11,7 @@
   <a href="https://github.com/fireflyframework"><img src="https://img.shields.io/badge/Firefly_Framework-official-ff6600?logo=data:image/svg+xml;base64,PHN2ZyB4bWxucz0iaHR0cDovL3d3dy53My5vcmcvMjAwMC9zdmciIHZpZXdCb3g9IjAgMCAyNCAyNCI+PHBhdGggZmlsbD0id2hpdGUiIGQ9Ik0xMiAyQzYuNDggMiAyIDYuNDggMiAxMnM0LjQ4IDEwIDEwIDEwIDEwLTQuNDggMTAtMTBTMTcuNTIgMiAxMiAyeiIvPjwvc3ZnPg==" alt="Firefly Framework"></a>
   <a href="https://www.python.org/"><img src="https://img.shields.io/badge/python-3.12%2B-blue?logo=python&logoColor=white" alt="Python 3.12+"></a>
   <a href="LICENSE"><img src="https://img.shields.io/badge/license-Apache%202.0-green" alt="License: Apache 2.0"></a>
-  <a href="#"><img src="https://img.shields.io/badge/version-26.06.66-brightgreen" alt="Version: 26.06.66"></a>
+  <a href="#"><img src="https://img.shields.io/badge/version-26.06.67-brightgreen" alt="Version: 26.06.67"></a>
   <a href="#"><img src="https://img.shields.io/badge/type--checked-mypy%20strict-blue?logo=python&logoColor=white" alt="Type Checked: mypy strict"></a>
   <a href="#"><img src="https://img.shields.io/badge/code%20style-ruff-purple?logo=ruff&logoColor=white" alt="Code Style: Ruff"></a>
   <a href="#"><img src="https://img.shields.io/badge/async-first-brightgreen" alt="Async First"></a>

pyproject.toml

@@ -7,7 +7,7 @@ name = "pyfly"
 # CalVer YY.MM.PATCH — package metadata uses PEP 440 normalized form (26.5.4);
 # git tag, GitHub release and human-readable display use leading-zero form
 # (v26.05.04) to match the Java/.NET/Go siblings.
-version = "26.6.66"
+version = "26.6.67"
 description = "The official Python implementation of the Firefly Framework — DI, CQRS, EDA, hexagonal architecture, and more."
 readme = "README.md"
 license = "Apache-2.0"

src/pyfly/__init__.py

@@ -13,4 +13,4 @@
 # limitations under the License.
 """PyFly — Enterprise Python Framework."""
 
-__version__ = "26.06.66"
+__version__ = "26.06.67"

src/pyfly/scheduling/auto_configuration.py

@@ -30,6 +30,7 @@
 from pyfly.context.conditions import auto_configuration, conditional_on_class
 from pyfly.core.config import Config
 from pyfly.scheduling.lock import DistributedLock, InProcessDistributedLock, LocalLock
+from pyfly.scheduling.ports.outbound import TaskExecutorPort
 
 
 @auto_configuration
@@ -74,11 +75,19 @@ def _engine() -> Any:
         return LocalLock()
 
     @bean
-    def task_scheduler(self, container: Container) -> TaskScheduler:
+    def task_scheduler(self, container: Container, config: Config) -> TaskScheduler:
         # Resolve the DistributedLock bean above for @scheduled(lock=...) coordination;
         # fall back to the scheduler's own LocalLock if (unexpectedly) absent.
         try:
             lock = container.resolve(DistributedLock)  # type: ignore[type-abstract]
         except (NoSuchBeanError, NoUniqueBeanError):
             lock = None
-        return TaskScheduler(lock=lock)
+
+        # Executor backend: pyfly.scheduling.executor.type = 'asyncio' (default, in-loop tasks)
+        # or 'thread' (offload blocking jobs to a pool of pyfly.scheduling.executor.max-workers).
+        executor: TaskExecutorPort | None = None
+        if str(config.get("pyfly.scheduling.executor.type", "asyncio")).lower() == "thread":
+            from pyfly.scheduling.adapters.thread_executor import ThreadPoolTaskExecutor
+
+            executor = ThreadPoolTaskExecutor(max_workers=int(config.get("pyfly.scheduling.executor.max-workers", 4)))
+        return TaskScheduler(executor=executor, lock=lock)

tests/scheduling/test_executor_selection.py

@@ -0,0 +1,44 @@
+# Copyright 2026 Firefly Software Foundation.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""TaskScheduler executor backend selection (v26.06.67)."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from pyfly.container.container import Container
+from pyfly.core.config import Config
+from pyfly.scheduling.adapters.asyncio_executor import AsyncIOTaskExecutor
+from pyfly.scheduling.adapters.thread_executor import ThreadPoolTaskExecutor
+from pyfly.scheduling.auto_configuration import SchedulingAutoConfiguration
+
+
+def _executor(config: Config) -> Any:
+    return SchedulingAutoConfiguration().task_scheduler(Container(), config)._executor
+
+
+def test_default_executor_is_asyncio() -> None:
+    assert isinstance(_executor(Config({})), AsyncIOTaskExecutor)
+
+
+def test_thread_executor_selected_by_config() -> None:
+    cfg = Config({"pyfly": {"scheduling": {"executor": {"type": "thread"}}}})
+    assert isinstance(_executor(cfg), ThreadPoolTaskExecutor)
+
+
+def test_thread_executor_respects_max_workers() -> None:
+    cfg = Config({"pyfly": {"scheduling": {"executor": {"type": "thread", "max-workers": 7}}}})
+    executor = _executor(cfg)
+    assert isinstance(executor, ThreadPoolTaskExecutor)
+    assert executor._executor._max_workers == 7