⚙️ FEATURE-#201: Separate Engine from Execution Strategy

docs/nav/concepts/concept-task-engine.md

@@ -0,0 +1,107 @@
+# Task Engine
+
+The `TaskEngine` manages the **lifecycle** of a single task — status transitions, duration tracking, retry, timeout, backoff, error handling, and tracer integration. Execution strategies (`Sequential`, `Parallel`, etc.) are responsible only for **ordering and parallelism**.
+
+## Architecture
+
+```mermaid
+sequenceDiagram
+    participant S as Strategy
+    participant E as TaskEngine
+    participant A as @action
+
+    S->>E: TaskEngine(task, workflow_id, previous_context)
+    S->>E: with engine.start()
+    E->>E: status = IN_PROGRESS
+    E->>E: tracer.start_task()
+    S->>E: engine.execute_with_retry()
+
+    loop retry loop (max_attempts)
+        alt timeout > 0
+            E->>E: _execute_with_timeout()
+        else no timeout
+            E->>E: _execute_single()
+        end
+        E->>A: task.step()
+        alt success
+            A-->>E: Context
+            E-->>S: result
+        else failure & attempts remaining
+            E->>E: status = RETRY
+            E->>E: sleep(delay)
+        end
+    end
+
+    E->>E: status = COMPLETED / FAILED
+    E->>E: tracer.end_task()
+    S->>S: task.callback()
+    S->>S: _flow_callback()
+```
+
+## How it works
+
+The engine uses a **context manager** pattern to separate lifecycle from execution:
+
+```python
+engine = TaskEngine(task=task, workflow_id=workflow_id, previous_context=previous_context)
+
+with engine.start():
+    engine.execute_with_retry()
+```
+
+### `start()` — lifecycle context manager
+
+Manages everything that happens **around** the execution:
+
+```mermaid
+stateDiagram-v2
+    [*] --> IN_PROGRESS: start()
+    IN_PROGRESS --> COMPLETED: success
+    IN_PROGRESS --> RETRY: retry attempt
+    RETRY --> COMPLETED: success after retry
+    RETRY --> FAILED: max attempts reached
+    IN_PROGRESS --> FAILED: exception
+    COMPLETED --> [*]: end_task tracer
+    FAILED --> [*]: end_task tracer
+```
+
+1. Sets `status = IN_PROGRESS` and starts the timer
+2. Starts the tracer span
+3. **Yields** — the execution block runs here
+4. On success: sets `duration` and `status = COMPLETED`
+5. On error: sets `errors` and `status = FAILED`
+6. Always: ends the tracer span
+
+### `execute_with_retry()` — retry, timeout, and backoff
+
+Reads `retry`, `timeout`, `retry_delay`, and `backoff` from the `@action` decorator and manages the full retry loop:
+
+```mermaid
+flowchart TD
+    A["execute_with_retry()"] --> B{"timeout > 0?"}
+    B -->|yes| C["_execute_with_timeout()"]
+    B -->|no| D["_execute_single()"]
+    C --> E{"success?"}
+    D --> E
+    E -->|yes| F["return result"]
+    E -->|no| G{"attempt < max?"}
+    G -->|yes| H["status = RETRY\nsleep(delay)"]
+    H -->|"backoff?"| I["delay *= 2"]
+    I --> B
+    H -->|no backoff| B
+    G -->|no| J["raise exception"]
+```
+
+- If `timeout > 0`: uses `ThreadPoolExecutor` with a real deadline
+- If execution fails and `attempt < max_attempts`: sets `status = RETRY`, waits, and retries
+- If `backoff = True`: doubles the delay after each failed attempt
+
+### `execute()` — single execution
+
+Calls the task function once without retry. Used internally by `execute_with_retry()` and available for cases where retry is not needed.
+
+## References
+
+- [Task lifecycle and status](concept-task-lifecycle.md)
+- [`@action` decorator](../reference/action.md)
+- [`TypeStatus`](../reference/type-status.md)

docs/nav/concepts/concept-task-lifecycle.md

@@ -11,7 +11,7 @@ Every task moves through **status** values tracked by Dotflow. They describe whe
 
 ## Retries and pauses
 
-- **`RETRY`** — A retry is scheduled (for example after backoff); thread-safe retry behavior is part of the action runner.
+- **`RETRY`** — A retry is scheduled (for example after backoff); retry behavior is managed by the `TaskEngine`.
 - **`PAUSED`** — Execution is held (depending on workflow configuration and provider behavior).
 
 Retry policy is configured per task (timeouts, backoff, etc.); see [Task retry](../tutorial/task-retry.md) and [Task backoff](../tutorial/task-backoff.md).

docs/nav/tutorial/tracer-opentelemetry.md

