Release version 0.6.0 with significant enhancements and new features

- Introduced a TaskRoutes extension mechanism, allowing custom TaskRoutes injection via `task_routes_class` parameter in `create_a2a_server()` and `_create_request_handler()`, eliminating the need for monkey patching.
- Added Task Tree lifecycle hooks with a new `register_task_tree_hook()` decorator for tracking events such as tree creation, start, completion, and failure.
- Implemented executor-specific hooks for pre and post-execution logic, enhancing task execution flexibility.
- Comprehensive documentation updates for new hook types and usage examples.
- Removed the redundant `decorators.py` file, consolidating functionality into `core/decorators.py`.
- Updated version references across project files to reflect the new release.
- Added tests to validate the new features and ensure robust functionality.

Author: tercel

CHANGELOG.md

@@ -5,7 +5,77 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
-## [Unreleased] 
+## [0.6.0] 2025-12-10
+
+### Added
+- **TaskRoutes Extension Mechanism**
+  - Added `task_routes_class` parameter to `create_a2a_server()` and `_create_request_handler()` for custom TaskRoutes injection
+  - Eliminates the need for monkey patching when extending TaskRoutes functionality
+  - Supports custom routes via `custom_routes` parameter in `CustomA2AStarletteApplication`
+  - Backward compatible: optional parameter with default `TaskRoutes` class
+  - Usage: `create_a2a_server(task_routes_class=CustomTaskRoutes, custom_routes=[...])`
+
+- **Task Tree Lifecycle Hooks**
+  - New `register_task_tree_hook()` decorator for task tree lifecycle events
+  - Four hook types: `on_tree_created`, `on_tree_started`, `on_tree_completed`, `on_tree_failed`
+  - Explicit lifecycle tracking without manual root task detection
+  - Hooks receive root task and relevant context (status, error message)
+  - Usage: `@register_task_tree_hook("on_tree_completed") async def on_completed(root_task, status): ...`
+
+- **Executor-Specific Hooks**
+  - Added `pre_hook` and `post_hook` parameters to `@executor_register()` decorator
+  - Runtime hook registration via `add_executor_hook(executor_id, hook_type, hook_func)`
+  - Inject custom logic (e.g., quota checks, demo data fallback) for specific executors
+  - `pre_hook` can return a result to skip executor execution (useful for demo mode)
+  - `post_hook` receives executor, task, inputs, and result for post-processing
+  - Supports both decorator-based and runtime registration for existing executors
+
+- **Automatic user_id Extraction**
+  - Automatic `user_id` extraction from JWT token in `TaskRoutes.handle_task_generate` and `handle_task_create`
+  - Only extracts from JWT token payload for security (HTTP headers can be spoofed)
+  - Supports `user_id` field or standard JWT `sub` claim in token payload
+  - Extracted `user_id` automatically set on task data
+  - Simplifies custom route implementations and ensures consistent user identification
+  - Security: Only trusted JWT tokens are used, no fallback to HTTP headers
+
+- **Demo Mode Support**
+  - Built-in demo mode via `use_demo` parameter in task inputs
+  - CLI support: `--use-demo` flag for `apflow run flow` command
+  - API support: `use_demo` parameter in task creation and execution
+  - Executors can override `get_demo_result()` method in `BaseTask` for custom demo data
+  - Default demo data format: `{"result": "Demo execution result", "demo_mode": True}`
+  - All built-in executors now implement `get_demo_result()` method:
+    - `SystemInfoExecutor`, `CommandExecutor`, `AggregateResultsExecutor`
+    - `RestExecutor`, `GenerateExecutor`, `ApiExecutor`
+    - `SshExecutor`, `GrpcExecutor`, `WebSocketExecutor`
+    - `McpExecutor`, `DockerExecutor`
+    - `CrewManager`, `BatchManager` (CrewAI executors)
+  - **Realistic Demo Execution Timing**: All executors include `_demo_sleep` values to simulate real execution time:
+    - Network operations (HTTP, SSH, API): 0.2-0.5 seconds
+    - Container operations (Docker): 1.0 second
+    - LLM operations (CrewAI, Generate): 1.0-1.5 seconds
+    - Local operations (SystemInfo, Command, Aggregate): 0.05-0.1 seconds
+  - **Global Demo Sleep Scale**: Configurable via `AIPARTNERUPFLOW_DEMO_SLEEP_SCALE` environment variable (default: 1.0)
+    - Allows adjusting demo execution speed globally (e.g., `0.5` for faster, `2.0` for slower)
+    - API: `set_demo_sleep_scale(scale)` and `get_demo_sleep_scale()` functions
+  - **CrewAI Demo Support**: `CrewManager` and `BatchManager` generate realistic demo results:
+    - Based on `works` definition (agents and tasks) from task params or schemas
+    - Includes simulated `token_usage` matching real LLM execution patterns
+    - `BatchManager` aggregates token usage across multiple works
+  - Demo mode helps developers test workflows without external dependencies
+
+- **TaskModel Customization Improvements**
+  - Enhanced `set_task_model_class()` with improved validation and error messages
+  - New `@task_model_register()` decorator for convenient TaskModel registration
+  - Validation ensures custom classes inherit from `TaskModel` with helpful error messages
+  - Supports `__table_args__ = {'extend_existing': True}` for extending existing table definitions
+  - Better support for user-defined `MyTaskModel(TaskModel)` with additional fields
+
+- **Documentation for Hook Types**
+  - Added comprehensive documentation explaining differences between hook types
+  - `pre_hook` / `post_hook`: Task-level hooks for individual task execution
+  - `task_tree_hook`: Task tree-level hooks for tree lifecycle events
+  - Clear usage scenarios and examples in `docs/development/extending.md`
 
 ### Changed
 - **LLM Model Parameter Naming**: Unified LLM model parameter naming to `model` across all components
