feat: add worker signal to skip task result update

Author: chrishagglund-ship-it

src/conductor/client/automator/async_task_runner.py

@@ -8,7 +8,7 @@
 
 from conductor.client.configuration.configuration import Configuration
 from conductor.client.configuration.settings.metrics_settings import MetricsSettings
-from conductor.client.context.task_context import _set_task_context, _clear_task_context, TaskInProgress
+from conductor.client.context.task_context import _set_task_context, _clear_task_context, TaskInProgress, LEAVE_TASK_UNRESOLVED
 from conductor.client.event.task_runner_events import (
     TaskRunnerEvent, PollStarted, PollCompleted, PollFailure,
     TaskExecutionStarted, TaskExecutionCompleted, TaskExecutionFailure,
@@ -619,6 +619,14 @@ async def __async_execute_and_update_task(self, task: Task) -> None:
                     task_result = await self.__async_execute_task(task)
                     if task_result is None:
                         return
+                    # Worker returned LEAVE_TASK_UNRESOLVED: send no update and
+                    # stop extending the lease (finally untracks) so the server
+                    # resolves the task itself (e.g. responseTimeout).
+                    if task_result is LEAVE_TASK_UNRESOLVED:
+                        logger.debug(
+                            "Async task %s left unresolved by worker; not updating server",
+                            task.task_id)
+                        return
                     self._untrack_lease(task.task_id)
                     # Update task and get next task from v2 response
                     task = await self.__async_update_task(task_result)
@@ -693,6 +701,15 @@ async def __async_execute_task(self, task: Task) -> TaskResult:
             # Direct await of async worker function - NO THREADS!
             task_output = await self.worker.execute_function(**task_input)
 
+            # Worker deliberately left the task unresolved: report nothing to
+            # the server (see LEAVE_TASK_UNRESOLVED docs). The finally block
+            # clears the task context; the caller stops managing the task.
+            if task_output is LEAVE_TASK_UNRESOLVED:
+                logger.debug(
+                    "Async task %s left unresolved by worker; not updating server",
+                    task.task_id)
+                return LEAVE_TASK_UNRESOLVED
+
             # Handle different return types (same as TaskRunner:441-474)
             if isinstance(task_output, TaskResult):
                 # Already a TaskResult - use as-is

src/conductor/client/automator/task_runner.py

@@ -11,7 +11,7 @@
 
 from conductor.client.configuration.configuration import Configuration
 from conductor.client.configuration.settings.metrics_settings import MetricsSettings
-from conductor.client.context.task_context import _set_task_context, _clear_task_context, TaskInProgress
+from conductor.client.context.task_context import _set_task_context, _clear_task_context, TaskInProgress, LEAVE_TASK_UNRESOLVED
 from conductor.client.event.task_runner_events import (
     TaskRunnerEvent, PollStarted, PollCompleted, PollFailure,
     TaskExecutionStarted, TaskExecutionCompleted, TaskExecutionFailure,
@@ -568,6 +568,14 @@ def __execute_and_update_task(self, task: Task) -> None:
                     logger.debug("Task %s is running async, will update when complete", task.task_id)
                     async_running = True
                     return
+                # Worker returned LEAVE_TASK_UNRESOLVED: send no update and stop
+                # extending the lease so the server can resolve the task itself
+                # (e.g. responseTimeout). The finally block untracks the lease.
+                if task_result is LEAVE_TASK_UNRESOLVED:
+                    logger.debug(
+                        "Task %s left unresolved by worker; not updating server",
+                        task.task_id)
+                    return
                 self._untrack_lease(task.task_id)
                 # Update task and get next task from v2 response
                 task = self.__update_task(task_result)
@@ -790,6 +798,13 @@ def __execute_task(self, task: Task) -> TaskResult:
                 _clear_task_context()
                 return None
 
+            # Worker deliberately left the task unresolved: report nothing to
+            # the server and let the caller stop managing the task (see
+            # LEAVE_TASK_UNRESOLVED docs).
+            if task_output is LEAVE_TASK_UNRESOLVED:
+                _clear_task_context()
+                return LEAVE_TASK_UNRESOLVED
+
             # Handle different return types
             if isinstance(task_output, TaskResult):
                 # Already a TaskResult - use as-is

src/conductor/client/context/__init__.py

@@ -26,10 +26,12 @@ def process_video(video_id: str) -> Union[GeneratedVideo, TaskInProgress]:
     TaskContext,
     get_task_context,
     TaskInProgress,
+    LEAVE_TASK_UNRESOLVED,
 )
 
 __all__ = [
     'TaskContext',
     'get_task_context',
     'TaskInProgress',
+    'LEAVE_TASK_UNRESOLVED',
 ]

src/conductor/client/context/task_context.py

@@ -96,6 +96,51 @@ def __repr__(self) -> str:
         return f"TaskInProgress(callback_after={self.callback_after_seconds}s, output={self.output})"
 
 
+class _LeaveTaskUnresolved:
+    """Sentinel return value telling the SDK to report NO result for the task.
+
+    When a worker returns ``LEAVE_TASK_UNRESOLVED`` the task runner sends no
+    update to the server (no COMPLETED / FAILED / IN_PROGRESS) and stops
+    extending the lease. The task is left exactly as the server already has it
+    (IN_PROGRESS with its existing lease), so the *server* decides its fate —
+    e.g. it TIMES_OUT after ``responseTimeoutSeconds`` if nothing updates it.
+
+    This is distinct from:
+    - a normal return value  -> task COMPLETED
+    - raising an exception    -> task FAILED
+    - ``TaskInProgress``      -> task reported IN_PROGRESS with a callback
+      (which resets the response-timeout clock)
+
+    It is fully backward compatible: workers that never return this behave
+    exactly as before. Useful when the task's resolution is genuinely
+    out-of-band (handed to another process/system that will update it later),
+    or in tests that need the server — not the worker — to resolve a task.
+
+    Usage:
+        from conductor.client.context import LEAVE_TASK_UNRESOLVED
+
+        @worker_task(task_definition_name='handoff_task')
+        def handoff(job_id: str):
+            enqueue_for_external_system(job_id)
+            # Don't complete/fail here; something else will update the task.
+            return LEAVE_TASK_UNRESOLVED
+    """
+
+    _instance = None
+
+    def __new__(cls):
+        if cls._instance is None:
+            cls._instance = super().__new__(cls)
+        return cls._instance
+
+    def __repr__(self) -> str:
+        return "LEAVE_TASK_UNRESOLVED"
+
+
+# Singleton sentinel; compare with `is`.
+LEAVE_TASK_UNRESOLVED = _LeaveTaskUnresolved()
+
+
 # Context variable for storing TaskContext (thread-safe and async-safe)
 _task_context_var: ContextVar[Optional['TaskContext']] = ContextVar('task_context', default=None)
 

src/conductor/client/worker/worker.py

@@ -514,7 +514,12 @@ def execute(self, task: Task) -> TaskResult:
                 task_output.workflow_instance_id = task.workflow_instance_id
                 return task_output
             # Import here to avoid circular dependency
-            from conductor.client.context.task_context import TaskInProgress
+            from conductor.client.context.task_context import TaskInProgress, LEAVE_TASK_UNRESOLVED
+            if task_output is LEAVE_TASK_UNRESOLVED:
+                # Worker is deliberately not reporting a result. Hand the
+                # sentinel back so the TaskRunner skips the server update and
+                # leaves the task as-is (see LEAVE_TASK_UNRESOLVED docs).
+                return task_output
             if isinstance(task_output, TaskInProgress):
                 # Return TaskInProgress as-is for TaskRunner to handle
                 return task_output
@@ -600,6 +605,16 @@ def check_completed_async_tasks(self) -> list:
                     # Get result (won't block since future is done)
                     task_output = future.result(timeout=0)
 
+                    # Worker deliberately left the task unresolved: drop it
+                    # from the pending set and report nothing to the server.
+                    from conductor.client.context.task_context import LEAVE_TASK_UNRESOLVED
+                    if task_output is LEAVE_TASK_UNRESOLVED:
+                        logger.debug(
+                            "Async task %s (%s) returned LEAVE_TASK_UNRESOLVED; "
+                            "not reporting a result", task_id, task.task_def_name)
+                        tasks_to_remove.append(task_id)
+                        continue
+
                     # Process result same as sync execution
                     if isinstance(task_output, TaskResult):
                         task_output.task_id = task.task_id

tests/integration/test_async_lease_extension.py

@@ -33,6 +33,7 @@
 
 from conductor.client.automator.task_handler import TaskHandler, get_registered_workers
 from conductor.client.configuration.configuration import Configuration
+from conductor.client.context.task_context import LEAVE_TASK_UNRESOLVED
 from conductor.client.worker.worker_task import worker_task
 from conductor.client.http.models.workflow_def import WorkflowDef
 from conductor.client.http.models.task_def import TaskDef
@@ -74,9 +75,10 @@ def _retry_on_transient(func, *args, retries=4, base_delay=1, **kwargs):
 # Short response timeout — task must heartbeat to stay alive
 RESPONSE_TIMEOUT_SECONDS = 10
 
-# Task sleeps longer than the response timeout to prove heartbeat works.
-# Must be long enough that the server's workflow sweeper catches the expired
-# task BEFORE the worker completes.
+# The heartbeat worker sleeps this long (well beyond responseTimeout) to prove
+# that lease extension keeps a genuinely long-running task alive. The
+# no-heartbeat worker no longer sleeps — it returns LEAVE_TASK_UNRESOLVED so the
+# server times the task out deterministically (see async_lease_no_heartbeat_task).
 TASK_SLEEP_SECONDS = 50
 
 # Short task duration for performance test — well within timeout
@@ -131,13 +133,20 @@ async def async_lease_heartbeat_task(job_id: str) -> dict:
     ),
     overwrite_task_def=True,
 )
