feat(flow): add restore_from_state_id kickoff parameter (#5674)

## Summary

- Reverts `b0e2fda` ("fix(flow): add execution_id separate from state.id", COR-48): removes `Flow.execution_id` and points `current_flow_id` / `current_flow_request_id` back at `flow_id` (i.e. `state.id`). The separate per-run tracking id was no longer the right abstraction once `restore_from_state_id` reshapes how `state.id` is assigned;

- Adds an optional `restore_from_state_id` kwarg to `Flow.kickoff` / `Flow.kickoff_async` that hydrates state from a previously-persisted flow's latest snapshot

- Reassigns `state.id` to a fresh value (or `inputs["id"]` if pinned) so the new run's `@persist` writes don't extend the source's history

- Existing `inputs["id"]` resume, `@persist`, and `from_checkpoint` paths are unchanged

## Problem
`@persist` only supports *resume* today: `kickoff(inputs={"id": <uuid>})` hydrates state and continues writing under the same `flow_uuid`. There's no way to **fork** — hydrate from a snapshot but persist under a separate key, leaving the source's history intact. This PR adds that.

| | `state.id` after kickoff | `@persist` writes land under |
|---|---|---|
| `inputs["id"]` (resume) | supplied id | supplied id (extends history) |
| `restore_from_state_id` (fork) | fresh id, or `inputs["id"]` if pinned | new id (source preserved) |

## Behavior

| `inputs.id` | `restore_from_state_id` | Effect |
|---|---|---|
| — | — | Fresh kickoff |
| set | — | Existing resume |
| — | UUID | Fork — new `state.id`, hydrated from source |
| set | UUID | Fork into a pinned `state.id`, hydrated from source |

- Source not found → silent fallback (mirrors existing resume)
- Both `from_checkpoint` and `restore_from_state_id` set → `ValueError`
- `restore_from_state_id=None` → byte-identical to current main

## Design
Fork hydration runs before the existing `inputs` block in `kickoff_async`. On a hit, it calls the same `_restore_state` primitive used by resume, then overwrites `state.id` with a fresh UUID (or `inputs["id"]`). A `fork_succeeded` flag gates the existing `inputs["id"]` path so we don't double-load. `_completed_methods` / `_is_execution_resuming` are intentionally untouched — skip-completed-methods remains the territory of `apply_checkpoint` and `from_pending`.

## Test plan
- [ ] `pytest tests/test_flow_persistence.py` — 5 new tests (four-row matrix, not-found fallback, default no-op, conflict raise) + 6 existing as regression
- [ ] `pytest tests/test_flow.py` — broader flow suite
- [ ] Manual end-to-end against an HITL `@persist` flow

lib/crewai/src/crewai/flow/flow.py

@@ -1074,7 +1074,6 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
     _human_feedback_method_outputs: dict[str, Any] = PrivateAttr(default_factory=dict)
     _input_history: list[InputHistoryEntry] = PrivateAttr(default_factory=list)
     _state: Any = PrivateAttr(default=None)
-    _execution_id: str = PrivateAttr(default_factory=lambda: str(uuid4()))
 
     def __class_getitem__(cls: type[Flow[T]], item: type[T]) -> type[Flow[T]]:  # type: ignore[override]
         class _FlowGeneric(cls):  # type: ignore[valid-type,misc]
@@ -1865,27 +1864,6 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
         except (AttributeError, TypeError):
             return ""  # Safely handle any unexpected attribute access issues
 
-    @property
-    def execution_id(self) -> str:
-        """Stable identifier for this flow execution.
-
-        Separate from ``flow_id`` / ``state.id``, which consumers may
-        override via ``kickoff(inputs={"id": ...})`` to resume a persisted
-        flow. ``execution_id`` is never affected by ``inputs`` and stays
-        stable for the lifetime of a single run, so it is the correct key
-        for telemetry, tracing, and any external correlation that must
-        uniquely identify a single execution even when callers pass an
-        ``id`` in ``inputs``.
-
-        Defaults to a fresh ``uuid4`` per ``Flow`` instance; assign to
-        override when an outer system already has an execution identity.
-        """
-        return self._execution_id
-
-    @execution_id.setter
-    def execution_id(self, value: str) -> None:
-        self._execution_id = value
-
     def _initialize_state(self, inputs: dict[str, Any]) -> None:
         """Initialize or update flow state with new inputs.
 
@@ -2054,6 +2032,7 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
         inputs: dict[str, Any] | None = None,
         input_files: dict[str, FileInput] | None = None,
         from_checkpoint: CheckpointConfig | None = None,
+        restore_from_state_id: str | None = None,
     ) -> Any | FlowStreamingOutput:
         """Start the flow execution in a synchronous context.
 
@@ -2065,10 +2044,24 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
             input_files: Optional dict of named file inputs for the flow.
             from_checkpoint: Optional checkpoint config. If ``restore_from``
                 is set, the flow resumes from that checkpoint.
+            restore_from_state_id: Optional UUID of a previously-persisted flow
+                whose latest snapshot should hydrate this run's state. The new
+                run is assigned a fresh ``state.id`` (or ``inputs["id"]`` if
+                pinned), so its ``@persist`` writes land under a separate
+                persistence key and the source flow's history is preserved.
+                If the referenced state is not found, the kickoff falls back
+                silently to baseline behavior. Cannot be combined with
+                ``from_checkpoint``; passing both raises ``ValueError``.
 
         Returns:
             The final output from the flow or FlowStreamingOutput if streaming.
         """
+        if from_checkpoint is not None and restore_from_state_id is not None:
+            raise ValueError(
+                "Cannot combine `from_checkpoint` and `restore_from_state_id`. "
+                "These parameters target different state systems "
+                "(Checkpointing and @persist) and cannot be used together."
+            )
         restored = apply_checkpoint(self, from_checkpoint)
         if restored is not None:
             return restored.kickoff(inputs=inputs, input_files=input_files)
@@ -2090,7 +2083,11 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
             def run_flow() -> None:
                 try:
                     self.stream = False
-                    result = self.kickoff(inputs=inputs, input_files=input_files)
+                    result = self.kickoff(
+                        inputs=inputs,
+                        input_files=input_files,
+                        restore_from_state_id=restore_from_state_id,
+                    )
                     result_holder.append(result)
                 except Exception as e:
                     # HumanFeedbackPending is expected control flow, not an error
@@ -2113,7 +2110,11 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
             return streaming_output
 
         async def _run_flow() -> Any:
-            return await self.kickoff_async(inputs, input_files)
+            return await self.kickoff_async(
+                inputs,
+                input_files,
+                restore_from_state_id=restore_from_state_id,
+            )
 
         try:
             asyncio.get_running_loop()
@@ -2128,6 +2129,7 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
         inputs: dict[str, Any] | None = None,
         input_files: dict[str, FileInput] | None = None,
         from_checkpoint: CheckpointConfig | None = None,
+        restore_from_state_id: str | None = None,
     ) -> Any | FlowStreamingOutput:
         """Start the flow execution asynchronously.
 
@@ -2141,10 +2143,23 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
             input_files: Optional dict of named file inputs for the flow.
             from_checkpoint: Optional checkpoint config. If ``restore_from``
                 is set, the flow resumes from that checkpoint.
+            restore_from_state_id: Optional UUID of a previously-persisted flow
+                whose latest snapshot should hydrate this run's state. The new
+                run is assigned a fresh ``state.id`` (or ``inputs["id"]`` if
+                pinned), so subsequent ``@persist`` writes land under a
+                separate persistence key. If the referenced state is not
+                found, falls back silently to baseline. Cannot be combined
+                with ``from_checkpoint``; passing both raises ``ValueError``.
 
         Returns:
             The final output from the flow, which is the result of the last executed method.
         """
+        if from_checkpoint is not None and restore_from_state_id is not None:
+            raise ValueError(
+                "Cannot combine `from_checkpoint` and `restore_from_state_id`. "
+                "These parameters target different state systems "
+                "(Checkpointing and @persist) and cannot be used together."
+            )
         restored = apply_checkpoint(self, from_checkpoint)
         if restored is not None:
             return await restored.kickoff_async(inputs=inputs, input_files=input_files)
@@ -2167,7 +2182,9 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
                 try:
                     self.stream = False
                     result = await self.kickoff_async(
-                        inputs=inputs, input_files=input_files
+                        inputs=inputs,
+                        input_files=input_files,
+                        restore_from_state_id=restore_from_state_id,
                     )
                     result_holder.append(result)
                 except Exception as e:
@@ -2199,9 +2216,9 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
         flow_id_token = None
         request_id_token = None
         if current_flow_id.get() is None:
-            flow_id_token = current_flow_id.set(self.execution_id)
+            flow_id_token = current_flow_id.set(self.flow_id)
         if current_flow_request_id.get() is None:
-            request_id_token = current_flow_request_id.set(self.execution_id)
+            request_id_token = current_flow_request_id.set(self.flow_id)
 
         try:
             # Reset flow state for fresh execution unless restoring from persistence
@@ -2224,16 +2241,54 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
                 if self._completed_methods:
                     self._is_execution_resuming = True
 
+            # Fork hydration: when restore_from_state_id is set and persistence is
+            # available, hydrate self._state from the source UUID's latest snapshot
+            # and reassign state.id to a fresh value so subsequent @persist writes
+            # don't extend the source flow's history. If the source state is not
+            # found, fall through silently to the existing inputs handling.
+            fork_succeeded = False
+            if restore_from_state_id is not None and self.persistence is not None:
+                stored_state = self.persistence.load_state(restore_from_state_id)
+                if stored_state:
+                    self._log_flow_event(
+                        f"Forking flow state from UUID: {restore_from_state_id}"
+                    )
+                    self._restore_state(stored_state)
+                    # Pin to inputs["id"] when provided, otherwise mint a fresh
+                    # UUID. NOTE: pinning inputs.id while forking shares a
+                    # persistence key with another flow — usually you want only
+                    # restore_from_state_id.
+                    new_state_id = (inputs.get("id") if inputs else None) or str(
+                        uuid4()
+                    )
+                    if isinstance(self._state, dict):
+                        self._state["id"] = new_state_id
+                    elif isinstance(self._state, BaseModel):
+                        setattr(self._state, "id", new_state_id)  # noqa: B010
+                    fork_succeeded = True
+                else:
+                    self._log_flow_event(
+                        "No flow state found for restore_from_state_id: "
+                        f"{restore_from_state_id}; proceeding without hydration",
+                        color="yellow",
+                    )
+
             if inputs:
-                # Override the id in the state if it exists in inputs
-                if "id" in inputs:
+                # Override the id in the state if it exists in inputs.
+                # Skip when the fork already assigned state.id above.
+                if "id" in inputs and not fork_succeeded:
                     if isinstance(self._state, dict):
                         self._state["id"] = inputs["id"]
                     elif isinstance(self._state, BaseModel):
                         setattr(self._state, "id", inputs["id"])  # noqa: B010
 
                 # If persistence is enabled, attempt to restore the stored state using the provided id.
-                if "id" in inputs and self.persistence is not None:
+                # Skip when the fork already restored self._state above.
+                if (
+                    "id" in inputs
+                    and self.persistence is not None
+                    and not fork_succeeded
+                ):
                     restore_uuid = inputs["id"]
                     stored_state = self.persistence.load_state(restore_uuid)
                     if stored_state:
@@ -2416,6 +2471,7 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
         inputs: dict[str, Any] | None = None,
         input_files: dict[str, FileInput] | None = None,
         from_checkpoint: CheckpointConfig | None = None,
+        restore_from_state_id: str | None = None,
     ) -> Any | FlowStreamingOutput:
         """Native async method to start the flow execution. Alias for kickoff_async.
 
@@ -2424,11 +2480,19 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
             input_files: Optional dict of named file inputs for the flow.
             from_checkpoint: Optional checkpoint config. If ``restore_from``
                 is set, the flow resumes from that checkpoint.
+            restore_from_state_id: Optional UUID of a previously-persisted flow
+                whose latest snapshot should hydrate this run's state. See
+                ``kickoff_async`` for full semantics.
 
         Returns:
             The final output from the flow, which is the result of the last executed method.
         """
-        return await self.kickoff_async(inputs, input_files, from_checkpoint)
+        return await self.kickoff_async(
+            inputs,
+            input_files,
+            from_checkpoint,
+            restore_from_state_id=restore_from_state_id,
+        )
 
     async def _replay_recorded_events(self) -> None:
         """Dispatch recorded ``MethodExecution*`` events from the event record."""

lib/crewai/tests/test_crew.py

@@ -4519,8 +4519,8 @@ def test_sets_flow_context_when_using_crewbase_pattern_inside_flow():
     flow.kickoff()
 
     assert captured_crew is not None
-    assert captured_crew._flow_id == flow.execution_id  # type: ignore[attr-defined]
-    assert captured_crew._request_id == flow.execution_id  # type: ignore[attr-defined]
+    assert captured_crew._flow_id == flow.flow_id  # type: ignore[attr-defined]
+    assert captured_crew._request_id == flow.flow_id  # type: ignore[attr-defined]
 
 
 def test_sets_flow_context_when_outside_flow(researcher, writer):
@@ -4554,8 +4554,8 @@ def test_sets_flow_context_when_inside_flow(researcher, writer):
 
     flow = MyFlow()
     result = flow.kickoff()
-    assert result._flow_id == flow.execution_id  # type: ignore[attr-defined]
-    assert result._request_id == flow.execution_id  # type: ignore[attr-defined]
+    assert result._flow_id == flow.flow_id  # type: ignore[attr-defined]
+    assert result._request_id == flow.flow_id  # type: ignore[attr-defined]
 
 
 def test_reset_knowledge_with_no_crew_knowledge(researcher, writer):

lib/crewai/tests/test_flow_execution_id.py

@@ -1,127 +0,0 @@
-"""Regression tests for ``Flow.execution_id``.
-
-``execution_id`` is the stable tracking identifier for a single flow run.
-It must stay independent of ``state.id`` so that consumers passing an
-``id`` in ``inputs`` (used for persistence restore) cannot destabilize
-the identity used by telemetry, tracing, and external correlation.
-"""
-
-from __future__ import annotations
-
-from typing import Any
-
-import pytest
-from crewai.flow.flow import Flow, FlowState, start
-from crewai.flow.flow_context import current_flow_id, current_flow_request_id
-
-
-class _CaptureState(FlowState):
-    captured_flow_id: str = ""
-    captured_state_id: str = ""
-    captured_current_flow_id: str = ""
-    captured_execution_id: str = ""
-
-
-class _IdentityCaptureFlow(Flow[_CaptureState]):
-    initial_state = _CaptureState
-
-    @start()
-    def capture(self) -> None:
-        self.state.captured_flow_id = self.flow_id
-        self.state.captured_state_id = self.state.id
-        self.state.captured_current_flow_id = current_flow_id.get() or ""
-        self.state.captured_execution_id = self.execution_id
-
-
-def test_execution_id_defaults_to_fresh_uuid_per_instance() -> None:
-    a = _IdentityCaptureFlow()
-    b = _IdentityCaptureFlow()
-
-    assert a.execution_id
-    assert b.execution_id
-    assert a.execution_id != b.execution_id
-
-
-def test_execution_id_survives_consumer_id_in_inputs() -> None:
-    flow = _IdentityCaptureFlow()
-    original_execution_id = flow.execution_id
-
-    flow.kickoff(inputs={"id": "consumer-supplied-id"})
-
-    assert flow.state.id == "consumer-supplied-id"
-    assert flow.flow_id == "consumer-supplied-id"
-    assert flow.execution_id == original_execution_id
-    assert flow.execution_id != "consumer-supplied-id"
-
-
-def test_two_runs_with_same_consumer_id_have_distinct_execution_ids() -> None:
-    flow_a = _IdentityCaptureFlow()
-    flow_b = _IdentityCaptureFlow()
-
-    colliding_id = "shared-consumer-id"
-    flow_a.kickoff(inputs={"id": colliding_id})
-    flow_b.kickoff(inputs={"id": colliding_id})
-
-    assert flow_a.state.id == colliding_id
-    assert flow_b.state.id == colliding_id
-    assert flow_a.execution_id != flow_b.execution_id
-
-
-def test_execution_id_is_writable() -> None:
-    flow = _IdentityCaptureFlow()
-    flow.execution_id = "external-task-id"
-
-    assert flow.execution_id == "external-task-id"
-
-    flow.kickoff(inputs={"id": "consumer-supplied-id"})
-    assert flow.execution_id == "external-task-id"
-    assert flow.state.id == "consumer-supplied-id"
-
-
-def test_current_flow_id_context_var_matches_execution_id() -> None:
-    flow = _IdentityCaptureFlow()
-    flow.execution_id = "external-task-id"
-
-    flow.kickoff(inputs={"id": "consumer-supplied-id"})
-
-    assert flow.state.captured_current_flow_id == "external-task-id"
-    assert flow.state.captured_flow_id == "consumer-supplied-id"
-    assert flow.state.captured_execution_id == "external-task-id"
-
-
-def test_execution_id_not_included_in_serialized_state() -> None:
-    flow = _IdentityCaptureFlow()
-    flow.execution_id = "external-task-id"
-    flow.kickoff()
-
-    dumped = flow.state.model_dump()
-    assert "execution_id" not in dumped
-    assert "_execution_id" not in dumped
-    assert dumped["id"] == flow.state.id
-
-
-def test_dict_state_flow_also_exposes_stable_execution_id() -> None:
-    class DictFlow(Flow[dict[str, Any]]):
-        initial_state = dict  # type: ignore[assignment]
-
-        @start()
-        def noop(self) -> None:
-            pass
-
-    flow = DictFlow()
-    original = flow.execution_id
-    flow.kickoff(inputs={"id": "consumer-supplied-id"})
-
-    assert flow.state["id"] == "consumer-supplied-id"
-    assert flow.execution_id == original
-
-
-@pytest.fixture(autouse=True)
-def _reset_flow_context_vars():
-    yield
-    for var in (current_flow_id, current_flow_request_id):
-        try:
-            var.set(None)
-        except LookupError:
-            # ContextVar was never set in this context; nothing to reset.
-            pass

lib/crewai/tests/test_flow_persistence.py

@@ -3,6 +3,7 @@
 import os
 from typing import Dict, List
 
+import pytest
 from crewai.flow.flow import Flow, FlowState, listen, start
 from crewai.flow.persistence import persist
 from crewai.flow.persistence.sqlite import SQLiteFlowPersistence
@@ -248,3 +249,242 @@ def test_persistence_with_base_model(tmp_path):
     assert message.type == "text"
     assert message.content == "Hello, World!"
     assert isinstance(flow.state._unwrap(), State)
+
+
+def test_fork_with_restore_from_state_id(tmp_path):
+    """Fork: restore_from_state_id hydrates state from source flow_uuid; new run gets a
+    fresh state.id; source's history is preserved (the fork's @persist writes go under
+    the new state.id, not the source's)."""
+    db_path = os.path.join(tmp_path, "test_flows.db")
+    persistence = SQLiteFlowPersistence(db_path)
+
+    class ForkableFlow(Flow[TestState]):
+        @start()
+        @persist(persistence)
+        def step(self):
+            self.state.counter += 1
+
+    # Run 1: build up source state. counter goes 0 -> 1.
+    flow1 = ForkableFlow(persistence=persistence)
+    flow1.kickoff()
+    source_uuid = flow1.state.id
+    assert flow1.state.counter == 1
+
+    # Resume on the same uuid bumps counter to 2 in the SAME flow_uuid history.
+    flow1b = ForkableFlow(persistence=persistence)
+    flow1b.kickoff(inputs={"id": source_uuid})
+    assert flow1b.state.counter == 2
+    assert persistence.load_state(source_uuid)["counter"] == 2
+
+    # Fork: hydrate from source, but persist under a fresh state.id.
+    flow2 = ForkableFlow(persistence=persistence)
+    flow2.kickoff(restore_from_state_id=source_uuid)
+
+    # Fork has a different state.id from the source.
+    assert flow2.state.id != source_uuid
+    # Hydrated from source's latest snapshot (counter=2), then incremented to 3.
+    assert flow2.state.counter == 3
+
+    # Source's history is unchanged after the fork.
+    assert persistence.load_state(source_uuid)["counter"] == 2
+
+    # Fork's writes landed under its own state.id.
+    assert persistence.load_state(flow2.state.id)["counter"] == 3
+
+
+def test_fork_with_pinned_state_id(tmp_path):
+    """Fork into a pinned state.id (inputs.id supplied alongside restore_from_state_id):
+    the new run uses inputs.id as state.id and hydrates from restore_from_state_id."""
+    db_path = os.path.join(tmp_path, "test_flows.db")
+    persistence = SQLiteFlowPersistence(db_path)
+
+    class PinnableFlow(Flow[TestState]):
+        @start()
+        @persist(persistence)
+        def step(self):
+            self.state.counter += 1
+
+    flow1 = PinnableFlow(persistence=persistence)
+    flow1.kickoff()
+    source_uuid = flow1.state.id
+    assert flow1.state.counter == 1
+
+    pinned_uuid = "pinned-fork-uuid-1234"
+    flow2 = PinnableFlow(persistence=persistence)
+    flow2.kickoff(
+        inputs={"id": pinned_uuid},
+        restore_from_state_id=source_uuid,
+    )
+
+    # state.id pinned to inputs.id, NOT the source uuid.
+    assert flow2.state.id == pinned_uuid
+    # Hydrated from source: counter started at 1, step incremented to 2.
+    assert flow2.state.counter == 2
+    # Source's history is unchanged.
+    assert persistence.load_state(source_uuid)["counter"] == 1
+    # Fork's writes are under the pinned uuid.
+    assert persistence.load_state(pinned_uuid)["counter"] == 2
+
+
+def test_restore_from_state_id_not_found_silent_fallback(tmp_path):
+    """Lookup miss on restore_from_state_id silently falls through to default behavior."""
+    db_path = os.path.join(tmp_path, "test_flows.db")
+    persistence = SQLiteFlowPersistence(db_path)
+
+    class FallbackFlow(Flow[TestState]):
+        @start()
+        @persist(persistence)
+        def step(self):
+            self.state.counter += 1
+
+    flow = FallbackFlow(persistence=persistence)
+    # No source UUID exists — should not raise.
+    flow.kickoff(restore_from_state_id="no-such-uuid")
+
+    # Default state path: counter starts at 0 and step increments to 1.
+    assert flow.state.counter == 1
+    # state.id is the auto-generated one, NOT the missing source.
+    assert flow.state.id != "no-such-uuid"
+
+
+def test_restore_from_state_id_none_is_no_op(tmp_path):
+    """restore_from_state_id=None (default) preserves baseline kickoff behavior."""
+    db_path = os.path.join(tmp_path, "test_flows.db")
+    persistence = SQLiteFlowPersistence(db_path)
+
+    class BaselineFlow(Flow[TestState]):
+        @start()
+        @persist(persistence)
+        def step(self):
+            self.state.counter += 1
+
+    flow = BaselineFlow(persistence=persistence)
+    flow.kickoff(restore_from_state_id=None)
+    assert flow.state.counter == 1
+
+
+def test_fork_conflict_with_from_checkpoint_raises():
+    """Passing both from_checkpoint and restore_from_state_id raises ValueError, naming
+    both parameters."""
+    from crewai.state import CheckpointConfig
+
+    class ConflictFlow(Flow[TestState]):
+        @start()
+        def step(self):
+            pass
+
+    flow = ConflictFlow()
+    with pytest.raises(ValueError) as excinfo:
+        flow.kickoff(
+            from_checkpoint=CheckpointConfig(),
+            restore_from_state_id="some-uuid",
+        )
+    msg = str(excinfo.value)
+    assert "from_checkpoint" in msg
+    assert "restore_from_state_id" in msg
+
+
+@pytest.mark.asyncio
+async def test_fork_via_kickoff_async(tmp_path):
+    """kickoff_async honors restore_from_state_id: hydrates from source, mints fresh
+    state.id, persists under the new id, source history preserved."""
+    db_path = os.path.join(tmp_path, "test_flows.db")
+    persistence = SQLiteFlowPersistence(db_path)
+
+    class AsyncForkableFlow(Flow[TestState]):
+        @start()
+        @persist(persistence)
+        def step(self):
+            self.state.counter += 1
+
+    flow1 = AsyncForkableFlow(persistence=persistence)
+    await flow1.kickoff_async()
+    source_uuid = flow1.state.id
+    assert flow1.state.counter == 1
+
+    flow2 = AsyncForkableFlow(persistence=persistence)
+    await flow2.kickoff_async(restore_from_state_id=source_uuid)
+
+    assert flow2.state.id != source_uuid
+    assert flow2.state.counter == 2
+    assert persistence.load_state(source_uuid)["counter"] == 1
+    assert persistence.load_state(flow2.state.id)["counter"] == 2
+
+
+@pytest.mark.asyncio
+async def test_fork_via_akickoff(tmp_path):
+    """akickoff is the public async alias and must accept restore_from_state_id with
+    the same semantics as kickoff_async."""
+    db_path = os.path.join(tmp_path, "test_flows.db")
+    persistence = SQLiteFlowPersistence(db_path)
+
+    class AkickoffForkableFlow(Flow[TestState]):
+        @start()
+        @persist(persistence)
+        def step(self):
+            self.state.counter += 1
+
+    flow1 = AkickoffForkableFlow(persistence=persistence)
+    await flow1.akickoff()
+    source_uuid = flow1.state.id
+    assert flow1.state.counter == 1
+
+    flow2 = AkickoffForkableFlow(persistence=persistence)
+    await flow2.akickoff(restore_from_state_id=source_uuid)
+
+    assert flow2.state.id != source_uuid
+    assert flow2.state.counter == 2
+    assert persistence.load_state(source_uuid)["counter"] == 1
+    assert persistence.load_state(flow2.state.id)["counter"] == 2
+
+
+@pytest.mark.asyncio
+async def test_akickoff_pinned_fork(tmp_path):
+    """akickoff with both inputs.id and restore_from_state_id pins state.id while
+    hydrating from the source."""
+    db_path = os.path.join(tmp_path, "test_flows.db")
+    persistence = SQLiteFlowPersistence(db_path)
+
+    class PinnableAsyncFlow(Flow[TestState]):
+        @start()
+        @persist(persistence)
+        def step(self):
+            self.state.counter += 1
+
+    flow1 = PinnableAsyncFlow(persistence=persistence)
+    await flow1.akickoff()
+    source_uuid = flow1.state.id
+
+    pinned_uuid = "pinned-akickoff-fork-uuid"
+    flow2 = PinnableAsyncFlow(persistence=persistence)
+    await flow2.akickoff(
+        inputs={"id": pinned_uuid},
+        restore_from_state_id=source_uuid,
+    )
+
+    assert flow2.state.id == pinned_uuid
+    assert flow2.state.counter == 2
+    assert persistence.load_state(source_uuid)["counter"] == 1
+    assert persistence.load_state(pinned_uuid)["counter"] == 2
+
+
+@pytest.mark.asyncio
+async def test_akickoff_fork_conflict_with_from_checkpoint_raises():
+    """akickoff must raise the same conflict ValueError as kickoff/kickoff_async when
+    both from_checkpoint and restore_from_state_id are set."""
+    from crewai.state import CheckpointConfig
+
+    class AsyncConflictFlow(Flow[TestState]):
+        @start()
+        def step(self):
+            pass
+
+    flow = AsyncConflictFlow()
+    with pytest.raises(ValueError) as excinfo:
+        await flow.akickoff(
+            from_checkpoint=CheckpointConfig(),
+            restore_from_state_id="some-uuid",
+        )
+    msg = str(excinfo.value)
+    assert "from_checkpoint" in msg
+    assert "restore_from_state_id" in msg