@@ -19,6 +89,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   - **Impact**: Only affects generate functionality introduced in 0.5.0, minimal breaking change
   - **Migration**: Update any code using `llm_model` parameter to use `model` instead
 
+### Removed
+- **Redundant decorators.py file**
+  - Removed `src/aipartnerupflow/decorators.py` as it was no longer used
+  - Functionality superseded by `src/aipartnerupflow/core/decorators.py`
+  - No impact on existing code (file was not imported by any other modules)
+
 
 ## [0.5.0] 2025-12-7
 

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "aipartnerupflow"
-version = "0.5.0"
+version = "0.6.0"
 description = "Agent workflow orchestration and execution platform"
 readme = "README.md"
 requires-python = ">=3.10"

src/aipartnerupflow/__init__.py

@@ -19,7 +19,7 @@
 Protocol Standard: A2A (Agent-to-Agent) Protocol
 """
 
-__version__ = "0.5.0"
+__version__ = "0.6.0"
 
 # Core framework - re-export from core module for convenience
 from aipartnerupflow.core import (
@@ -38,8 +38,11 @@
 from aipartnerupflow.core.decorators import (
     register_pre_hook,
     register_post_hook,
+    register_task_tree_hook,
+    get_task_tree_hooks,
     set_task_model_class,
     get_task_model_class,
+    task_model_register,
     clear_config,
     set_use_task_creator,
     get_use_task_creator,
@@ -51,6 +54,9 @@
     tool_register,
 )
 
+# Extension registry utilities
+from aipartnerupflow.core.extensions import add_executor_hook
+
 __all__ = [
     # Core framework (from core module)
     "ExecutableTask",
@@ -65,8 +71,11 @@
     # Unified decorators (Flask-style API)
     "register_pre_hook",
     "register_post_hook",
+    "register_task_tree_hook",
+    "get_task_tree_hooks",
     "set_task_model_class",
     "get_task_model_class",
+    "task_model_register",
     "clear_config",
     "set_use_task_creator",
     "get_use_task_creator",
@@ -76,6 +85,8 @@
     "storage_register",
     "hook_register",
     "tool_register",
+    # Extension registry utilities
+    "add_executor_hook",
     # Version
     "__version__",
 ]

src/aipartnerupflow/api/a2a/agent_executor.py

@@ -235,11 +235,28 @@ async def _execute_simple_mode(
             if require_existing_tasks is None:
                 require_existing_tasks = None  # Use global configuration
 
+            # Extract use_demo from metadata, message.metadata, or configuration
+            use_demo = False
+            if context.metadata:
+                use_demo = context.metadata.get("use_demo", False)
+            # Also check message.metadata if available (A2A protocol supports metadata in message)
+            if not use_demo and context.message and hasattr(context.message, "metadata"):
+                message_metadata = context.message.metadata
+                if isinstance(message_metadata, dict):
+                    use_demo = message_metadata.get("use_demo", False)
+            if not use_demo and context.configuration:
+                try:
+                    config_dict = context.configuration.model_dump(exclude_none=True) if hasattr(context.configuration, 'model_dump') else context.configuration.dict(exclude_none=True)
+                    use_demo = config_dict.get("use_demo", False)
+                except (AttributeError, TypeError):
+                    pass
+            
             execution_result = await self.task_executor.execute_tasks(
                 tasks=tasks,
                 root_task_id=None,  # Use actual root task ID from created task tree
                 use_streaming=False,
                 require_existing_tasks=require_existing_tasks,  # Allow override via metadata
+                use_demo=use_demo,
                 db_session=db_session
             )
 
@@ -419,12 +436,29 @@ async def _execute_streaming_mode(
             # Behavior controlled by global configuration (get_require_existing_tasks())
             # Default: require_existing_tasks=False (auto-create for convenience)
             # root_task_id=None means use the actual root task ID from the created task tree
+            # Extract use_demo from metadata, message.metadata, or configuration
+            use_demo = False
+            if context.metadata:
+                use_demo = context.metadata.get("use_demo", False)
+            # Also check message.metadata if available (A2A protocol supports metadata in message)
+            if not use_demo and context.message and hasattr(context.message, "metadata"):
+                message_metadata = context.message.metadata
+                if isinstance(message_metadata, dict):
+                    use_demo = message_metadata.get("use_demo", False)
+            if not use_demo and context.configuration:
+                try:
+                    config_dict = context.configuration.model_dump(exclude_none=True) if hasattr(context.configuration, 'model_dump') else context.configuration.dict(exclude_none=True)
+                    use_demo = config_dict.get("use_demo", False)
+                except (AttributeError, TypeError):
+                    pass
+            
             execution_result = await self.task_executor.execute_tasks(
                 tasks=tasks,
                 root_task_id=None,  # Use actual root task ID from created task tree
                 use_streaming=True,
                 streaming_callbacks_context=event_queue_bridge,
                 require_existing_tasks=None,  # Use global configuration (default: False, auto-create)
+                use_demo=use_demo,
                 db_session=db_session
             )
 

src/aipartnerupflow/api/a2a/custom_starlette_app.py

@@ -164,6 +164,8 @@ def __init__(
         enable_system_routes: bool = True,
         enable_docs: bool = True,
         task_model_class: Optional[Type[TaskModel]] = None,
+        task_routes_class: Optional[Type[TaskRoutes]] = None,
+        custom_routes: Optional[List[Route]] = None,
         **kwargs
     ):
         """
