fix(sdk/python): address review feedback — stop/start race + cross-loop lock (#623)

1. ResultCache.stop(): compare-before-clear so a concurrent start()
   that installs fresh references between the await and the cleanup
   doesn't get its new task orphaned.

2. AsyncExecutionManager.stop(): skip the _execution_lock section on
   cross-loop stop — the lock is bound to the owning loop, taking it
   from a foreign loop raises the same 'got Future attached to a
   different loop' error one line below the fix.

Author: 7vignesh

sdk/python/agentfield/async_execution_manager.py

@@ -365,12 +365,17 @@ async def stop(self) -> None:
         self._event_stream_task = None
         self._loop = None
 
-        # Cancel all active executions
-        async with self._execution_lock:
-            for execution in self._executions.values():
-                if execution.is_active:
-                    execution.cancel("Manager shutdown")
-                    self._release_capacity_for_execution(execution)
+        # Cancel all active executions. The _execution_lock is an asyncio.Lock
+        # bound to the owning loop — taking it from a foreign loop would raise
+        # "got Future attached to a different loop". On a cross-loop stop we
+        # skip this section: the executions will be cancelled when the owning
+        # loop tears down or the manager is restarted (#623 review feedback).
+        if owning_loop is current_loop:
+            async with self._execution_lock:
+                for execution in self._executions.values():
+                    if execution.is_active:
+                        execution.cancel("Manager shutdown")
+                        self._release_capacity_for_execution(execution)
 
         # Stop components
         await self.connection_manager.close()

sdk/python/agentfield/result_cache.py

@@ -236,9 +236,16 @@ async def stop(self) -> None:
             shutdown_event.set()
         await cancel_and_await_if_same_loop(task, owning_loop)
 
-        self._cleanup_task = None
-        self._shutdown_event = None
-        self._loop = None
+        # Compare-before-clear: only null the references if they're still the
+        # ones we snapshotted. A concurrent start() on another thread/loop
+        # could have installed fresh references in between; clobbering those
+        # would orphan the new cleanup task (#623 review feedback).
+        if self._cleanup_task is task:
+            self._cleanup_task = None
+        if self._shutdown_event is shutdown_event:
+            self._shutdown_event = None
+        if self._loop is owning_loop:
+            self._loop = None
 
         with self._lock:
             self._cache.clear()