From 36d5c1a150c46d6ee92d9dc255a34ae66ff1db4c Mon Sep 17 00:00:00 2001
From: Joshua Harlow <harlowja@yahoo-inc.com>
Date: Tue, 21 May 2013 18:35:48 -0700
Subject: [PATCH] Ensure we associate with parent flows as well.

---
 taskflow/job.py | 14 +++++++++++++-
 1 file changed, 13 insertions(+), 1 deletion(-)

diff --git a/taskflow/job.py b/taskflow/job.py
index aaaa424d..3b32a5e4 100644
--- a/taskflow/job.py
+++ b/taskflow/job.py
@@ -153,7 +153,19 @@ class Job(object):
         self._change_state(states.CLAIMED)
 
     def run(self, flow, *args, **kwargs):
-        self.associate(flow)
+        already_associated = []
+
+        def associate_all(f):
+            # Associate with the flow and the flows parents and so on.
+            if f in already_associated:
+                return
+            self.associate(f)
+            already_associated.append(f)
+            if f.parents:
+                for p in f.parents:
+                    associate_all(p)
+
+        associate_all(flow)
         return flow.run(self.context, *args, **kwargs)
 
     def unclaim(self):