RunnableWithFallbacks.Batch/Abatch Leave Root Runs Open On Unhandled Exception
When an input raises an exception not in exceptions_to_handle, batch/abatch re-raise immediately, skipping the cleanup loop that closes root callback runs, causing them to remain pending forever in tracers like LangSmith.
In RunnableWithFallbacks.batch and abatch, the line `if first_to_raise: raise first_to_raise` is placed before the cleanup loop that calls `on_chain_error`/`on_chain_end` on each run manager. This causes an early return that bypasses run closure.
Run the provided reproduction code with LangChain Core 1.6.4. Observe that for batch and abatch, root runs started=3 but closed=0, whereas invoke correctly closes the root run.
Fixing Code Block
--- a/langchain_core/runnables/fallbacks.py
+++ b/langchain_core/runnables/fallbacks.py
@@ batch method @@
- if first_to_raise:
- raise first_to_raise
for i, run_manager in enumerate(run_managers):
if not run_manager.is_done():
if first_exceptions[i]:
run_manager.on_chain_error(first_exceptions[i])
else:
run_manager.on_chain_end(outputs[i])
+ if first_to_raise:
+ raise first_to_raise
return outputs
@@ abatch method @@
- if first_to_raise:
- raise first_to_raise
- for i, run_manager in enumerate(run_managers):
- if not run_manager.is_done():
- if first_exceptions[i]:
- run_manager.on_chain_error(first_exceptions[i])
- else:
- run_manager.on_chain_end(outputs[i])
+ await asyncio.gather(*[
+ run_manager.on_chain_error(first_exceptions[i])
+ if first_exceptions[i] else run_manager.on_chain_end(outputs[i])
+ for i, run_manager in enumerate(run_managers)
+ if not run_manager.is_done()
+ ])
+ if first_to_raise:
+ raise first_to_raise
return outputs
Remove the early raise inside the while loop so that the cleanup loop is always executed. After closing all pending run managers, raise the first unhandled exception. In abatch, use asyncio.gather to close runs concurrently, matching the asynchronous execution model.
Edge Case Audit
The change ensures callbacks are closed before raising, which may alter the timing of exception propagation relative to callback completion. In abatch, using asyncio.gather introduces concurrent run completion; ensure that run managers are safe for concurrent close operations. Test thoroughly with return_exceptions=True and multiple mixed handled/unhandled exceptions. Rollback: revert this patch if any downstream code depends on the early raise before run closure.