-async def async_lease_no_heartbeat_task(job_id: str) -> dict:
-    """Async long-running task without heartbeat — should time out."""
-    logger.info("[async_no_heartbeat] Starting job %s, sleeping %ss (timeout=%ss)",
-                job_id, TASK_SLEEP_SECONDS, RESPONSE_TIMEOUT_SECONDS)
-    await asyncio.sleep(TASK_SLEEP_SECONDS)
-    logger.info("[async_no_heartbeat] Completed job %s", job_id)
-    return {'job_id': job_id, 'status': 'completed', 'slept': TASK_SLEEP_SECONDS}
+async def async_lease_no_heartbeat_task(job_id: str):
+    """Async task without heartbeat that deliberately never resolves itself.
+
+    Returning LEAVE_TASK_UNRESOLVED makes the SDK lease the task but report no
+    result and stop extending the lease. With no heartbeat, the lease expires
+    after responseTimeoutSeconds and the server TIMES_OUT the task — with no
+    race against a worker completion. (The old version slept 50s then completed,
+    so a slow server sweeper could let the completion win, producing the flaky
+    "expected TIMED_OUT but got COMPLETED" failures.)
+    """
+    logger.info("[async_no_heartbeat] Picked up job %s; leaving it unresolved so "
+                "the server times it out (responseTimeout=%ss)",
+                job_id, RESPONSE_TIMEOUT_SECONDS)
+    return LEAVE_TASK_UNRESOLVED
 
 
 @worker_task(
@@ -357,10 +366,11 @@ def test_01_async_with_heartbeat_completes(self):
         print("\n  PASS: Async task completed with heartbeat keeping lease alive")
 
     def test_02_async_without_heartbeat_times_out(self):
-        """Async task WITHOUT lease_extend_enabled times out when sleep > responseTimeout."""
+        """Async task WITHOUT lease_extend_enabled times out when it isn't resolved."""
         print("\n" + "=" * 80)
         print("TEST: Async without heartbeat — task should TIME OUT")
-        print(f"  responseTimeoutSeconds={RESPONSE_TIMEOUT_SECONDS}s, task sleeps {TASK_SLEEP_SECONDS}s")
+        print(f"  responseTimeoutSeconds={RESPONSE_TIMEOUT_SECONDS}s, "
+              f"worker leaves task unresolved (server must time it out)")
         print("=" * 80)
 
         wf_name = f'test_async_lease_no_heartbeat_{RUN_ID}'

tests/integration/test_lease_extension.py

@@ -31,6 +31,7 @@
 
 from conductor.client.automator.task_handler import TaskHandler
 from conductor.client.configuration.configuration import Configuration
+from conductor.client.context.task_context import LEAVE_TASK_UNRESOLVED
 from conductor.client.worker.worker_task import worker_task
 from conductor.client.http.models.workflow_def import WorkflowDef
 from conductor.client.http.models.task_def import TaskDef
@@ -72,9 +73,10 @@ def _retry_on_transient(func, *args, retries=4, base_delay=1, **kwargs):
 # Short response timeout — task must heartbeat to stay alive
 RESPONSE_TIMEOUT_SECONDS = 10
 
-# Task sleeps longer than the response timeout to prove heartbeat works.
-# Must be long enough that the server's workflow sweeper (which runs every
-# ~30s) catches the expired task BEFORE the worker completes.
+# The heartbeat worker sleeps this long (well beyond responseTimeout) to prove
+# that lease extension keeps a genuinely long-running task alive. The
+# no-heartbeat worker no longer sleeps — it returns LEAVE_TASK_UNRESOLVED so the
+# server times the task out deterministically (see lease_no_heartbeat_task).
 TASK_SLEEP_SECONDS = 50
 
 # Per-run suffix so this suite's task/workflow names don't collide with other
@@ -123,13 +125,20 @@ def lease_heartbeat_task(job_id: str) -> dict:
     ),
     overwrite_task_def=True,
 )
