Host-future error inside a coroutine under `asyncio.gather` bypasses its `try/except`; two such errors stall the scheduler (deterministic repro for #732)
Version: monty-pool / monty-runtime 0.0.23 (worker built with cargo install monty-runtime --no-default-features), macOS arm64.
Summary
When a pending external call is settled with ResumeValue::Error via Checkout::resume_futures, and the await on that future sits inside a coroutine that is running as a task of asyncio.gather (even a gather of one), the exception is not delivered to that coroutine's try/except. It surfaces at the gather await instead. When two futures in the same gather are settled with errors, the run ends with
RuntimeError: Internal error in monty: asyncio scheduler stalled: no ready tasks and no pending external calls— the downgraded assert from #732, which was closed without a repro. This is a deterministic one.
Repro (monty-pool 0.0.23)
Host answers every FunctionCall with ResumeValue::Future, then answers ResolveFutures with Return(Int(i)) for tools not in errs and Error(RuntimeError("tool i: boom")) for those in errs.
import asyncio
async def safe(coro):
try:
return await coro
except Exception as e:
return f'ERROR: {e}'
await asyncio.gather(safe(a()), safe(b()), safe(c()))| errs | expected (CPython) | monty 0.0.23 |
|---|---|---|
[] |
[0, 1, 2] |
[0, 1, 2] ✓ |
[0] |
['ERROR: tool 0: boom', 1, 2] |
uncaught RuntimeError: tool 0: boom at the gather line |
[2] |
[0, 1, 'ERROR: tool 2: boom'] |
uncaught RuntimeError: tool 2: boom at the gather line |
[0, 2] |
['ERROR: …', 1, 'ERROR: …'] |
RuntimeError: Internal error in monty: asyncio scheduler stalled: no ready tasks and no pending external calls |
except RuntimeError instead of Exception, errs [0] |
caught | uncaught at the gather line |
gather of one: await asyncio.gather(safe(a())), errs [0] |
['ERROR: tool 0: boom'] |
uncaught at the gather line |
Control cases that behave correctly:
await safe(a())(same wrapper, no gather), errs[0]→'ERROR: tool 0: boom'✓- top-level
try: r = await a() except RuntimeError→ caught ✓ try: await asyncio.gather(a(), b(), c()) except RuntimeError→ caught at the gather ✓await asyncio.gather(a(), b(), c())with errs[0]→ propagatesRuntimeError: tool 0: boom✓ (correct: no handler)
So the exception is delivered to the gather's waiter rather than to the task whose await produced it; with a second failing future the first teardown (fail_for_call → cancel_task on the gather's tasks) leaves nothing ready and nothing pending, which is the #732 state.
Likely origin: #772 (merged before v0.0.23) — "drops a gather child whose own external call failed so it doesn't stay parked forever". The child is dropped instead of having the exception raised at its await, so its try/except never runs; the gather is settled directly from fail_for_call. docs/limitations/asyncio.md does not list this as a divergence.
cargo run --example gather_spike -- $(which monty)//! Throwaway: three host futures under `asyncio.gather`, answered with a mix of
//! `Return` and `Error`, to reproduce "asyncio scheduler stalled".
use monty_pool::{Pool, PoolConfig, ReplConfig, ResumeValue, TurnEvent, on_print_sync};
use monty_types::{ExcType, MontyException, MontyObject};
#[tokio::main]
async fn main() {
let bin = std::env::args().nth(1).expect("monty path");
let pool = Pool::new(PoolConfig::subprocess(bin)).await.unwrap();
for (label, code, errs) in [
("wrapper safe(), two err", "async def safe(coro):\n try:\n return await coro\n except Exception as e:\n return f'ERROR: {e}'\nawait asyncio.gather(safe(a()), safe(b()), safe(c()))", vec![0, 2]),
("wrapper safe(), all ok", "async def safe(coro):\n try:\n return await coro\n except Exception as e:\n return f'ERROR: {e}'\nawait asyncio.gather(safe(a()), safe(b()), safe(c()))", vec![]),
("wrapper safe(), one err first", "async def safe(coro):\n try:\n return await coro\n except Exception as e:\n return f'ERROR: {e}'\nawait asyncio.gather(safe(a()), safe(b()), safe(c()))", vec![0]),
("wrapper safe(), one err last", "async def safe(coro):\n try:\n return await coro\n except Exception as e:\n return f'ERROR: {e}'\nawait asyncio.gather(safe(a()), safe(b()), safe(c()))", vec![2]),
("wrapper no try, one err", "async def w(coro):\n return await coro\nawait asyncio.gather(w(a()), w(b()), w(c()))", vec![0]),
("wrapper except RuntimeError, one err", "async def safe(coro):\n try:\n return await coro\n except RuntimeError as e:\n return f'ERROR: {e}'\nawait asyncio.gather(safe(a()), safe(b()), safe(c()))", vec![0]),
("wrapper try, single task not gathered", "async def safe(coro):\n try:\n return await coro\n except Exception as e:\n return f'ERROR: {e}'\nawait safe(a())", vec![0]),
("wrapper try, gather of one", "async def safe(coro):\n try:\n return await coro\n except Exception as e:\n return f'ERROR: {e}'\nawait asyncio.gather(safe(a()))", vec![0]),
("top-level try around direct await", "try:\n r = await a()\nexcept RuntimeError as e:\n r = f'ERROR: {e}'\nr", vec![0]),
("all ok", "await asyncio.gather(a(), b(), c())", vec![]),
("one err, gather default", "await asyncio.gather(a(), b(), c())", vec![0]),
("two err, gather default", "await asyncio.gather(a(), b(), c())", vec![0, 2]),
("two err, return_exceptions", "await asyncio.gather(a(), b(), c(), return_exceptions=True)", vec![0, 2]),
("two err, try/except around gather", "try:\n r = await asyncio.gather(a(), b(), c())\nexcept RuntimeError as e:\n r = str(e)\nr", vec![0, 2]),
("sequential try/except", "out = []\nfor f in (a, b, c):\n try:\n out.append(await f())\n except RuntimeError as e:\n out.append(str(e))\nout", vec![0, 2]),
] {
let mut c = pool.checkout(&ReplConfig::default()).await.unwrap();
let mut quiet = on_print_sync(|_, _| {});
let mut ev = c
.feed(&format!("import asyncio\n{code}"), Vec::new(), Vec::new(), true, &mut quiet)
.await;
let mut n = 0u32;
let mut names = std::collections::HashMap::new();
let result = loop {
match ev {
Ok(TurnEvent::Complete(v)) => break format!("Complete({v:?})"),
Ok(TurnEvent::FunctionCall { function_name, call_id, .. }) => {
let idx = match function_name.as_str() { "a" => 0, "b" => 1, _ => 2 };
names.insert(call_id, idx);
n += 1;
ev = c.resume(ResumeValue::Future, &mut quiet).await;
}
Ok(TurnEvent::ResolveFutures { pending_call_ids }) => {
let answers = pending_call_ids
.iter()
.map(|id| {
let idx = names[id];
if errs.contains(&idx) {
(*id, ResumeValue::Error(MontyException::new(ExcType::RuntimeError, Some(format!("tool {idx}: boom")))))
} else {
(*id, ResumeValue::Return(MontyObject::Int((idx as i64).into())))
}
})
.collect();
ev = c.resume_futures(answers, &mut quiet).await;
}
Ok(other) => break format!("unexpected {other:?}"),
Err(e) => break format!("Err({e})"),
}
};
println!("{label:38} calls={n} → {result}");
let _ = c.finish().await;
}
}
Impact
The async def safe(coro): try: return await coro except … wrapper is the idiomatic way to gather several fallible host calls and keep the successful results. Today a model-written cell using it loses the whole batch when any call fails.
Source: pydantic/monty