add message arg on task run

Author: lazToum

docs/openapi.json

@@ -23,6 +23,7 @@ async def trigger_run_task(
     storage: Storage,
     env_vars: dict[str, str],
     skip_deps: bool | None = None,
+    message: str | None = None,
 ) -> None:
     """Trigger a task.
 
@@ -38,6 +39,9 @@ async def trigger_run_task(
         The environment variables for the task
     skip_deps : bool, Optional
         Whether to skip installing dependencies before the task.
+    message : str, Optional
+        Optional initial message to pass to the task.
+
     Raises
     ------
     RuntimeError
@@ -55,6 +59,7 @@ async def trigger_run_task(
                 storage=storage,
                 redis_manager=app_state.redis,
                 skip_deps=skip_deps,
+                message=message,
             )
         )
         bg_task.add_done_callback(
@@ -66,7 +71,10 @@ async def trigger_run_task(
         )
     else:
         await run_task_job.kiq(
-            task=task, env_vars=env_vars, skip_deps=skip_deps
+            task=task,
+            env_vars=env_vars,
+            skip_deps=skip_deps,
+            message=message,
         )
 
 
@@ -120,6 +128,7 @@ async def schedule_task(
     storage: Storage,
     env_vars: dict[str, str],
     skip_deps: bool | None = None,
+    message: str | None = None,
 ) -> None:
     """Schedule a task.
 
@@ -135,6 +144,8 @@ async def schedule_task(
         The environment variables for the task.
     skip_deps : bool, Optional
         Whether to skip installing dependencies before the task.
+    message : str
+        Optional initial message to pass to the task.
 
     Raises
     ------

waldiez_runner/routes/v1/task_jobs.py

@@ -202,6 +202,7 @@ async def create_task(
     ),
     input_timeout: int = Form(180),
     skip_deps: bool | None = Form(None),
+    message: str | None = Form(None),
     schedule_type: Literal["once", "cron"] | None = Form(None),
     scheduled_time: datetime | None = Form(None),
     cron_expression: str | None = Form(None),
@@ -234,6 +235,8 @@ async def create_task(
         The timeout for input requests, by default 180
     skip_deps : bool, optional
         Whether to skip installing dependencies before the task.
+    message : str, optional
+        Optional initial message to pass to the task.
     schedule_type : Optional[Literal["once", "cron"]], optional
         The type of schedule, by default None
     scheduled_time : Optional[datetime], optional
@@ -337,6 +340,7 @@ async def create_task(
             db_manager=db_manager,
             storage=storage,
             skip_deps=skip_deps,
+            message=message,
         )
     if task.schedule_type is not None:
         await schedule_task(
@@ -345,6 +349,7 @@ async def create_task(
             storage=storage,
             env_vars=environment_vars,
             skip_deps=skip_deps,
+            message=message,
         )
     return task_response
 

waldiez_runner/routes/v1/task_router.py

@@ -1,7 +1,9 @@
 # SPDX-License-Identifier: Apache-2.0.
 # Copyright (c) 2024 - 2026 Waldiez and contributors.
 
-# pylint: disable=broad-exception-caught, unused-argument
+# pylint: disable=broad-exception-caught,unused-argument
+# pylint: disable=too-many-arguments,too-many-positional-arguments
+
 """Handle running the task in a virtual environment."""
 
 import asyncio
@@ -42,6 +44,7 @@ async def execute_task(
     debug: bool,
     max_duration: int,
     skip_deps: bool,
+    message: str,
 ) -> tuple[TaskStatus, dict[str, Any] | list[dict[str, Any]] | None]:
     """Execute the task in a virtual environment.
 
@@ -69,6 +72,8 @@ async def execute_task(
         The task's max duration.
     skip_deps : bool
         Whether to skip installing dependencies before the task.
+    message : str
+        Optional initial message to pass to the task.
 
     Returns
     -------
@@ -90,6 +95,7 @@ async def execute_task(
             debug=debug,
             max_duration=max_duration,
             skip_deps=skip_deps,
+            message=message,
         )
         LOG.info("Task %s exited with code %s", task.id, exit_code)
         return interpret_exit_code(exit_code)
@@ -266,6 +272,7 @@ async def run_app_in_venv(
     debug: bool,
     max_duration: int,
     skip_deps: bool,
+    message: str,
 ) -> int:
     """Run the app in the venv.
 
@@ -295,6 +302,8 @@ async def run_app_in_venv(
         The task's max duration.
     skip_deps : bool
         Whether to skip installing deps before the task.
+    message : str
+        Optional initial message to pass to the task.
 
     Returns
     -------
@@ -317,6 +326,8 @@ async def run_app_in_venv(
         skip_arg,
         str(file_path),
     ]
+    if message:
+        args.extend(["--message", message])
     if debug:
         args.append("--debug")
 

waldiez_runner/tasks/runner.py

@@ -35,6 +35,7 @@ async def run_task(
     task: TaskResponse,
     env_vars: dict[str, str],
     skip_deps: bool | None = None,
+    message: str | None = None,
     db_manager: DatabaseManager = TaskiqDepends(get_db_manager),
     storage: Storage = TaskiqDepends(get_storage),
     redis_manager: RedisManager = TaskiqDepends(get_redis_manager),
@@ -49,6 +50,8 @@ async def run_task(
         Environment variables for the task.
     skip_deps : bool, Optional
         Whether to skip installing dependencies before the task.
+    message : str
+        Optional initial message to pass to the task.
     db_manager : DatabaseManager
         Database session manager dependency.
     storage : Storage
@@ -99,6 +102,7 @@ async def run_task(
             debug=debug,
             max_duration=settings.max_task_duration,
             skip_deps=skip_deps is True,
+            message=message or "",
         )
         LOG.info("Task %s finished with status %s", task.id, status.value)
         LOG.debug("Task %s finished with results %s", task.id, results)