>>;
type StartMachine = Box<
dyn FnOnce(Python<'_>, &Bound<'_, PyDict>,
::Request) -> PyResult
+ Send
+ Sync,
>;
+type Body = Coroutine>>;
pub struct CallOptions {
pub asynchronous: bool,
- pub lifecycle: PythonLifecycle,
+ pub steps: StepTypes,
pub observers: Option,
}
impl CallOptions {
- pub fn new(asynchronous: bool, lifecycle: PythonLifecycle) -> Self {
+ pub fn new(asynchronous: bool, steps: StepTypes) -> Self {
Self {
asynchronous,
- lifecycle,
+ steps,
observers: None,
}
}
}
-enum Stage {
- Begin,
- Call,
- Streaming,
- AfterSuccess,
- Succeeded(Py),
- Failed(Py),
+fn is_cancellation(py: Python<'_>, error: &PyErr) -> bool {
+ !error.is_instance_of::(py)
}
-enum EventNext {
- Started,
- Emitted(Reply<()>),
- Terminal,
+struct Observation {
+ observers: Option,
+ observing: bool,
+ started_at: f64,
+ ended_at: Option,
+ terminal: Option,
}
-enum Pending {
- Host,
- Native,
- Arguments(HookResume>),
- Wire(HookResume>, Reply),
- Response(HookResume>),
- Event(HookResume, EventNext),
- Consumer(Reply>),
-}
+impl Observation {
+ fn timing(&self) -> Timing {
+ Timing {
+ start_time: self.started_at,
+ end_time: self.ended_at.unwrap_or_else(epoch_seconds),
+ }
+ }
-/// A route answer as the driver resumes on it: a Python exception interrupts the call as
-/// raised, a native rejection resumes the machine with it.
-fn answered(answer: Result<(), InvokeError>) -> PyResult> {
- match answer {
- Ok(()) => Ok(Ok(())),
- Err(InvokeError::Native(error)) => Ok(Err(error)),
- Err(InvokeError::Python(error)) => Err(error),
+ fn emit(&self, event: &PythonCallEvent<'_>) {
+ if let Some(observers) = &self.observers {
+ observers.emit(event.snapshot());
+ }
+ }
+
+ fn retain_terminal(&mut self, event: &PythonCallEvent<'_>) {
+ self.terminal = self.observers.as_ref().map(|_| event.snapshot());
+ }
+
+ fn finish(&mut self) {
+ if let Some(event) = self.terminal.take()
+ && let Some(observers) = &self.observers
+ {
+ observers.emit(event);
+ }
+ self.observing = false;
}
}
-enum Next> {
- Return(ExecutionStep),
- Continue(NativePoll>),
+/// What the running call shares with its handle: the binding and hooks the handle
+/// traverses and closes, and the Python values the call retains across a suspension.
+struct Parts {
+ binding: H,
+ hooks: L,
+ arguments: Option>,
+ interrupted: Option>,
+ selected: Option>,
+ observation: Observation,
}
-struct PythonDriver
+type SharedParts = Arc>>;
+
+enum Setup {
+ Host(PyErr),
+ Decode(InvokeError),
+}
+
+/// The route's public exception for a native failure. When classification itself fails,
+/// that failure is raised with the native error's text as its `__context__`.
+fn classified(py: Python<'_>, binding: &H, error: ErrorOf) -> PyErr {
+ let native = error.to_string();
+ let classifier_error = match binding.map_error(py, error) {
+ Ok(failure) => return failure.into(),
+ Err(classifier_error) => classifier_error,
+ };
+ classifier_error.set_context(py, Some(PyRuntimeError::new_err(native)));
+ classifier_error
+}
+
+/// The call as one async body. Every hook step and host call is awaited in place on the
+/// caller's task; the parts lock is never held across an await.
+struct CallBody
+where
+ H: PythonBinding,
+ M: Machine,
+{
+ caller: PythonCaller,
+ parts: SharedParts,
+ native: NativeMachine,
+ start: Option, M>>,
+ streaming: bool,
+}
+
+impl CallBody
where
L: PythonCallHooks + 'static,
- H: PythonBinding + PythonHostCalls,
+ H: PythonBinding + PythonHostCalls + 'static,
M: Machine + 'static,
M::Complete: Into>>,
{
- binding: H,
- hooks: L,
- native: NativeMachine,
- start: Option, M>>,
- closed: bool,
- arguments: Option>,
- started_at: f64,
- ended_at: Option,
- stage: Stage,
- pending: Option>,
- interrupted: Option>,
- observers: Option,
- observing: bool,
- terminal_observation: Option,
+ fn parts(&self) -> MutexGuard<'_, Parts> {
+ self.parts.lock()
+ }
+
+ async fn run(mut self) -> PyResult> {
+ match self.begin().await {
+ Ok(()) => self.execute().await,
+ Err(Setup::Host(error)) => self.failure(error, FailureOrigin::Host).await,
+ Err(Setup::Decode(InvokeError::Python(error))) => {
+ self.failure(error, FailureOrigin::Call).await
+ }
+ Err(Setup::Decode(InvokeError::Native(error))) => self.machine_failed(error).await,
+ }
+ }
+
+ async fn begin(&mut self) -> Result<(), Setup>> {
+ let started_at = epoch_seconds();
+ let started = {
+ let mut parts = self.parts();
+ parts.observation.started_at = started_at;
+ parts.observation.observing = true;
+ let event = PythonCallEvent::Started {
+ start_time: started_at,
+ };
+ parts.observation.emit(&event);
+ Python::attach(|py| parts.hooks.on_event(py, event)).map_err(Setup::Host)?
+ };
+ self.settle(started).await.map_err(Setup::Host)?;
+ let arguments = self
+ .parts()
+ .arguments
+ .take()
+ .ok_or_else(|| Setup::Host(Error::MissingState.into()))?;
+ let prepare = {
+ let mut parts = self.parts();
+ Python::attach(|py| parts.hooks.prepare_arguments(py, arguments, started_at))
+ .map_err(Setup::Host)?
+ };
+ let prepared = self.settle(prepare).await.map_err(Setup::Host)?;
+ let request = {
+ let mut parts = self.parts();
+ let Parts {
+ binding,
+ hooks,
+ arguments,
+ ..
+ } = &mut *parts;
+ Python::attach(|py| {
+ hooks
+ .arguments_prepared(py, &prepared)
+ .map_err(Setup::Host)?;
+ let decoded = binding.decode_request(py, prepared.bind(py));
+ *arguments = Some(prepared);
+ decoded.map_err(Setup::Decode)
+ })?
+ };
+ let start = self
+ .start
+ .take()
+ .ok_or_else(|| Setup::Host(Error::MissingState.into()))?;
+ let machine = {
+ let parts = self.parts();
+ let arguments = parts
+ .arguments
+ .as_ref()
+ .ok_or_else(|| Setup::Host(Error::MissingState.into()))?;
+ Python::attach(|py| start(py, arguments.bind(py), request)).map_err(Setup::Host)?
+ };
+ self.native.start(machine);
+ Ok(())
+ }
+
+ /// Runs one hook step to its value, handing each awaitable it produces to the caller.
+ async fn settle(&self, step: HookStep) -> PyResult {
+ let mut step = step;
+ loop {
+ let (awaitable, resume) = match step {
+ HookStep::Ready(value) => return Ok(value),
+ HookStep::Await(awaitable, resume) => (awaitable, resume),
+ };
+ let answer = self.caller.await_python(awaitable).await;
+ step = {
+ let mut parts = self.parts();
+ Python::attach(|py| resume(&mut parts.hooks, py, answer))?
+ };
+ }
+ }
+
+ async fn execute(&mut self) -> PyResult> {
+ let mut interruption: Interruption = None;
+ loop {
+ let poll = Python::attach(|py| self.native.resume(py, interruption.take()))?;
+ let result = match poll {
+ NativePoll::Ready(result) => result,
+ NativePoll::Suspend(awaitable) => match self.caller.await_python(awaitable).await {
+ Ok(_) => self.native.take_result()?,
+ Err(error) if self.parts().interrupted.is_none() => {
+ interruption = Some(self.interrupt(error));
+ continue;
+ }
+ Err(error) => return Err(error),
+ },
+ };
+ let op = match result {
+ Ok(MachineStep::Suspended(op)) => op,
+ Ok(MachineStep::Complete(completion)) => {
+ return self.completed(completion).await;
+ }
+ Err(error) => return self.machine_failed(error).await,
+ };
+ interruption = match self.answer(op).await {
+ Ok(()) => None,
+ Err(InvokeError::Native(error)) => Some(HostFailure::Error(error)),
+ Err(InvokeError::Python(error)) => Some(self.interrupt(error)),
+ };
+ }
+ }
+
+ fn interrupt(&self, error: PyErr) -> HostFailure> {
+ Python::attach(|py| {
+ let cancelled = is_cancellation(py, &error);
+ let native = H::host_error(&error);
+ self.parts().interrupted = Some(error.into_value(py));
+ if cancelled {
+ HostFailure::Cancelled(native)
+ } else {
+ HostFailure::Error(native)
+ }
+ })
+ }
+
+ /// Answers one machine op. A Python error interrupts the call as raised, a native
+ /// rejection resumes the machine with it.
+ async fn answer(
+ &mut self,
+ op: HostRequest>,
+ ) -> Result<(), InvokeError>> {
+ match op {
+ HostRequest::HostCall(call) => self.host_call(call).await,
+ HostRequest::Intercept(InterceptRequest::BeforeProviderRequest {
+ wire,
+ context,
+ reply,
+ }) => {
+ let step = {
+ let mut parts = self.parts();
+ Python::attach(|py| parts.hooks.before_provider_request(py, wire, &context))?
+ };
+ let wire = self.settle(step).await?;
+ reply.send(*wire);
+ Ok(())
+ }
+ HostRequest::Intercept(InterceptRequest::ResultReady { facts, reply }) => {
+ self.notify(PythonCallEvent::Execution(ExecutionEvent::ResultReady {
+ facts,
+ }))
+ .await?;
+ reply.send(());
+ Ok(())
+ }
+ HostRequest::Intercept(InterceptRequest::AfterProviderResponse { raw, reply }) => {
+ self.notify(PythonCallEvent::Execution(
+ ExecutionEvent::ProviderResponseReceived { raw: &raw },
+ ))
+ .await?;
+ reply.send(());
+ Ok(())
+ }
+ HostRequest::Stream(StreamDelivery::Open(head, reply)) => {
+ self.streaming = true;
+ let head = {
+ let mut parts = self.parts();
+ let Parts { binding, hooks, .. } = &mut *parts;
+ Python::attach(|py| {
+ let head = binding.encode_stream_head(py, head)?;
+ hooks.on_stream_open(py, &head)?;
+ Ok::<_, PyErr>(head)
+ })?
+ };
+ reply.send(self.caller.open_stream(head).await);
+ Ok(())
+ }
+ HostRequest::Stream(StreamDelivery::Chunk(chunk, reply)) => {
+ let chunk = {
+ let mut parts = self.parts();
+ let Parts { binding, hooks, .. } = &mut *parts;
+ Python::attach(|py| {
+ let chunk = binding.encode_chunk(py, chunk)?;
+ hooks.on_stream_chunk(py, &chunk)?;
+ Ok::<_, PyErr>(chunk)
+ })?
+ };
+ reply.send(self.caller.yield_chunk(chunk).await);
+ Ok(())
+ }
+ }
+ }
+
+ async fn host_call(
+ &self,
+ call: as Protocol>::HostCall,
+ ) -> Result<(), InvokeError>> {
+ let mut pending = {
+ let mut parts = self.parts();
+ Python::attach(|py| parts.binding.begin_host_call(py, call))?
+ };
+ while let Some(awaitable) = pending {
+ let result = self.caller.await_python(awaitable).await;
+ pending = {
+ let mut parts = self.parts();
+ Python::attach(|py| parts.binding.resume_host_call(py, result))?
+ };
+ }
+ Ok(())
+ }
+
+ async fn notify(&self, event: PythonCallEvent<'_>) -> PyResult<()> {
+ let step = {
+ let mut parts = self.parts();
+ parts.observation.emit(&event);
+ Python::attach(|py| parts.hooks.on_event(py, event))?
+ };
+ self.settle(step).await
+ }
+
+ async fn completed(
+ &mut self,
+ completion: HostedCompletion>,
+ ) -> PyResult> {
+ self.parts().observation.ended_at = Some(epoch_seconds());
+ let response = match completion {
+ HostedCompletion::Complete(response) => response,
+ HostedCompletion::StreamEnded | HostedCompletion::Detached => {
+ let none = Python::attach(|py| py.None());
+ return self.succeeded(none).await;
+ }
+ };
+ let encoded = {
+ let mut parts = self.parts();
+ Python::attach(|py| parts.binding.encode_response(py, response))
+ };
+ let public = match encoded {
+ Ok(public) => public,
+ Err(error) => return self.failure(error, FailureOrigin::Call).await,
+ };
+ if self.streaming {
+ return self.succeeded(public).await;
+ }
+ let transform = {
+ let mut parts = self.parts();
+ let timing = parts.observation.timing();
+ Python::attach(|py| parts.hooks.transform_response(py, public, timing))
+ };
+ let step = match transform {
+ Ok(step) => step,
+ Err(error) => return self.failure(error, FailureOrigin::Host).await,
+ };
+ match self.settle(step).await {
+ Ok(response) => self.succeeded(response).await,
+ Err(error) => self.failure(error, FailureOrigin::Host).await,
+ }
+ }
+
+ async fn machine_failed(&mut self, error: ErrorOf) -> PyResult> {
+ let error = {
+ let mut parts = self.parts();
+ parts.observation.ended_at.get_or_insert_with(epoch_seconds);
+ Python::attach(|py| match parts.interrupted.take() {
+ Some(retained) => PyErr::from_value(retained.into_bound(py).into_any()),
+ None => classified(py, &parts.binding, error),
+ })
+ };
+ self.failure(error, FailureOrigin::Call).await
+ }
+
+ /// The terminal hooks may suspend. Until they finish, the selected response is held
+ /// only in the traversed parts, so a cycle through it stays collectable.
+ async fn succeeded(&mut self, response: Py) -> PyResult> {
+ let step = {
+ let mut parts = self.parts();
+ let timing = parts.observation.timing();
+ let event = PythonCallEvent::Succeeded {
+ timing,
+ response: &response,
+ };
+ parts.observation.retain_terminal(&event);
+ let step = Python::attach(|py| parts.hooks.on_event(py, event))?;
+ parts.selected = Some(response);
+ step
+ };
+ self.settle(step).await?;
+ let mut parts = self.parts();
+ parts.observation.finish();
+ parts
+ .selected
+ .take()
+ .ok_or_else(|| Error::MissingState.into())
+ }
+
+ /// As [`Self::succeeded`], with the selected exception held as its value and raised
+ /// again once the hooks finish.
+ async fn failure(&mut self, error: PyErr, origin: FailureOrigin) -> PyResult> {
+ let step = {
+ let mut parts = self.parts();
+ parts.observation.ended_at.get_or_insert_with(epoch_seconds);
+ let timing = parts.observation.timing();
+ Python::attach(|py| {
+ if is_cancellation(py, &error) {
+ return Err(error);
+ }
+ let event = PythonCallEvent::Failed {
+ timing,
+ origin,
+ error: &error,
+ };
+ parts.observation.retain_terminal(&event);
+ let step = parts.hooks.on_event(py, event)?;
+ parts.selected = Some(error.into_value(py).into_any());
+ Ok(step)
+ })?
+ };
+ self.settle(step).await?;
+ let mut parts = self.parts();
+ parts.observation.finish();
+ let selected = parts.selected.take().ok_or(Error::MissingState)?;
+ Err(Python::attach(|py| {
+ PyErr::from_value(selected.into_bound(py))
+ }))
+ }
+}
+
+struct CallDriver
+where
+ H: PythonBinding,
+ L: PythonCallHooks,
+{
+ body: Mutex,
+ reply: Option>,
+ parts: SharedParts,
+}
+
+impl CallDriver
+where
+ H: PythonBinding + 'static,
+ L: PythonCallHooks + 'static,
+{
+ /// A failure the body raised past its terminal hooks, or a cancellation, is the last
+ /// thing the observers hear about the call.
+ fn observe_host_failure(&self, error: &PyErr) {
+ let mut parts = self.parts.lock();
+ if !parts.observation.observing {
+ return;
+ }
+ let timing = parts.observation.timing();
+ Python::attach(|py| {
+ let event = if is_cancellation(py, error) {
+ PythonCallEvent::Cancelled { timing }
+ } else {
+ PythonCallEvent::Failed {
+ timing,
+ origin: FailureOrigin::Host,
+ error,
+ }
+ };
+ parts.observation.emit(&event);
+ });
+ parts.observation.observing = false;
+ parts.observation.terminal = None;
+ }
+}
+
+impl ExecutionBody for CallDriver
+where
+ H: PythonBinding + 'static,
+ L: PythonCallHooks + 'static,
+{
+ fn resume(&mut self, answer: Option) -> PyResult {
+ match (self.reply.take(), answer) {
+ (Some(reply), Some(answer)) => reply.send(answer),
+ (None, None) => {}
+ (Some(_), None) | (None, Some(_)) => return Err(Error::MissingState.into()),
+ }
+ let step = {
+ let body = self.body.get_mut();
+ let mut resume = pin!(body.resume());
+ match resume
+ .as_mut()
+ .poll(&mut Context::from_waker(Waker::noop()))
+ {
+ Poll::Ready(step) => step,
+ Poll::Pending => return Err(Error::MissingState.into()),
+ }
+ };
+ match step {
+ Ok(CoroutineState::Yielded(Yielded { suspension, reply })) => {
+ self.reply = Some(reply);
+ Ok(ExecutionStep::Suspended(suspension))
+ }
+ Ok(CoroutineState::Complete(Ok(value))) => Ok(ExecutionStep::Return(value)),
+ Ok(CoroutineState::Complete(Err(error))) => {
+ self.observe_host_failure(&error);
+ Err(error)
+ }
+ Err(_) => Err(Error::MissingState.into()),
+ }
+ }
+
+ fn traverse(&self, visit: &PyVisit<'_>) -> Result<(), PyTraverseError> {
+ let Some(parts) = self.parts.try_lock() else {
+ return Ok(());
+ };
+ parts.binding.traverse(visit)?;
+ parts.hooks.traverse(visit)?;
+ visit.call(&parts.arguments)?;
+ visit.call(&parts.interrupted)?;
+ visit.call(&parts.selected)
+ }
+}
+
+impl Drop for CallDriver
+where
+ H: PythonBinding,
+ L: PythonCallHooks,
+{
+ fn drop(&mut self) {
+ self.body.get_mut().cancel();
+ self.reply = None;
+ let mut parts = self.parts.lock();
+ Python::attach(|py| {
+ if parts.observation.observing {
+ let timing = parts.observation.timing();
+ parts
+ .observation
+ .emit(&PythonCallEvent::Cancelled { timing });
+ parts.observation.observing = false;
+ }
+ parts.hooks.close(py);
+ parts.binding.close(py);
+ });
+ }
}
pub fn run_call(
@@ -132,2412 +605,38 @@ where
M: Machine + 'static,
M::Complete: Into>>,
{
- let mut driver = PythonDriver {
+ let parts = Arc::new(Mutex::new(Parts {
binding,
hooks,
- native: NativeMachine::new(options.asynchronous),
- start: Some(Box::new(start)),
- closed: false,
arguments: Some(arguments),
- started_at: 0.0,
- ended_at: None,
- stage: Stage::Begin,
- pending: None,
interrupted: None,
- observers: options.observers,
- observing: false,
- terminal_observation: None,
- };
- if options.asynchronous {
- return Execution::new(driver, options.lifecycle)
- .into_coroutine(py)
- .map(Bound::unbind);
- }
- match driver.resume(None)? {
- ExecutionStep::Return(value) => Ok(value),
- ExecutionStep::Open(head) => Execution::suspended(driver, options.lifecycle)
- .into_sync_stream(py, head)
- .map(Bound::unbind),
- ExecutionStep::Await(_) | ExecutionStep::Yield(_) => {
- Err(PyRuntimeError::new_err("sync call suspended"))
- }
- }
-}
-
-fn is_cancellation(py: Python<'_>, error: &PyErr) -> bool {
- !error.is_instance_of::(py)
-}
-
-impl PythonDriver
-where
- L: PythonCallHooks + 'static,
- H: PythonBinding + PythonHostCalls,
- M: Machine + 'static,
- M::Complete: Into>>,
-{
- fn timing(&self) -> Timing {
- Timing {
- start_time: self.started_at,
- end_time: self.ended_at.unwrap_or_else(epoch_seconds),
- }
- }
-
- fn drive(
- &mut self,
- py: Python<'_>,
- result: Option>>,
- ) -> PyResult {
- match (self.pending.take(), result) {
- (None, None) => {
- self.started_at = epoch_seconds();
- let started = PythonCallEvent::Started {
- start_time: self.started_at,
- };
- self.observing = true;
- self.observe(&started);
- match self.hooks.on_event(py, started) {
- Ok(step) => self.on_event(py, step, EventNext::Started),
- Err(error) => self.hook_failed(py, error),
- }
- }
- (Some(Pending::Host), Some(result)) => {
- match self.binding.resume_host_call(py, result) {
- Ok(Some(awaitable)) => {
- self.pending = Some(Pending::Host);
- Ok(ExecutionStep::Await(awaitable))
- }
- Ok(None) => self.resume_machine(py, None),
- Err(InvokeError::Python(error)) => self.interrupt(py, error),
- Err(InvokeError::Native(error)) => {
- self.resume_machine(py, Some(HostFailure::Error(error)))
- }
- }
- }
- (Some(Pending::Native), Some(Ok(_))) => {
- let result = self.native.take_result()?;
- self.run_steps(py, NativePoll::Ready(result))
- }
- (Some(Pending::Native), Some(Err(error))) => self.interrupt(py, error),
- (Some(Pending::Consumer(reply)), Some(read)) => {
- reply.send(if read.is_ok() {
- ControlFlow::Continue(())
- } else {
- ControlFlow::Break(())
- });
- self.resume_machine(py, None)
- }
- (Some(Pending::Arguments(resume)), Some(result)) => {
- let step = resume(&mut self.hooks, py, result);
- match step {
- Ok(step) => self.on_arguments(py, step),
- Err(error) => self.hook_failed(py, error),
- }
- }
- (Some(Pending::Wire(resume, reply)), Some(result)) => {
- let step = resume(&mut self.hooks, py, result);
- match step {
- Ok(step) => self.on_wire(py, step, reply),
- Err(error) => self.hook_failed(py, error),
- }
- }
- (Some(Pending::Response(resume)), Some(result)) => {
- let step = resume(&mut self.hooks, py, result);
- match step {
- Ok(step) => self.on_response(py, step),
- Err(error) => self.hook_failed(py, error),
- }
- }
- (Some(Pending::Event(resume, next)), Some(result)) => {
- let step = resume(&mut self.hooks, py, result);
- match step {
- Ok(step) => self.on_event(py, step, next),
- Err(error) => self.hook_failed(py, error),
- }
- }
- _ => Err(missing_state()),
- }
- }
-
- fn on_arguments(
- &mut self,
- py: Python<'_>,
- step: HookStep>,
- ) -> PyResult {
- match step {
- HookStep::Await(awaitable, resume) => {
- self.pending = Some(Pending::Arguments(resume));
- Ok(ExecutionStep::Await(awaitable))
- }
- HookStep::Ready(arguments) => {
- if let Err(error) = self.hooks.arguments_prepared(py, &arguments) {
- return self.hook_failed(py, error);
- }
- let decoded = self.binding.decode_request(py, arguments.bind(py));
- self.arguments = Some(arguments);
- let request = match decoded {
- Ok(request) => request,
- Err(InvokeError::Native(error)) => return self.machine_failed(py, error),
- Err(InvokeError::Python(error)) => {
- return self.failure(py, error, FailureOrigin::Call);
- }
- };
- let start = self.start.take().ok_or_else(missing_state)?;
- let arguments = self.arguments.as_ref().ok_or_else(missing_state)?;
- let machine = match start(py, arguments.bind(py), request) {
- Ok(machine) => machine,
- Err(error) => return self.failure(py, error, FailureOrigin::Host),
- };
- self.native.start(machine);
- self.stage = Stage::Call;
- self.resume_machine(py, None)
- }
- }
- }
-
- fn on_wire(
- &mut self,
- py: Python<'_>,
- step: HookStep>,
- reply: Reply,
- ) -> PyResult {
- match step {
- HookStep::Await(awaitable, resume) => {
- self.pending = Some(Pending::Wire(resume, reply));
- Ok(ExecutionStep::Await(awaitable))
- }
- HookStep::Ready(wire) => {
- reply.send(*wire);
- self.resume_machine(py, None)
- }
- }
- }
-
- fn on_response(
- &mut self,
- py: Python<'_>,
- step: HookStep>,
- ) -> PyResult {
- match step {
- HookStep::Await(awaitable, resume) => {
- self.pending = Some(Pending::Response(resume));
- Ok(ExecutionStep::Await(awaitable))
- }
- HookStep::Ready(response) => self.succeeded(py, response),
- }
- }
-
- fn on_event(
- &mut self,
- py: Python<'_>,
- step: HookStep,
- next: EventNext,
- ) -> PyResult {
- match step {
- HookStep::Await(awaitable, resume) => {
- self.pending = Some(Pending::Event(resume, next));
- Ok(ExecutionStep::Await(awaitable))
- }
- HookStep::Ready(()) => match next {
- EventNext::Started => self.begin(py),
- EventNext::Emitted(reply) => {
- reply.send(());
- self.resume_machine(py, None)
- }
- EventNext::Terminal => {
- if let Some(event) = self.terminal_observation.take()
- && let Some(observers) = &self.observers
- {
- observers.emit(event);
- }
- self.observing = false;
- match &self.stage {
- Stage::Succeeded(response) => {
- Ok(ExecutionStep::Return(response.clone_ref(py)))
- }
- Stage::Failed(error) => {
- Err(PyErr::from_value(error.bind(py).clone().into_any()))
- }
- _ => Err(missing_state()),
- }
- }
- },
- }
- }
-
- fn begin(&mut self, py: Python<'_>) -> PyResult {
- let arguments = self.arguments.take().ok_or_else(missing_state)?;
- match self.hooks.prepare_arguments(py, arguments, self.started_at) {
- Ok(step) => self.on_arguments(py, step),
- Err(error) => self.hook_failed(py, error),
- }
- }
-
- fn hook_failed(&mut self, py: Python<'_>, error: PyErr) -> PyResult {
- match self.stage {
- Stage::Begin | Stage::AfterSuccess => self.failure(py, error, FailureOrigin::Host),
- Stage::Call | Stage::Streaming => self.interrupt(py, error),
- Stage::Succeeded(_) | Stage::Failed(_) => Err(error),
- }
- }
-
- fn resume_machine(
- &mut self,
- py: Python<'_>,
- interruption: Interruption,
- ) -> PyResult {
- let step = self.native.resume(py, interruption)?;
- self.run_steps(py, step)
- }
-
- fn run_steps(
- &mut self,
- py: Python<'_>,
- mut step: NativePoll>,
- ) -> PyResult {
- loop {
- let result = match step {
- NativePoll::Suspend(awaitable) => {
- self.pending = Some(Pending::Native);
- return Ok(ExecutionStep::Await(awaitable));
- }
- NativePoll::Ready(result) => result,
- };
- step = match self.handle_native(py, result)? {
- Next::Return(step) => return Ok(step),
- Next::Continue(step) => step,
- };
- }
- }
-
- /// Answers one machine step: performs the op it asked for, or finishes the call.
- fn handle_native(&mut self, py: Python<'_>, result: NativeResult) -> PyResult> {
- let op = match result {
- Ok(MachineStep::Suspended(op)) => op,
- Ok(MachineStep::Complete(response)) => {
- return self.completed(py, response).map(Next::Return);
- }
- Err(error) => return self.machine_failed(py, error).map(Next::Return),
- };
- let answered = match op {
- HostRequest::HostCall(op) => match self.binding.begin_host_call(py, op) {
- Ok(Some(awaitable)) => {
- self.pending = Some(Pending::Host);
- return Ok(Next::Return(ExecutionStep::Await(awaitable)));
- }
- result => answered(result.map(|_| ())),
- },
- HostRequest::Intercept(InterceptRequest::BeforeProviderRequest {
- wire,
- context,
- reply,
- }) => match self.hooks.before_provider_request(py, wire, &context) {
- Ok(HookStep::Ready(wire)) => {
- reply.send(*wire);
- Ok(Ok(()))
- }
- Ok(HookStep::Await(awaitable, resume)) => {
- self.pending = Some(Pending::Wire(resume, reply));
- return Ok(Next::Return(ExecutionStep::Await(awaitable)));
- }
- Err(error) => Err(error),
- },
- HostRequest::Stream(StreamDelivery::Open(head, reply)) => {
- return self.opened(py, head, reply).map(Next::Return);
- }
- HostRequest::Stream(StreamDelivery::Chunk(chunk, reply)) => {
- return self.delivered(py, chunk, reply).map(Next::Return);
- }
- HostRequest::Intercept(InterceptRequest::ResultReady { facts, reply }) => {
- let event = PythonCallEvent::Execution(ExecutionEvent::ResultReady { facts });
- self.observe(&event);
- match self.hooks.on_event(py, event) {
- Ok(HookStep::Ready(())) => {
- reply.send(());
- Ok(Ok(()))
- }
- Ok(HookStep::Await(awaitable, resume)) => {
- self.pending = Some(Pending::Event(resume, EventNext::Emitted(reply)));
- return Ok(Next::Return(ExecutionStep::Await(awaitable)));
- }
- Err(error) => Err(error),
- }
- }
- HostRequest::Intercept(InterceptRequest::AfterProviderResponse { raw, reply }) => {
- let event = PythonCallEvent::Execution(ExecutionEvent::ProviderResponseReceived {
- raw: &raw,
- });
- self.observe(&event);
- match self.hooks.on_event(py, event) {
- Ok(HookStep::Ready(())) => {
- reply.send(());
- Ok(Ok(()))
- }
- Ok(HookStep::Await(awaitable, resume)) => {
- self.pending = Some(Pending::Event(resume, EventNext::Emitted(reply)));
- return Ok(Next::Return(ExecutionStep::Await(awaitable)));
- }
- Err(error) => Err(error),
- }
- }
- };
- match answered {
- Ok(Ok(())) => self.native.resume(py, None).map(Next::Continue),
- Ok(Err(native)) => self
- .native
- .resume(py, Some(HostFailure::Error(native)))
- .map(Next::Continue),
- Err(error) => self.interrupt(py, error).map(Next::Return),
- }
- }
-
- fn opened(
- &mut self,
- py: Python<'_>,
- head: as Protocol>::StreamHead,
- reply: Reply>,
- ) -> PyResult {
- self.stage = Stage::Streaming;
- let head = match self.binding.encode_stream_head(py, head) {
- Ok(head) => head,
- Err(error) => return self.interrupt(py, error),
- };
- match self.hooks.on_stream_open(py, &head) {
- Ok(()) => {
- self.pending = Some(Pending::Consumer(reply));
- Ok(ExecutionStep::Open(head))
- }
- Err(error) => self.interrupt(py, error),
- }
- }
-
- fn delivered(
- &mut self,
- py: Python<'_>,
- chunk: as Protocol>::Chunk,
- reply: Reply>,
- ) -> PyResult {
- let chunk = match self.binding.encode_chunk(py, chunk) {
- Ok(chunk) => chunk,
- Err(error) => return self.interrupt(py, error),
- };
- match self.hooks.on_stream_chunk(py, &chunk) {
- Ok(()) => {
- self.pending = Some(Pending::Consumer(reply));
- Ok(ExecutionStep::Yield(chunk))
- }
- Err(error) => self.interrupt(py, error),
- }
- }
-
- fn interrupt(&mut self, py: Python<'_>, error: PyErr) -> PyResult {
- let cancelled = is_cancellation(py, &error);
- let native = H::host_error(&error);
- self.interrupted = Some(error.into_value(py));
- let failure = if cancelled {
- HostFailure::Cancelled(native)
- } else {
- HostFailure::Error(native)
- };
- self.resume_machine(py, Some(failure))
- }
-
- fn completed(
- &mut self,
- py: Python<'_>,
- response: HostedCompletion>,
- ) -> PyResult {
- self.ended_at = Some(epoch_seconds());
- let response = match response {
- HostedCompletion::Complete(response) => response,
- HostedCompletion::StreamEnded | HostedCompletion::Detached => {
- return self.succeeded(py, py.None());
- }
- };
- let public = match self.binding.encode_response(py, response) {
- Ok(public) => public,
- Err(error) => return self.failure(py, error, FailureOrigin::Call),
- };
- if let Stage::Streaming = self.stage {
- return self.succeeded(py, public);
- }
- self.stage = Stage::AfterSuccess;
- match self.hooks.transform_response(py, public, self.timing()) {
- Ok(step) => self.on_response(py, step),
- Err(error) => self.failure(py, error, FailureOrigin::Host),
- }
- }
-
- fn machine_failed(&mut self, py: Python<'_>, error: ErrorOf) -> PyResult {
- self.ended_at.get_or_insert_with(epoch_seconds);
- let error = match self.interrupted.take() {
- Some(retained) => PyErr::from_value(retained.into_bound(py).into_any()),
- None => self.classified(py, error),
- };
- self.failure(py, error, FailureOrigin::Call)
- }
-
- /// The route's public exception for a native failure. When classification itself
- /// fails, that failure is raised with the native error's text as its `__context__`.
- fn classified(&self, py: Python<'_>, error: ErrorOf) -> PyErr {
- let native = error.to_string();
- let classifier_error = match self.binding.map_error(py, error) {
- Ok(failure) => return failure.into(),
- Err(classifier_error) => classifier_error,
- };
- classifier_error.set_context(py, Some(PyRuntimeError::new_err(native)));
- classifier_error
- }
-
- fn succeeded(&mut self, py: Python<'_>, response: Py) -> PyResult {
- let event = PythonCallEvent::Succeeded {
- timing: self.timing(),
- response: &response,
- };
- self.terminal_observation = self.observers.as_ref().map(|_| event.snapshot());
- let step = self.hooks.on_event(py, event)?;
- self.stage = Stage::Succeeded(response);
- self.on_event(py, step, EventNext::Terminal)
- }
-
- fn failure(
- &mut self,
- py: Python<'_>,
- error: PyErr,
- origin: FailureOrigin,
- ) -> PyResult {
- self.ended_at.get_or_insert_with(epoch_seconds);
- if is_cancellation(py, &error) {
- return Err(error);
- }
- let event = PythonCallEvent::Failed {
- timing: self.timing(),
- origin,
- error: &error,
- };
- self.terminal_observation = self.observers.as_ref().map(|_| event.snapshot());
- let step = self.hooks.on_event(py, event)?;
- self.stage = Stage::Failed(error.into_value(py));
- self.on_event(py, step, EventNext::Terminal)
- }
-
- fn observe(&self, event: &PythonCallEvent<'_>) {
- if let Some(observers) = &self.observers {
- observers.emit(event.snapshot());
- }
- }
-
- fn clear(&mut self) {
- if !self.closed {
- self.closed = true;
- if self.observing {
- self.observe(&PythonCallEvent::Cancelled {
- timing: self.timing(),
- });
- self.observing = false;
- }
- self.native.close();
- self.start = None;
- Python::attach(|py| {
- self.hooks.close(py);
- self.binding.close(py);
- });
- }
- }
-}
-
-impl ExecutionBody for PythonDriver
-where
- L: PythonCallHooks + 'static,
- H: PythonBinding + PythonHostCalls,
- M: Machine + 'static,
- M::Complete: Into>>,
-{
- fn resume(&mut self, result: Option>>) -> PyResult {
- Python::attach(|py| {
- let outcome = self.drive(py, result);
- if let Err(error) = &outcome
- && self.observing
- {
- let event = if is_cancellation(py, error) {
- PythonCallEvent::Cancelled {
- timing: self.timing(),
- }
- } else {
- PythonCallEvent::Failed {
- timing: self.timing(),
- origin: FailureOrigin::Host,
- error,
- }
- };
- self.observe(&event);
- self.observing = false;
- self.terminal_observation = None;
- }
- outcome
- })
- }
-
- fn traverse(&self, visit: &PyVisit<'_>) -> Result<(), PyTraverseError> {
- self.binding.traverse(visit)?;
- self.hooks.traverse(visit)?;
- visit.call(&self.arguments)?;
- visit.call(&self.interrupted)?;
- match &self.stage {
- Stage::Succeeded(response) => visit.call(response),
- Stage::Failed(error) => visit.call(error),
- _ => Ok(()),
- }
- }
-}
-
-impl Drop for PythonDriver
-where
- L: PythonCallHooks + 'static,
- H: PythonBinding + PythonHostCalls,
- M: Machine + 'static,
- M::Complete: Into>>,
-{
- fn drop(&mut self) {
- self.clear();
- }
-}
-
-#[cfg(test)]
-mod tests {
- use std::sync::{Arc, Mutex};
-
- use litellm_host::{
- interceptors::{RawResponse, RequestContext},
- machine::{CallMachine, MachineFault},
- };
- use pyo3::exceptions::{PyBaseException, PyValueError};
- use pyo3::types::PyDict;
-
- use super::*;
- use crate::{PythonOwned, PythonRuntime};
- use litellm_host::hooks::CallHooks;
- use litellm_host::interceptors::Interceptors;
-
- static PYTHON_GLOBALS: Mutex<()> = Mutex::new(());
-
- fn lifecycle_binding(py: Python<'_>) -> PyResult> {
- py.import("host_test_lifecycle")
- }
-
- fn call_options(asynchronous: bool) -> CallOptions {
- CallOptions::new(asynchronous, lifecycle_binding)
- }
-
- fn install_lifecycle_module(py: Python<'_>) -> Bound<'_, PyModule> {
- let source =
- std::ffi::CString::new(include_str!("../../../../litellm/rust_bridge/lifecycle.py"))
- .unwrap();
- PyModule::from_code(
- py,
- &source,
- pyo3::ffi::c_str!("lifecycle.py"),
- pyo3::ffi::c_str!("host_test_lifecycle"),
- )
- .unwrap()
- }
-
- #[derive(Clone, Debug, PartialEq, Eq)]
- struct Error(String);
-
- impl std::fmt::Display for Error {
- fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
- formatter.write_str(&self.0)
- }
- }
-
- impl From for Error {
- fn from(fault: MachineFault) -> Self {
- Self(format!("{fault:?}"))
- }
- }
-
- struct Synthetic;
-
- impl Protocol for Synthetic {
- type Response = String;
- type Error = Error;
- type Request = String;
- type HostCall = (&'static str, Reply);
- type Chunk = std::convert::Infallible;
- type StreamHead = std::convert::Infallible;
- }
-
- fn wire() -> WireRequest {
- WireRequest {
- url: "https://example.invalid".into(),
- headers: Vec::new(),
- body: serde_json::json!({}),
- }
- }
-
- fn context() -> RequestContext {
- RequestContext {
- model: "model".into(),
- custom_llm_provider: "provider".into(),
- optional_params: serde_json::json!({}),
- secret_fields: Vec::new(),
- api_key: None,
- }
- }
-
- #[derive(Default)]
- struct Log(Arc>>);
-
- impl Log {
- fn push(&self, entry: impl Into) {
- self.0.lock().unwrap().push(entry.into());
- }
-
- fn entries(&self) -> Vec {
- self.0.lock().unwrap().clone()
- }
- }
-
- #[derive(Clone, Copy)]
- enum OpScript {
- Answer,
- RaisePython,
- RejectNatively,
- RejectRequestNatively,
- RaiseRequestPython,
- AwaitAnswer,
- AwaitFailure,
- }
-
- struct SyntheticBinding {
- log: Log,
- op: OpScript,
- classifier_fails: bool,
- pending_reply: Option>,
- }
-
- /// The fake route's public exception, kept as a value so a test sees what `classify`
- /// produced before the driver raises it.
- #[derive(Debug, PartialEq, Eq)]
- struct Classified(String);
-
- impl From for PyErr {
- fn from(classified: Classified) -> Self {
- PyValueError::new_err(format!("classified: {}", classified.0))
- }
- }
-
- impl SyntheticBinding {
- fn answer(&self, value: impl FnOnce() -> String) -> Result> {
- match self.op {
- OpScript::Answer | OpScript::AwaitAnswer | OpScript::AwaitFailure => Ok(value()),
- OpScript::RaisePython | OpScript::RaiseRequestPython => {
- Err(PyValueError::new_err("op failed").into())
- }
- OpScript::RejectNatively | OpScript::RejectRequestNatively => {
- Err(InvokeError::Native(Error("op rejected".into())))
- }
- }
- }
- }
-
- impl PythonBinding for SyntheticBinding {
- type Protocol = Synthetic;
- type Failure = Classified;
-
- fn decode_request(
- &mut self,
- _: Python<'_>,
- arguments: &Bound<'_, PyDict>,
- ) -> Result> {
- self.log.push("project");
- match self.op {
- OpScript::RejectRequestNatively | OpScript::RaiseRequestPython => {
- self.answer(String::new)
- }
- _ => Ok(format!("project:{}", arguments.len())),
- }
- }
-
- fn encode_stream_head(
- &mut self,
- _: Python<'_>,
- head: std::convert::Infallible,
- ) -> PyResult> {
- match head {}
- }
-
- fn encode_chunk(
- &mut self,
- _: Python<'_>,
- chunk: std::convert::Infallible,
- ) -> PyResult> {
- match chunk {}
- }
-
- fn encode_response(&mut self, py: Python<'_>, response: String) -> PyResult> {
- self.log.push("complete");
- Ok(pyo3::types::PyString::new(py, &response)
- .into_any()
- .unbind())
- }
-
- fn map_error(&self, _: Python<'_>, error: Error) -> PyResult {
- self.log.push(format!("classify:{error}"));
- if self.classifier_fails {
- return Err(pyo3::exceptions::PyTypeError::new_err("classifier failed"));
- }
- Ok(Classified(error.0))
- }
-
- fn host_error(error: &PyErr) -> Error {
- Error(error.to_string())
- }
- }
-
- impl PythonHostCalls for SyntheticBinding {
- fn handle_host_call(
- &mut self,
- _: Python<'_>,
- (op, reply): (&'static str, Reply),
- ) -> Result<(), InvokeError> {
- self.log.push(format!("op:{op}"));
- self.answer(|| op.to_string())
- .map(|answer| reply.send(answer))
- }
-
- fn begin_host_call(
- &mut self,
- py: Python<'_>,
- (op, reply): (&'static str, Reply),
- ) -> Result