@@ -194,6 +196,12 @@ def __init__(
                              Users can pass their custom TaskModel subclass that inherits TaskModel
                              to add custom fields (e.g., project_id, department, etc.).
                              If None, default TaskModel will be used.
+            task_routes_class: Optional custom TaskRoutes class to use instead of default TaskRoutes.
+                             Allows extending TaskRoutes functionality without monkey patching.
+                             If None, default TaskRoutes will be used.
+            custom_routes: Optional list of custom Starlette Route objects to add to the application.
+                          Routes are merged after default routes (custom routes can override defaults if needed).
+                          If None, no custom routes are added.
             **kwargs: Keyword arguments for A2AStarletteApplication
         """
         super().__init__(*args, **kwargs)
@@ -211,8 +219,14 @@ def __init__(
         # Store task_model_class for task management APIs
         self.task_model_class = task_model_class or TaskModel
 
+        # Store custom routes
+        self.custom_routes = custom_routes or []
+        
+        # Use provided task_routes_class or default TaskRoutes
+        task_routes_cls = task_routes_class or TaskRoutes
+        
         # Initialize protocol-agnostic route handlers
-        self.task_routes = TaskRoutes(
+        self.task_routes = task_routes_cls(
             task_model_class=self.task_model_class,
             verify_token_func=self.verify_token_func,
             verify_permission_func=self.verify_permission_func
@@ -353,6 +367,10 @@ def routes(
                 ),
             ])
 
+        # Add user-provided custom routes
+        if self.custom_routes:
+            custom_routes.extend(self.custom_routes)
+        
         # Combine standard routes with custom routes
         return app_routes + custom_routes
 

src/aipartnerupflow/api/a2a/server.py

@@ -3,7 +3,7 @@
 """
 
 import httpx
-from typing import Optional
+from typing import Optional, Type
 from a2a.server.request_handlers import DefaultRequestHandler
 from a2a.server.tasks import (
     InMemoryTaskStore,
@@ -207,7 +207,11 @@ def load_skills() -> list[AgentSkill]:
     config_store=push_config_store
 )
 
-def _create_request_handler(verify_token_func=None, verify_permission_func=None):
+def _create_request_handler(
+    verify_token_func=None,
+    verify_permission_func=None,
+    task_routes_class: Optional[Type[TaskRoutes]] = None,
+):
     """
     Create request handler with agent executor
     
@@ -217,12 +221,16 @@ def _create_request_handler(verify_token_func=None, verify_permission_func=None)
     Args:
         verify_token_func: Optional JWT verification function
         verify_permission_func: Optional permission verification function
+        task_routes_class: Optional custom TaskRoutes class (default: TaskRoutes)
     """
     # Get task_model_class from registry
     task_model_class = get_task_model_class()
 
+    # Use provided task_routes_class or default TaskRoutes
+    task_routes_cls = task_routes_class or TaskRoutes
+    
     # Create TaskRoutes instance for the adapter
-    task_routes = TaskRoutes(
+    task_routes = task_routes_cls(
         task_model_class=task_model_class,
         verify_token_func=verify_token_func,
         verify_permission_func=verify_permission_func
@@ -276,6 +284,7 @@ def create_a2a_server(
     base_url: Optional[str] = None,
     enable_system_routes: bool = True,
     enable_docs: bool = True,
+    task_routes_class: Optional[Type[TaskRoutes]] = None,
 ) -> CustomA2AStarletteApplication:
     """
     Create A2A server instance with configuration
@@ -311,6 +320,9 @@ async def my_post_hook(task, inputs, result):
         enable_system_routes: Whether to enable system routes like /system (default: True)
         enable_docs: Whether to enable interactive API documentation at /docs (default: True).
                     Only available when API server is running, not when used as a library.
+        task_routes_class: Optional custom TaskRoutes class to use instead of default TaskRoutes.
+                         Allows extending TaskRoutes functionality without monkey patching.
+                         Example: task_routes_class=MyCustomTaskRoutes
     
     Returns:
         CustomA2AStarletteApplication instance
@@ -328,7 +340,8 @@ async def my_post_hook(task, inputs, result):
     # Permission checking will be handled at the middleware level
     request_handler = _create_request_handler(
         verify_token_func=verify_token_func,
-        verify_permission_func=None
+        verify_permission_func=None,
+        task_routes_class=task_routes_class,
     )
 
     # Create agent card
@@ -361,5 +374,6 @@ def verify_token_func_callback(token: str) -> Optional[dict]:
         enable_system_routes=enable_system_routes,
         enable_docs=enable_docs,
         task_model_class=final_task_model_class,
+        task_routes_class=task_routes_class,
     )
 

src/aipartnerupflow/api/routes/base.py

@@ -65,6 +65,39 @@ def _get_user_info(self, request: Request) -> Tuple[Optional[str], Optional[list
                 roles = [roles]
         return user_id, roles
 
+    def _extract_user_id_from_request(self, request: Request) -> Optional[str]:
+        """
+        Extract user_id from request JWT token
+        
+        This method extracts user_id from JWT token payload.
+        Only JWT tokens are trusted for security reasons - HTTP headers
+        can be easily spoofed by clients and should not be used for
+        authentication or user identification.
+        
+        Args:
+            request: Starlette request object
+            
+        Returns:
+            User ID string from JWT token payload, or None if not found
+        """
+        # Only extract from JWT token (if verify_token_func is available)
+        if self.verify_token_func:
+            auth_header = request.headers.get("Authorization")
+            if auth_header and auth_header.startswith("Bearer "):
+                token = auth_header.split(" ", 1)[1]
+                try:
+                    payload = self.verify_token_func(token)
+                    if payload and "user_id" in payload:
+                        return payload["user_id"]
+                    # Also try "sub" field (standard JWT claim)
+                    if payload and "sub" in payload:
+                        return payload["sub"]
+                except Exception:
+                    # Token verification failed
+                    pass
+        
+        return None
+    
     def _check_permission(
         self,
         request: Request,