Skip to content

Commit a12ba5a

Browse files
hellovaiclaude
andauthored
feat: Use AbortSignal in typescript, add a native timeout capability in python. (#2373)
## Summary Modernizes the abort controller API documentation to use the AbortSignal pattern, aligning with web standards while improving developer experience. **Key Changes:** - **Renamed documentation**: "Abort Controllers" → "AbortSignals / Timeouts" to better reflect modern usage patterns - **Updated TypeScript API**: Changed from `abortController` parameter to `abortSignal` to align with web standards - **Added modern timeout pattern for TS**: Introduced `AbortSignal.timeout(5000)` for simple timeout scenarios - **Enhanced Python support**: Added `AbortController(timeout_ms=5000)` constructor for built-in timeouts - **Updated cross-references**: Fixed all internal links and redirects throughout documentation ## Code Examples ### Before (old AbortController pattern): ```typescript const controller = new AbortController() const result = await b.ExtractResume(text, { abortController: controller // Old parameter name }) ``` ### After (modern AbortSignal pattern): ```typescript // Simple timeout approach const result = await b.ExtractResume(text, { abortSignal: AbortSignal.timeout(5000) // Modern web standard }) // Manual control when needed const controller = new AbortController() const result = await b.ExtractResume(text, { abortSignal: controller.signal // Updated parameter name }) ``` ### Python Simplification: **Before (complex manual timeout):** ```python async def with_timeout(operation, timeout_seconds): controller = AbortController() async def run_with_timeout(): timeout_task = asyncio.create_task(asyncio.sleep(timeout_seconds)) operation_task = asyncio.create_task(operation(controller)) done, pending = await asyncio.wait( [timeout_task, operation_task], return_when=asyncio.FIRST_COMPLETED ) if timeout_task in done: controller.abort() raise TimeoutError(f"Operation timed out after {timeout_seconds}s") return operation_task.result() return await run_with_timeout() # Usage result = await with_timeout( lambda ctrl: b.ExtractData(input, baml_options={"abort_controller": ctrl}), 5 # 5 second timeout ) ``` **After (built-in timeout support):** ```python # Much simpler with built-in timeout controller = AbortController(timeout_ms=5000) # Automatic timeout result = await b.ExtractData(input, baml_options={"abort_controller": controller}) ``` ## Impact This refactor eliminates the need for complex `asyncio.wait()` and manual task management by moving timeout functionality directly into the API, making it much cleaner and less error-prone while following modern web standards. 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <noreply@anthropic.com>
1 parent 99204c8 commit a12ba5a

80 files changed

Lines changed: 3728 additions & 3878 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎engine/baml-runtime/src/async_vm_runtime.rs‎

Lines changed: 16 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -30,7 +30,7 @@ use crate::{
3030
tracingv2::storage::storage::Collector,
3131
type_builder::TypeBuilder,
3232
BamlRuntime as LlmRuntime, BamlSrcReader, FunctionResult, FunctionResultStream,
33-
InnerTraceStats, InternalRuntimeInterface, RuntimeContextManager,
33+
InnerTraceStats, InternalRuntimeInterface, RuntimeContextManager, TripWire,
3434
};
3535

3636
/// Async VM runtime.
@@ -144,7 +144,7 @@ impl BamlAsyncVmRuntime {
144144
cb: Option<&ClientRegistry>,
145145
collectors: Option<Vec<Arc<Collector>>>,
146146
env_vars: HashMap<String, String>,
147-
cancel_tripwire: Option<stream_cancel::Tripwire>,
147+
cancel_tripwire: Arc<TripWire>,
148148
) -> (anyhow::Result<FunctionResult>, FunctionCallId) {
149149
// TODO: Proper error handling. Refactor the API to return a Result.
150150
let (function_index, function_kind) = self
@@ -308,7 +308,7 @@ impl BamlAsyncVmRuntime {
308308

309309
// Spanwed future basically awaits the LLM call and
310310
// sends the result to the futures channel.
311-
let cancel_tripwire = cancel_tripwire.clone();
311+
let cancel_tripwire = cancel_tripwire.to_owned();
312312
async move {
313313
let result = llm_runtime
314314
.call_function(
@@ -397,7 +397,7 @@ impl BamlAsyncVmRuntime {
397397
cb: Option<&ClientRegistry>,
398398
collectors: Option<Vec<Arc<Collector>>>,
399399
env_vars: HashMap<String, String>,
400-
cancel_tripwire: Option<stream_cancel::Tripwire>,
400+
cancel_tripwire: Arc<TripWire>,
401401
) -> (anyhow::Result<FunctionResult>, FunctionCallId) {
402402
self.async_runtime.block_on(self.call_function(
403403
function_name,
@@ -420,9 +420,19 @@ impl BamlAsyncVmRuntime {
420420
cb: Option<&ClientRegistry>,
421421
collectors: Option<Vec<Arc<Collector>>>,
422422
env_vars: HashMap<String, String>,
423+
// FunctionResultStream is responsible for freeing the TripWire and the clean up.
424+
cancel_tripwire: Arc<TripWire>,
423425
) -> anyhow::Result<FunctionResultStream> {
424-
self.llm_runtime
425-
.stream_function(function_name, params, ctx, tb, cb, collectors, env_vars)
426+
self.llm_runtime.stream_function(
427+
function_name,
428+
params,
429+
ctx,
430+
tb,
431+
cb,
432+
collectors,
433+
env_vars,
434+
cancel_tripwire,
435+
)
426436
}
427437

428438
pub async fn build_request(

‎engine/baml-runtime/src/cli/repl.rs‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -31,7 +31,7 @@ use reedline::{
3131
Signal,
3232
};
3333

34-
use crate::BamlRuntime;
34+
use crate::{BamlRuntime, TripWire};
3535

3636
#[derive(Args, Clone, Debug)]
3737
pub struct ReplArgs {
@@ -287,7 +287,7 @@ impl ReplState {
287287
None,
288288
None,
289289
env_vars,
290-
None,
290+
TripWire::new(None),
291291
)
292292
.await;
293293
let function_result = res.0?;

‎engine/baml-runtime/src/cli/serve/mod.rs‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -39,7 +39,7 @@ use tokio_stream::StreamExt;
3939

4040
use crate::{
4141
client_registry::ClientRegistry, errors::ExposedError, internal::llm_client::LLMResponse,
42-
BamlRuntime, FunctionResult, RuntimeContextManager,
42+
BamlRuntime, FunctionResult, RuntimeContextManager, TripWire,
4343
};
4444

4545
#[derive(clap::Args, Clone, Debug)]
@@ -355,7 +355,7 @@ Tip: test that the server is up using `curl http://localhost:{}/_debug/ping`
355355
client_registry.as_ref(),
356356
None,
357357
env_vars,
358-
None,
358+
TripWire::new(None),
359359
)
360360
.await;
361361

@@ -456,6 +456,7 @@ Tip: test that the server is up using `curl http://localhost:{}/_debug/ping`
456456
client_registry.as_ref(),
457457
Some(vec![]),
458458
env_vars,
459+
TripWire::new(None),
459460
);
460461

461462
match result_stream {

‎engine/baml-runtime/src/eval_expr.rs‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,7 @@ use internal_baml_core::{
2222
use internal_baml_jinja::types::OutputFormatContent;
2323
use jsonish::{deserializer::deserialize_flags::Flag, helpers::render_output_format};
2424

25-
use crate::{BamlRuntime, FunctionResult};
25+
use crate::{BamlRuntime, FunctionResult, TripWire};
2626

2727
const MAX_STEPS: usize = 1000;
2828

@@ -327,7 +327,7 @@ async fn beta_reduce<'a>(
327327
None,
328328
None,
329329
env.env_vars.clone(),
330-
None,
330+
TripWire::new(None),
331331
)
332332
.await
333333
.0;
@@ -1149,7 +1149,7 @@ test Poems {
11491149
Some(on_event),
11501150
None,
11511151
HashMap::new(),
1152-
None,
1152+
TripWire::new(None),
11531153
None::<Box<dyn Fn()>>,
11541154
)
11551155
// .run_test("MakePerson", "TestMakePerson", &ctx, Some(on_event), None)
@@ -1283,7 +1283,7 @@ test TestMakePerson() {
12831283
Some(on_event),
12841284
None,
12851285
HashMap::new(),
1286-
None,
1286+
TripWire::new(None),
12871287
None::<Box<dyn Fn()>>,
12881288
)
12891289
// .run_test("MakePerson", "TestMakePerson", &ctx, Some(on_event), None)
@@ -1361,7 +1361,7 @@ test UseFunction() {
13611361
"OPENAI_API_KEY".to_string(),
13621362
std::env::var("OPENAI_API_KEY").expect("OPENAI_API_KEY is not set."),
13631363
)]),
1364-
None,
1364+
TripWire::new(None),
13651365
None::<Box<dyn Fn()>>,
13661366
)
13671367
.await;

‎engine/baml-runtime/src/lib.rs‎

Lines changed: 48 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -185,6 +185,43 @@ pub struct BamlRuntime {
185185
pub async_runtime: Arc<tokio::runtime::Runtime>,
186186
}
187187

188+
pub struct TripWire {
189+
trip_wire: Option<stream_cancel::Tripwire>,
190+
191+
on_drop: Option<Box<dyn Fn() + 'static + Send + Sync>>,
192+
}
193+
194+
impl TripWire {
195+
pub fn new(trip_wire: Option<stream_cancel::Tripwire>) -> Arc<Self> {
196+
Arc::new(Self {
197+
trip_wire,
198+
on_drop: None,
199+
})
200+
}
201+
202+
pub fn new_with_on_drop(
203+
trip_wire: Option<stream_cancel::Tripwire>,
204+
on_drop: Box<dyn Fn() + 'static + Send + Sync>,
205+
) -> Arc<Self> {
206+
Arc::new(Self {
207+
trip_wire,
208+
on_drop: Some(on_drop),
209+
})
210+
}
211+
212+
fn trip_wire(&self) -> Option<stream_cancel::Tripwire> {
213+
self.trip_wire.clone()
214+
}
215+
}
216+
217+
impl Drop for TripWire {
218+
fn drop(&mut self) {
219+
if let Some(on_drop) = self.on_drop.take() {
220+
on_drop();
221+
}
222+
}
223+
}
224+
188225
impl BamlRuntime {
189226
#[cfg(not(target_arch = "wasm32"))]
190227
fn get_tokio_singleton() -> Result<Arc<tokio::runtime::Runtime>> {
@@ -410,7 +447,7 @@ impl BamlRuntime {
410447
expr_tx: Option<mpsc::UnboundedSender<Vec<internal_baml_diagnostics::SerializedSpan>>>,
411448
collector: Option<Arc<Collector>>,
412449
env_vars: HashMap<String, String>,
413-
cancel_tripwire: Option<stream_cancel::Tripwire>,
450+
cancel_tripwire: Arc<TripWire>,
414451
on_tick: Option<G>,
415452
) -> (Result<TestResponse>, FunctionCallId)
416453
where
@@ -524,7 +561,7 @@ impl BamlRuntime {
524561
ExprEvalResult::LLMCall { name, args } => (name, args),
525562
};
526563

527-
let mut stream = self.inner.stream_function_impl_with_tripwire(
564+
let mut stream = self.inner.stream_function_impl(
528565
function_name,
529566
&params,
530567
self.tracer_wrapper.get_or_create_tracer(&env_vars),
@@ -533,7 +570,7 @@ impl BamlRuntime {
533570
self.async_runtime.clone(),
534571
// TODO: collectors here?
535572
vec![],
536-
cancel_tripwire.clone(),
573+
cancel_tripwire,
537574
)?;
538575
let (response_res, call_uuid) = stream
539576
.run(
@@ -626,7 +663,7 @@ impl BamlRuntime {
626663
on_event: Option<F>,
627664
collector: Option<Arc<Collector>>,
628665
env_vars: HashMap<String, String>,
629-
cancel_tripwire: Option<stream_cancel::Tripwire>,
666+
cancel_tripwire: Arc<TripWire>,
630667
on_tick: Option<G>,
631668
) -> (Result<TestResponse>, FunctionCallId)
632669
where
@@ -659,7 +696,7 @@ impl BamlRuntime {
659696
cb: Option<&ClientRegistry>,
660697
collectors: Option<Vec<Arc<Collector>>>,
661698
env_vars: HashMap<String, String>,
662-
cancel_tripwire: Option<stream_cancel::Tripwire>,
699+
cancel_tripwire: Arc<TripWire>,
663700
) -> (Result<FunctionResult>, FunctionCallId) {
664701
let fut = self.call_function(
665702
function_name,
@@ -683,7 +720,7 @@ impl BamlRuntime {
683720
cb: Option<&ClientRegistry>,
684721
collectors: Option<Vec<Arc<Collector>>>,
685722
env_vars: HashMap<String, String>,
686-
cancel_tripwire: Option<stream_cancel::Tripwire>,
723+
cancel_tripwire: Arc<TripWire>,
687724
) -> (Result<FunctionResult>, FunctionCallId) {
688725
let res = Box::pin(self.call_function_with_expr_events(
689726
function_name,
@@ -730,7 +767,7 @@ impl BamlRuntime {
730767
collectors: Option<Vec<Arc<Collector>>>,
731768
env_vars: HashMap<String, String>,
732769
expr_tx: Option<mpsc::UnboundedSender<Vec<internal_baml_diagnostics::SerializedSpan>>>,
733-
cancel_tripwire: Option<stream_cancel::Tripwire>,
770+
cancel_tripwire: Arc<TripWire>,
734771
) -> (Result<FunctionResult>, FunctionCallId) {
735772
// baml_log::info!("env vars: {:#?}", env_vars.clone());
736773
baml_log::set_from_env(&env_vars).unwrap();
@@ -937,6 +974,7 @@ impl BamlRuntime {
937974
collectors: Option<Vec<Arc<Collector>>>,
938975
env_vars: HashMap<String, String>,
939976
expr_tx: Option<mpsc::UnboundedSender<Vec<SerializedSpan>>>,
977+
cancel_tripwire: Arc<TripWire>,
940978
) -> Result<FunctionResultStream> {
941979
baml_log::set_from_env(&env_vars).unwrap();
942980
self.inner.stream_function_impl(
@@ -947,6 +985,7 @@ impl BamlRuntime {
947985
#[cfg(not(target_arch = "wasm32"))]
948986
self.async_runtime.clone(),
949987
collectors.unwrap_or_default(),
988+
cancel_tripwire,
950989
)
951990
}
952991

@@ -959,6 +998,7 @@ impl BamlRuntime {
959998
cb: Option<&ClientRegistry>,
960999
collectors: Option<Vec<Arc<Collector>>>,
9611000
env_vars: HashMap<String, String>,
1001+
cancel_tripwire: Arc<TripWire>,
9621002
) -> Result<FunctionResultStream> {
9631003
self.stream_function_with_expr_events(
9641004
function_name,
@@ -969,6 +1009,7 @@ impl BamlRuntime {
9691009
collectors,
9701010
env_vars,
9711011
None,
1012+
cancel_tripwire,
9721013
)
9731014
}
9741015

‎engine/baml-runtime/src/runtime_methods/call_function.rs‎

Lines changed: 3 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -15,7 +15,6 @@ use internal_baml_core::{
1515
};
1616
use internal_baml_jinja::RenderedPrompt;
1717
use internal_llm_client::{AllowedRoleMetadata, ClientSpec};
18-
use stream_cancel::Tripwire;
1918

2019
use super::prepare_function::PreparedFunction;
2120
use crate::{
@@ -40,15 +39,15 @@ use crate::{
4039
tracingv2::storage::storage::{Collector, BAML_TRACER},
4140
type_builder::TypeBuilder,
4241
FunctionResult, FunctionResultStream, InternalRuntimeInterface, RenderCurlSettings,
43-
RuntimeContext, RuntimeContextManager,
42+
RuntimeContext, RuntimeContextManager, TripWire,
4443
};
4544

4645
impl InternalBamlRuntime {
4746
pub(crate) async fn call_function_impl<'ir>(
4847
&'ir self,
4948
prepared_func_call: PreparedFunction<'ir>,
5049
ctx: RuntimeContext,
51-
cancel_tripwire: Option<Tripwire>,
50+
cancel_tripwire: Arc<TripWire>,
5251
) -> Result<crate::FunctionResult> {
5352
let future = async {
5453
let renderer =
@@ -65,7 +64,7 @@ impl InternalBamlRuntime {
6564
&renderer,
6665
&baml_args,
6766
|s| renderer.parse(self.ir(), &ctx, s, false),
68-
cancel_tripwire,
67+
cancel_tripwire.trip_wire(),
6968
)
7069
.await;
7170

0 commit comments

Comments
 (0)