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.

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<CallbackKind>,
 }
 
-pub fn plan_success(facts: &SuccessFacts) -> Vec<SuccessDispatch> {
+pub fn plan_success(facts: &SuccessFacts) -> Vec<Dispatch> {
     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<SuccessDispatch> {
         .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<SuccessDispatch> {
     background.into_iter().chain(worker).collect()
 }
 
-pub fn plan_failure(
-    phase: HostPhase,
-    asynchronous: bool,
-    internal: bool,
-) -> Option<CallbackFamily> {
+pub fn plan_failure(phase: HostPhase, asynchronous: bool, internal: bool) -> Option<Dispatch> {
     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<CallbackId>,
-    stream: bool,
+    facts: CursorFacts,
     position: Position,
 }
 
 impl DispatchCursor {
-    pub fn start(
-        family: CallbackFamily,
-        targets: Vec<CallbackId>,
-        already_logged: bool,
-        stream: bool,
-    ) -> Self {
-        let position = if family.marker().is_some() && already_logged {
+    pub fn start(dispatch: Dispatch, targets: Vec<CallbackId>, 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<CallbackId>,
+        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);
     }
 }

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,

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<CallbackId>,
-    pub family: CallbackFamily,
+    pub dispatch: Dispatch,
     pub response: Option<Py<PyAny>>,
     pub error: Option<Py<PyBaseException>>,
     pub start: Py<PyAny>,
     pub end: Py<PyAny>,
     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<Self> {
         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::<bool>()?,
             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<CallbackId> = 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(),

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<dispatch::Job> {
+    fn job(&self, py: Python<'_>, selected: Dispatch) -> PyResult<dispatch::Job> {
         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()),

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