From 26b5cd57a72ba3d539d843dfae6eff94a1316bc3 Mon Sep 17 00:00:00 2001 From: Yujong Lee Date: Wed, 16 Sep 2026 10:37:11 -0700 Subject: [PATCH] refactor(core): let the plan own delivery and gate sync leaves on the request kind CallbackFamily::delivery is gone. plan_success, plan_failure and plan_request return a Dispatch carrying family, delivery and gate, and DispatchCursor takes that Dispatch plus CursorFacts. Delivery was never a property of the family: legacy runs failure_handler inline from @client but on the executor from dispatch_failure_handlers, and the streaming port will need both. object_target_eligible encodes the is_sync_request gate from success_handler and failure_handler: on an async SDK request the sync handler pass runs string integrations only, so CustomLogger and plain callable targets are not logged twice alongside their async methods. Parity-neutral for OCR because _is_sync_litellm_request does not know aocr, but required before the first async route lands. --- .../core/src/call_lifecycle/callbacks.rs | 262 +++++++++++++----- .../crates/core/src/call_lifecycle/mod.rs | 7 +- .../python-bridge/src/lifecycle/dispatch.rs | 63 ++++- .../crates/python-bridge/src/lifecycle/mod.rs | 23 +- litellm/rust_bridge/leaves.py | 8 + 5 files changed, 263 insertions(+), 100 deletions(-) diff --git a/litellm-rust/crates/core/src/call_lifecycle/callbacks.rs b/litellm-rust/crates/core/src/call_lifecycle/callbacks.rs index 82d33898bd0..c1b523cef6d 100644 --- a/litellm-rust/crates/core/src/call_lifecycle/callbacks.rs +++ b/litellm-rust/crates/core/src/call_lifecycle/callbacks.rs @@ -90,18 +90,6 @@ pub enum CallbackFamily { } impl CallbackFamily { - pub const fn delivery(self) -> Delivery { - match self { - Self::RequestPreCall | Self::RequestPostCall | Self::SyncFailure => Delivery::Inline, - Self::DeploymentPreCall - | Self::DeploymentPostCall - | Self::DeploymentFailure - | Self::AsyncFailure => Delivery::Await, - Self::SyncSuccess => Delivery::Worker, - Self::AsyncSuccess => Delivery::Background, - } - } - pub const fn dispatch_method(self) -> CallbackMethod { match self { Self::RequestPreCall => CallbackMethod::LogPreApiCall, @@ -178,12 +166,22 @@ pub enum ReleaseGate { } #[derive(Clone, Copy, Debug, PartialEq, Eq)] -pub struct SuccessDispatch { +pub struct Dispatch { pub family: CallbackFamily, pub delivery: Delivery, pub gate: ReleaseGate, } +impl Dispatch { + pub const fn immediate(family: CallbackFamily, delivery: Delivery) -> Self { + Self { + family, + delivery, + gate: ReleaseGate::Immediate, + } + } +} + #[derive(Clone, Debug, PartialEq, Eq)] pub struct SuccessFacts { pub asynchronous: bool, @@ -193,15 +191,15 @@ pub struct SuccessFacts { pub sync_target_kinds: Vec, } -pub fn plan_success(facts: &SuccessFacts) -> Vec { +pub fn plan_success(facts: &SuccessFacts) -> Vec { if !facts.asynchronous { - return vec![SuccessDispatch { + return vec![Dispatch { family: CallbackFamily::SyncSuccess, delivery: Delivery::Worker, gate: ReleaseGate::Immediate, }]; } - let background = (!facts.internal && !facts.fallbacks).then_some(SuccessDispatch { + let background = (!facts.internal && !facts.fallbacks).then_some(Dispatch { family: CallbackFamily::AsyncSuccess, delivery: Delivery::Background, gate: if facts.deferred { @@ -214,7 +212,7 @@ pub fn plan_success(facts: &SuccessFacts) -> Vec { .sync_target_kinds .iter() .any(|kind| kind.runs_sync_handler_for_async_call()) - .then_some(SuccessDispatch { + .then_some(Dispatch { family: CallbackFamily::SyncSuccess, delivery: Delivery::Worker, gate: ReleaseGate::Immediate, @@ -222,21 +220,45 @@ pub fn plan_success(facts: &SuccessFacts) -> Vec { background.into_iter().chain(worker).collect() } -pub fn plan_failure( - phase: HostPhase, - asynchronous: bool, - internal: bool, -) -> Option { +pub fn plan_failure(phase: HostPhase, asynchronous: bool, internal: bool) -> Option { if asynchronous && internal { return None; } match phase { - HostPhase::Failure => Some(CallbackFamily::SyncFailure), - HostPhase::AsyncFailure => Some(CallbackFamily::AsyncFailure), + HostPhase::Failure => Some(Dispatch::immediate( + CallbackFamily::SyncFailure, + Delivery::Inline, + )), + HostPhase::AsyncFailure => Some(Dispatch::immediate( + CallbackFamily::AsyncFailure, + Delivery::Await, + )), _ => None, } } +pub fn plan_request(family: CallbackFamily) -> Dispatch { + let delivery = match family { + CallbackFamily::RequestPreCall | CallbackFamily::RequestPostCall => Delivery::Inline, + _ => Delivery::Await, + }; + Dispatch::immediate(family, delivery) +} + +pub fn object_target_eligible( + sync_request: bool, + method: CallbackMethod, + kind: CallbackKind, +) -> bool { + match (method, kind) { + ( + CallbackMethod::LogSuccessEvent | CallbackMethod::LogFailureEvent, + CallbackKind::CustomLogger | CallbackKind::Callable { .. }, + ) => sync_request, + _ => true, + } +} + pub trait DispatchFacts { fn eligible(&mut self, target: CallbackId, method: CallbackMethod) -> bool; } @@ -264,21 +286,24 @@ enum Position { Complete { aborted: bool }, } +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct CursorFacts { + pub already_logged: bool, + pub stream: bool, + pub sync_request: bool, +} + pub struct DispatchCursor { - family: CallbackFamily, + dispatch: Dispatch, targets: Vec, - stream: bool, + facts: CursorFacts, position: Position, } impl DispatchCursor { - pub fn start( - family: CallbackFamily, - targets: Vec, - already_logged: bool, - stream: bool, - ) -> Self { - let position = if family.marker().is_some() && already_logged { + pub fn start(dispatch: Dispatch, targets: Vec, facts: CursorFacts) -> Self { + let family = dispatch.family; + let position = if family.marker().is_some() && facts.already_logged { Position::Complete { aborted: false } } else if family.prepares_logging() { Position::Prepare @@ -286,15 +311,23 @@ impl DispatchCursor { Position::Dispatch(0) }; Self { - family, + dispatch, targets, - stream, + facts, position, } } + pub fn object_target_eligible(&self, method: CallbackMethod, kind: CallbackKind) -> bool { + object_target_eligible(self.facts.sync_request, method, kind) + } + pub const fn family(&self) -> CallbackFamily { - self.family + self.dispatch.family + } + + pub const fn delivery(&self) -> Delivery { + self.dispatch.delivery } pub fn targets(&self) -> &[CallbackId] { @@ -303,7 +336,7 @@ impl DispatchCursor { pub fn accept(&mut self, outcome: InvocationOutcome) { if outcome == InvocationOutcome::Failed - && self.family.error_policy() == TargetErrorPolicy::Propagate + && self.dispatch.family.error_policy() == TargetErrorPolicy::Propagate { self.position = Position::Complete { aborted: true }; } @@ -317,7 +350,7 @@ impl DispatchCursor { return DispatchStep::PrepareLogging; } Position::Hook(index) => { - let Some(method) = self.family.hook_method() else { + let Some(method) = self.dispatch.family.hook_method() else { self.position = Position::Mark; continue; }; @@ -332,8 +365,10 @@ impl DispatchCursor { } Position::Mark => { self.position = Position::Dispatch(0); - match self.family.marker() { - Some(marker) if !self.stream => return DispatchStep::MarkLogged(marker), + match self.dispatch.family.marker() { + Some(marker) if !self.facts.stream => { + return DispatchStep::MarkLogged(marker); + } _ => continue, } } @@ -343,7 +378,7 @@ impl DispatchCursor { continue; }; self.position = Position::Dispatch(index + 1); - let method = self.family.dispatch_method(); + let method = self.dispatch.family.dispatch_method(); if facts.eligible(target, method) { return DispatchStep::Invoke(self.invocation(target, method)); } @@ -354,7 +389,7 @@ impl DispatchCursor { } fn after_prepare(&self) -> Position { - if self.family.hook_method().is_some() { + if self.dispatch.family.hook_method().is_some() { Position::Hook(0) } else { Position::Mark @@ -365,7 +400,7 @@ impl DispatchCursor { CallbackInvocation { target, method, - delivery: self.family.delivery(), + delivery: self.dispatch.delivery, } } } @@ -407,6 +442,34 @@ mod tests { values.iter().copied().map(CallbackId).collect() } + fn delivery_for(family: CallbackFamily) -> Delivery { + match family { + CallbackFamily::SyncSuccess => Delivery::Worker, + CallbackFamily::AsyncSuccess => Delivery::Background, + CallbackFamily::SyncFailure + | CallbackFamily::RequestPreCall + | CallbackFamily::RequestPostCall => Delivery::Inline, + _ => Delivery::Await, + } + } + + fn start( + family: CallbackFamily, + targets: Vec, + already_logged: bool, + stream: bool, + ) -> DispatchCursor { + DispatchCursor::start( + Dispatch::immediate(family, delivery_for(family)), + targets, + CursorFacts { + already_logged, + stream, + sync_request: true, + }, + ) + } + #[test] fn terminal_families_order_dynamic_before_global_and_keep_first_duplicate() { let combined = CallbackFamily::SyncSuccess.targets(&ids(&[3, 1, 4]), Some(&ids(&[1, 2]))); @@ -427,8 +490,7 @@ mod tests { #[test] fn success_runs_every_hook_before_any_dispatch_and_marks_between_passes() { - let mut cursor = - DispatchCursor::start(CallbackFamily::SyncSuccess, ids(&[1, 2]), false, false); + let mut cursor = start(CallbackFamily::SyncSuccess, ids(&[1, 2]), false, false); let steps = drain(&mut cursor, &mut AllEligible); let invocation = |target, method| { DispatchStep::Invoke(CallbackInvocation { @@ -453,8 +515,7 @@ mod tests { #[test] fn async_success_uses_async_leaf_methods_and_background_delivery() { - let mut cursor = - DispatchCursor::start(CallbackFamily::AsyncSuccess, ids(&[7]), false, false); + let mut cursor = start(CallbackFamily::AsyncSuccess, ids(&[7]), false, false); let steps = drain(&mut cursor, &mut AllEligible); let methods: Vec<_> = steps .iter() @@ -489,7 +550,7 @@ mod tests { CallbackMethod::AsyncLogFailureEvent, ), ] { - let mut cursor = DispatchCursor::start(family, ids(&[1, 2]), false, false); + let mut cursor = start(family, ids(&[1, 2]), false, false); let steps = drain(&mut cursor, &mut AllEligible); assert_eq!(steps[0], DispatchStep::PrepareLogging); assert!(matches!(steps[1], DispatchStep::MarkLogged(_))); @@ -510,8 +571,7 @@ mod tests { ] ); } - let mut cursor = - DispatchCursor::start(CallbackFamily::RequestPreCall, ids(&[1]), false, false); + let mut cursor = start(CallbackFamily::RequestPreCall, ids(&[1]), false, false); let steps = drain(&mut cursor, &mut AllEligible); assert_eq!( steps, @@ -528,14 +588,12 @@ mod tests { #[test] fn already_logged_marker_skips_the_whole_terminal_family_but_not_request_families() { - let mut cursor = - DispatchCursor::start(CallbackFamily::AsyncSuccess, ids(&[1]), true, false); + let mut cursor = start(CallbackFamily::AsyncSuccess, ids(&[1]), true, false); assert_eq!( cursor.next(&mut AllEligible), DispatchStep::Complete { aborted: false } ); - let mut cursor = - DispatchCursor::start(CallbackFamily::RequestPostCall, ids(&[1]), true, false); + let mut cursor = start(CallbackFamily::RequestPostCall, ids(&[1]), true, false); assert!(matches!( cursor.next(&mut AllEligible), DispatchStep::Invoke(_) @@ -544,7 +602,7 @@ mod tests { #[test] fn streaming_skips_the_marker_write_but_still_dispatches() { - let mut cursor = DispatchCursor::start(CallbackFamily::SyncSuccess, ids(&[1]), false, true); + let mut cursor = start(CallbackFamily::SyncSuccess, ids(&[1]), false, true); let steps = drain(&mut cursor, &mut AllEligible); assert!( !steps @@ -562,8 +620,7 @@ mod tests { #[test] fn ineligible_targets_are_skipped_per_method_without_affecting_others() { - let mut cursor = - DispatchCursor::start(CallbackFamily::SyncSuccess, ids(&[1, 2]), false, false); + let mut cursor = start(CallbackFamily::SyncSuccess, ids(&[1, 2]), false, false); let mut facts = Gate(|target, method| { !(target == CallbackId(1) && method == CallbackMethod::LoggingHook) && !(target == CallbackId(2) && method == CallbackMethod::LogSuccessEvent) @@ -586,8 +643,7 @@ mod tests { #[test] fn contained_failures_continue_and_propagating_failures_abort() { - let mut cursor = - DispatchCursor::start(CallbackFamily::SyncFailure, ids(&[1, 2]), false, false); + let mut cursor = start(CallbackFamily::SyncFailure, ids(&[1, 2]), false, false); assert_eq!(cursor.next(&mut AllEligible), DispatchStep::PrepareLogging); assert!(matches!( cursor.next(&mut AllEligible), @@ -606,7 +662,7 @@ mod tests { }) )); - let mut cursor = DispatchCursor::start( + let mut cursor = start( CallbackFamily::DeploymentPreCall, ids(&[1, 2]), false, @@ -634,7 +690,7 @@ mod tests { }); assert_eq!( plan, - [SuccessDispatch { + [Dispatch { family: CallbackFamily::SyncSuccess, delivery: Delivery::Worker, gate: ReleaseGate::Immediate, @@ -656,7 +712,7 @@ mod tests { }; assert_eq!( plan_success(&base), - [SuccessDispatch { + [Dispatch { family: CallbackFamily::AsyncSuccess, delivery: Delivery::Background, gate: ReleaseGate::Immediate, @@ -673,12 +729,12 @@ mod tests { assert_eq!( plan_success(&with_external), [ - SuccessDispatch { + Dispatch { family: CallbackFamily::AsyncSuccess, delivery: Delivery::Background, gate: ReleaseGate::Deferred, }, - SuccessDispatch { + Dispatch { family: CallbackFamily::SyncSuccess, delivery: Delivery::Worker, gate: ReleaseGate::Immediate, @@ -692,7 +748,7 @@ mod tests { }; assert_eq!( plan_success(&internal_or_fallback), - [SuccessDispatch { + [Dispatch { family: CallbackFamily::SyncSuccess, delivery: Delivery::Worker, gate: ReleaseGate::Immediate, @@ -719,29 +775,83 @@ mod tests { fn failure_families_follow_the_phase_and_skip_internal_async_calls() { assert_eq!( plan_failure(HostPhase::Failure, false, true), - Some(CallbackFamily::SyncFailure) + Some(Dispatch::immediate( + CallbackFamily::SyncFailure, + Delivery::Inline + )) ); assert_eq!( plan_failure(HostPhase::AsyncFailure, true, false), - Some(CallbackFamily::AsyncFailure) + Some(Dispatch::immediate( + CallbackFamily::AsyncFailure, + Delivery::Await + )) ); assert_eq!(plan_failure(HostPhase::Failure, true, true), None); assert_eq!(plan_failure(HostPhase::Success, false, false), None); } #[test] - fn delivery_is_a_property_of_the_family_not_of_the_callable() { - assert_eq!(CallbackFamily::RequestPreCall.delivery(), Delivery::Inline); + fn delivery_is_selected_by_the_plan_and_carried_on_every_invocation() { + let mut cursor = DispatchCursor::start( + Dispatch::immediate(CallbackFamily::SyncFailure, Delivery::Worker), + ids(&[1]), + CursorFacts { + already_logged: false, + stream: false, + sync_request: true, + }, + ); + let deliveries: Vec<_> = drain(&mut cursor, &mut AllEligible) + .into_iter() + .filter_map(|step| match step { + DispatchStep::Invoke(invocation) => Some(invocation.delivery), + _ => None, + }) + .collect(); + assert_eq!(deliveries, [Delivery::Worker]); assert_eq!( - CallbackFamily::DeploymentPreCall.delivery(), + plan_request(CallbackFamily::RequestPreCall).delivery, + Delivery::Inline + ); + assert_eq!( + plan_request(CallbackFamily::DeploymentPreCall).delivery, Delivery::Await ); - assert_eq!(CallbackFamily::SyncSuccess.delivery(), Delivery::Worker); - assert_eq!( - CallbackFamily::AsyncSuccess.delivery(), - Delivery::Background + } + + #[test] + fn sync_leaf_methods_skip_object_targets_on_async_requests() { + let asynchronous = DispatchCursor::start( + Dispatch::immediate(CallbackFamily::SyncSuccess, Delivery::Worker), + ids(&[1]), + CursorFacts { + already_logged: false, + stream: false, + sync_request: false, + }, + ); + for kind in [ + CallbackKind::CustomLogger, + CallbackKind::Callable { internal: false }, + ] { + assert!(!asynchronous.object_target_eligible(CallbackMethod::LogSuccessEvent, kind)); + assert!(!asynchronous.object_target_eligible(CallbackMethod::LogFailureEvent, kind)); + assert!(asynchronous.object_target_eligible(CallbackMethod::LoggingHook, kind)); + assert!( + asynchronous.object_target_eligible(CallbackMethod::AsyncLogSuccessEvent, kind) + ); + } + assert!(asynchronous.object_target_eligible( + CallbackMethod::LogSuccessEvent, + CallbackKind::Named { known: false } + )); + let synchronous = start(CallbackFamily::SyncSuccess, ids(&[1]), false, false); + assert!( + synchronous.object_target_eligible( + CallbackMethod::LogSuccessEvent, + CallbackKind::CustomLogger + ) ); - assert_eq!(CallbackFamily::SyncFailure.delivery(), Delivery::Inline); - assert_eq!(CallbackFamily::AsyncFailure.delivery(), Delivery::Await); } } diff --git a/litellm-rust/crates/core/src/call_lifecycle/mod.rs b/litellm-rust/crates/core/src/call_lifecycle/mod.rs index f5625ac6537..8788e0926b0 100644 --- a/litellm-rust/crates/core/src/call_lifecycle/mod.rs +++ b/litellm-rust/crates/core/src/call_lifecycle/mod.rs @@ -7,9 +7,10 @@ pub mod registration; pub mod types; pub use callbacks::{ - CallbackFamily, CallbackId, CallbackInvocation, CallbackKind, CallbackMethod, Delivery, - DispatchCursor, DispatchFacts, DispatchStep, InvocationOutcome, LoggedMarker, ReleaseGate, - SuccessDispatch, SuccessFacts, TargetErrorPolicy, plan_failure, plan_success, + CallbackFamily, CallbackId, CallbackInvocation, CallbackKind, CallbackMethod, CursorFacts, + Delivery, Dispatch, DispatchCursor, DispatchFacts, DispatchStep, InvocationOutcome, + LoggedMarker, ReleaseGate, SuccessFacts, TargetErrorPolicy, object_target_eligible, + plan_failure, plan_request, plan_success, }; pub use types::{ CallLifecycleContext, CallLifecyclePhase, CallLifecyclePhaseTiming, CallLifecycleRequest, diff --git a/litellm-rust/crates/python-bridge/src/lifecycle/dispatch.rs b/litellm-rust/crates/python-bridge/src/lifecycle/dispatch.rs index 706a31ba7d7..ca204d1d01b 100644 --- a/litellm-rust/crates/python-bridge/src/lifecycle/dispatch.rs +++ b/litellm-rust/crates/python-bridge/src/lifecycle/dispatch.rs @@ -1,6 +1,7 @@ use litellm_core::call_lifecycle::{ - CallbackFamily, CallbackId, CallbackInvocation, CallbackKind, CallbackMethod, Delivery, - DispatchCursor, DispatchFacts, DispatchStep, InvocationOutcome, LoggedMarker, + CallbackFamily, CallbackId, CallbackInvocation, CallbackKind, CallbackMethod, CursorFacts, + Delivery, Dispatch, DispatchCursor, DispatchFacts, DispatchStep, InvocationOutcome, + LoggedMarker, object_target_eligible, plan_request, }; use pyo3::exceptions::{PyBaseException, PyException, PyRuntimeError}; use pyo3::gc::{PyTraverseError, PyVisit}; @@ -125,12 +126,13 @@ pub(super) struct Job { pub logger: PythonLogger, pub targets: Targets, pub ids: Vec, - pub family: CallbackFamily, + pub dispatch: Dispatch, pub response: Option>, pub error: Option>, pub start: Py, pub end: Py, pub stream: bool, + pub sync_request: bool, } impl Job { @@ -144,7 +146,7 @@ impl Job { } fn family_name(&self) -> &'static str { - match self.family { + match self.dispatch.family { CallbackFamily::SyncSuccess => "sync_success", CallbackFamily::AsyncSuccess => "async_success", CallbackFamily::SyncFailure => "sync_failure", @@ -154,7 +156,7 @@ impl Job { } fn outcome(&self) -> Outcome { - match self.family { + match self.dispatch.family { CallbackFamily::SyncSuccess | CallbackFamily::AsyncSuccess => Outcome::Success, _ => Outcome::Failure, } @@ -171,6 +173,9 @@ impl DispatchFacts for Eligibility<'_, '_> { fn eligible(&mut self, target: CallbackId, method: CallbackMethod) -> bool { let object = self.job.targets.object(self.py, target); let kind = self.job.targets.kind(target); + if !object_target_eligible(self.job.sync_request, method, kind) { + return false; + } let result = match method { CallbackMethod::LoggingHook | CallbackMethod::AsyncLoggingHook => { if kind != CallbackKind::CustomLogger { @@ -227,14 +232,22 @@ pub(super) struct Runner { impl Runner { pub(super) fn start(py: Python<'_>, job: Job) -> PyResult { let leaves = leaves(py)?; - let already = match job.family.marker() { + let already_logged = match job.dispatch.family.marker() { Some(marker) => leaves .getattr("already_logged")? .call1((job.logger.object(py), marker.key()))? .extract::()?, None => false, }; - let cursor = DispatchCursor::start(job.family, job.ids.clone(), already, job.stream); + let cursor = DispatchCursor::start( + job.dispatch, + job.ids.clone(), + CursorFacts { + already_logged, + stream: job.stream, + sync_request: job.sync_request, + }, + ); let result = job.response.as_ref().map(|value| value.clone_ref(py)); Ok(Self { job, @@ -410,7 +423,7 @@ impl Runner { Delivery::Await | Delivery::Background => Err(PyRuntimeError::new_err(format!( "{:?} leaf for {:?} returned a non-awaitable {}", invocation.method, - self.job.family, + self.job.dispatch.family, value.get_type().name()? ))), } @@ -552,7 +565,15 @@ pub(super) struct RequestJob<'a> { pub(super) fn dispatch_request(py: Python<'_>, job: RequestJob<'_>) -> PyResult<()> { let leaves = leaves(py)?; let (targets, ids) = family_targets(py, job.logger, job.family)?; - let mut cursor = DispatchCursor::start(job.family, ids, false, false); + let mut cursor = DispatchCursor::start( + plan_request(job.family), + ids, + CursorFacts { + already_logged: false, + stream: false, + sync_request: true, + }, + ); let logger = job.logger.object(py); let event = match job.family { CallbackFamily::RequestPreCall => "pre_api_call", @@ -712,7 +733,15 @@ impl DeploymentBody { Ok(Self { logger: logger.clone_ref(py), targets, - cursor: DispatchCursor::start(family, ids, false, false), + cursor: DispatchCursor::start( + plan_request(family), + ids, + CursorFacts { + already_logged: false, + stream: false, + sync_request: true, + }, + ), event, call_type, current, @@ -872,13 +901,23 @@ target = CustomLogger() let list = PyList::new(py, [&target]).unwrap().into_any(); let (targets, ids) = Targets::read(py, &[list]).unwrap(); let ids: Vec = ids.into_iter().flatten().collect(); + let dispatch = Dispatch::immediate(family, Delivery::Worker); Runner { - cursor: DispatchCursor::start(family, ids.clone(), false, false), + cursor: DispatchCursor::start( + dispatch, + ids.clone(), + CursorFacts { + already_logged: false, + stream: false, + sync_request: true, + }, + ), job: Job { logger: logger.extract().unwrap(), targets, ids, - family, + dispatch, + sync_request: true, response: Some(target.clone().unbind()), error: None, start: py.None(), diff --git a/litellm-rust/crates/python-bridge/src/lifecycle/mod.rs b/litellm-rust/crates/python-bridge/src/lifecycle/mod.rs index fa07262fe7b..bf6c537e240 100644 --- a/litellm-rust/crates/python-bridge/src/lifecycle/mod.rs +++ b/litellm-rust/crates/python-bridge/src/lifecycle/mod.rs @@ -13,7 +13,7 @@ use litellm_core::call_lifecycle::host::{ HostCall as NativeCall, HostCallStep as NativeCallStep, HostFailure, HostPhase, HostStep, }; use litellm_core::call_lifecycle::{ - CallbackFamily, Delivery, ReleaseGate, SuccessFacts, plan_failure, plan_success, + CallbackFamily, Delivery, Dispatch, ReleaseGate, SuccessFacts, plan_failure, plan_success, }; mod arguments; @@ -479,14 +479,19 @@ impl PythonCallState { } } - fn job(&self, py: Python<'_>, family: CallbackFamily) -> PyResult { + fn job(&self, py: Python<'_>, selected: Dispatch) -> PyResult { let logger = self.logger()?; - let (targets, ids) = dispatch::family_targets(py, logger, family)?; + let (targets, ids) = dispatch::family_targets(py, logger, selected.family)?; + let sync_request = dispatch::leaves(py)? + .getattr("is_sync_request")? + .call1((logger.object(py),))? + .extract()?; Ok(dispatch::Job { logger: logger.clone_ref(py), targets, ids, - family, + dispatch: selected, + sync_request, response: self.response.as_ref().map(|value| value.clone_ref(py)), error: self.error.as_ref().map(|value| value.clone_ref(py)), start: self.start.clone_ref(py), @@ -518,7 +523,7 @@ impl PythonCallState { sync_target_kinds: sync_targets.kinds(&sync_ids), }; for selected in plan_success(&facts) { - let runner = dispatch::Runner::start(py, self.job(py, selected.family)?)?; + let runner = dispatch::Runner::start(py, self.job(py, selected)?)?; match (selected.delivery, selected.gate) { (Delivery::Worker, _) => { let job = Py::new(py, dispatch::WorkerJob::new(runner))?; @@ -559,14 +564,14 @@ impl PythonCallState { } else { HostPhase::Failure }; - let Some(family) = plan_failure(phase, self.asynchronous, self.internal) else { + let Some(selected) = plan_failure(phase, self.asynchronous, self.internal) else { return Ok(None); }; if self.supplied { - return compat::dispatch_failure(py, self, family); + return compat::dispatch_failure(py, self, selected.family); } - let mut runner = dispatch::Runner::start(py, self.job(py, family)?)?; - match family.delivery() { + let mut runner = dispatch::Runner::start(py, self.job(py, selected)?)?; + match selected.delivery { Delivery::Inline => match runner.resume(py, None)? { dispatch::Step::Done => Ok(None), dispatch::Step::Await(_) => Err(missing_state()), diff --git a/litellm/rust_bridge/leaves.py b/litellm/rust_bridge/leaves.py index 7707a8938f1..d10a348c306 100644 --- a/litellm/rust_bridge/leaves.py +++ b/litellm/rust_bridge/leaves.py @@ -525,6 +525,14 @@ async def async_logging_hook(logger: Logging, callback: CustomLogger, result: ob return replaced +def is_sync_request(logger: Logging) -> bool: + from litellm.litellm_core_utils.litellm_logging import Logging as LoggingClass + + params: Final = _details_of(_logger(logger).model_call_details.get("litellm_params") or _EMPTY) + decide: Final = _call_of(LoggingClass._is_sync_litellm_request) # pyright: ignore[reportPrivateUsage] # legacy predicate + return decide(params) is True + + def mark_logged(logger: Logging, marker: str) -> None: _logger(logger).model_call_details[marker] = True