-def lease_no_heartbeat_task(job_id: str) -> dict:
-    """Long-running task without heartbeat — should time out."""
-    logger.info("[no_heartbeat_task] Starting job %s, sleeping %ss (timeout=%ss)",
-                job_id, TASK_SLEEP_SECONDS, RESPONSE_TIMEOUT_SECONDS)
-    time.sleep(TASK_SLEEP_SECONDS)
-    logger.info("[no_heartbeat_task] Completed job %s", job_id)
-    return {'job_id': job_id, 'status': 'completed', 'slept': TASK_SLEEP_SECONDS}
+def lease_no_heartbeat_task(job_id: str):
+    """Task without heartbeat that deliberately never resolves itself.
+
+    Returning LEAVE_TASK_UNRESOLVED makes the SDK lease the task but report no
+    result and stop extending the lease. With no heartbeat, the lease expires
+    after responseTimeoutSeconds and the server TIMES_OUT the task — with no
+    race against a worker completion. (The old version slept 50s then completed,
+    so a slow server sweeper could let the completion win, producing the flaky
+    "expected TIMED_OUT but got COMPLETED" failures.)
+    """
+    logger.info("[no_heartbeat_task] Picked up job %s; leaving it unresolved so "
+                "the server times it out (responseTimeout=%ss)",
+                job_id, RESPONSE_TIMEOUT_SECONDS)
+    return LEAVE_TASK_UNRESOLVED
 
 
 # -- Test class --------------------------------------------------------------
@@ -286,10 +295,11 @@ def test_01_with_heartbeat_completes(self):
             stop_workers()
 
     def test_02_without_heartbeat_times_out(self):
-        """Task WITHOUT lease_extend_enabled times out when sleep > responseTimeout."""
+        """Task WITHOUT lease_extend_enabled times out when it isn't resolved."""
         print("\n" + "=" * 80)
         print("TEST: Without heartbeat — task should TIME OUT")
-        print(f"  responseTimeoutSeconds={RESPONSE_TIMEOUT_SECONDS}s, task sleeps {TASK_SLEEP_SECONDS}s")
+        print(f"  responseTimeoutSeconds={RESPONSE_TIMEOUT_SECONDS}s, "
+              f"worker leaves task unresolved (server must time it out)")
         print("=" * 80)
 
         wf_name = f'test_lease_no_heartbeat_{RUN_ID}'