From e7d7f47d28bbdaa729607222c6bd3b047ee82380 Mon Sep 17 00:00:00 2001
From: Jeremiah Lowin <153965+jlowin@users.noreply.github.com>
Date: Fri, 6 Sep 2024 11:02:38 -0400
Subject: [PATCH] Reset task parent tracking when nested flows are created

---
 src/controlflow/flows/flow.py |  3 ++-
 tests/flows/test_flows.py     | 11 +++++++++++
 2 files changed, 13 insertions(+), 1 deletion(-)

diff --git a/src/controlflow/flows/flow.py b/src/controlflow/flows/flow.py
index 35a78952..8bcf9369 100644
--- a/src/controlflow/flows/flow.py
+++ b/src/controlflow/flows/flow.py
@@ -110,7 +110,8 @@ def add_events(self, events: list[Event]):
 
     @contextmanager
     def create_context(self):
-        with ctx(flow=self):
+        # creating a new flow will reset any parent task tracking
+        with ctx(flow=self, tasks=None):
             yield self
 
 
diff --git a/tests/flows/test_flows.py b/tests/flows/test_flows.py
index 9da9851f..f5a85ac3 100644
--- a/tests/flows/test_flows.py
+++ b/tests/flows/test_flows.py
@@ -66,6 +66,17 @@ def test_get_flow_nested_contexts(self):
             assert get_flow() == flow1
         assert get_flow() is None
 
+    def test_flow_context_resets_task_tracking(self):
+        parent_task = Task("Parent task")
+        with parent_task:
+            assert ctx.get("tasks") == [parent_task]
+            with Flow():
+                assert ctx.get("tasks") is None
+                nested_task = Task("Nested task")
+                assert nested_task.parent is None
+            assert ctx.get("tasks") == [parent_task]
+        assert ctx.get("tasks") == []
+
 
 class TestFlowHistory:
     def test_get_events_empty(self):