@@ -61,8 +61,8 @@ Run the example and open [http://localhost:16686](http://localhost:16686) to see
 | ABC method | When | Span action |
 |------------|------|-------------|
 | `start_workflow` | Manager.__init__ | Creates parent span |
-| `start_task` | Execution.__init__ | Creates child span |
-| `end_task` | Execution finally | Sets attributes, status, ends span |
+| `start_task` | TaskEngine.start() | Creates child span |
+| `end_task` | TaskEngine.start() finally | Sets attributes, status, ends span |
 | `end_workflow` | _callback_workflow | Sets workflow status, ends parent span |
 
 ## Compatible backends

docs_src/checkpoint/checkpoint.py

@@ -1,3 +1,5 @@
+from uuid import UUID
+
 from dotflow import Config, DotFlow, action
 from dotflow.providers import StorageFile
 
@@ -21,7 +23,7 @@ def step_three(previous_context):
 
 
 def main():
-    workflow = DotFlow(config=config, workflow_id="my-etl-pipeline")
+    workflow = DotFlow(config=config, workflow_id=UUID("12345678-1234-5678-1234-567812345678"))
 
     workflow.task.add(step=step_one)
     workflow.task.add(step=step_two)

dotflow/abc/flow.py

@@ -3,7 +3,9 @@
 from abc import ABC, abstractmethod
 from uuid import UUID
 
+from dotflow.core.context import Context
 from dotflow.core.task import Task
+from dotflow.core.types import TypeStatus
 
 
 class Flow(ABC):
@@ -40,3 +42,24 @@ def _flow_callback(self, task: Task) -> None:
     @abstractmethod
     def run(self) -> None:
         return None
+
+    def _has_checkpoint(self, task: Task) -> bool:
+        if not self.resume:
+            return False
+
+        context = task.config.storage.get(
+            key=task.config.storage.key(task=task)
+        )
+
+        return context.storage is not None
+
+    def _restore_checkpoint(self, task: Task) -> Context:
+        previous_context = task.config.storage.get(
+            key=task.config.storage.key(task=task)
+        )
+
+        task.status = TypeStatus.COMPLETED
+        task.current_context = previous_context
+        self._flow_callback(task=task)
+
+        return previous_context

dotflow/core/action.py

@@ -2,13 +2,10 @@
 
 import asyncio
 from collections.abc import Callable
-from concurrent.futures import ThreadPoolExecutor
-from time import sleep
 from types import FunctionType
 
 from dotflow.core.context import Context
-from dotflow.core.exception import ExecutionWithClassError, TaskError
-from dotflow.core.types.status import TypeStatus
+from dotflow.core.exception import ExecutionWithClassError
 
 
 def is_execution_with_class_internal_error(error: Exception) -> bool:
@@ -103,13 +100,13 @@ def __call__(self, *args, **kwargs):
 
             if contexts:
                 return Context(
-                    storage=self._run_action(*args, task=task, **contexts),
+                    storage=self._run_action(*args, **contexts),
                     task_id=task.task_id,
                     workflow_id=task.workflow_id,
                 )
 
             return Context(
-                storage=self._run_action(*args, task=task),
+                storage=self._run_action(*args),
                 task_id=task.task_id,
                 workflow_id=task.workflow_id,
             )
@@ -123,77 +120,33 @@ def action(*_args, **_kwargs):
 
             if contexts:
                 return Context(
-                    storage=self._run_action(*_args, task=task, **contexts),
+                    storage=self._run_action(*_args, **contexts),
                     task_id=task.task_id,
                     workflow_id=task.workflow_id,
                 )
 
             return Context(
-                storage=self._run_action(*_args, task=task),
+                storage=self._run_action(*_args),
                 task_id=task.task_id,
                 workflow_id=task.workflow_id,
             )
 
-        return action
+        action.retry = self.retry
+        action.timeout = self.timeout
+        action.retry_delay = self.retry_delay
+        action.backoff = self.backoff
 
-    def _run_action(self, *args, task=None, **kwargs):
-        current_delay = self.retry_delay
+        return action
 
+    def _run_action(self, *args, **kwargs):
         is_async = asyncio.iscoroutinefunction(self.func)
 
-        max_attempts = max(1, self.retry)
-
-        for attempt in range(1, max_attempts + 1):
-            try:
-                if self.timeout:
-                    executor = ThreadPoolExecutor(max_workers=1)
-                    try:
-                        future = executor.submit(
-                            self._call_func,
-                            is_async,
-                            *args,
-                            **kwargs,
-                        )
-                        result = future.result(timeout=self.timeout)
-                    except TimeoutError:
-                        future.cancel()
-                        executor.shutdown(wait=False, cancel_futures=True)
-                        raise
-                    except Exception:
-                        executor.shutdown(wait=False)
-                        raise
-                    else:
-                        executor.shutdown(wait=False)
-                else:
-                    result = self._call_func(is_async, *args, **kwargs)
-
-                return result
-
-            except TimeoutError:
-                raise
-
-            except Exception as error:
-                last_exception = error
-
-                if is_execution_with_class_internal_error(
-                    error=last_exception
-                ):
-                    raise ExecutionWithClassError() from None
-
-                if attempt == max_attempts:
-                    raise last_exception from None
-
-                if task is not None:
-                    task.retry_count += 1
-                    task.errors = TaskError(
-                        error=error,
-                        attempt=attempt,
-                    )
-                    task.status = TypeStatus.RETRY
-
-                sleep(current_delay)
-                if self.backoff:
-                    current_delay *= 2
+        try:
+            return self._call_func(is_async, *args, **kwargs)
+        except Exception as error:
+            if is_execution_with_class_internal_error(error=error):
+                raise ExecutionWithClassError() from None
+            raise
 
     def _call_func(self, is_async, *args, **kwargs):
         if is_async: