objectiveai_cli/command/functions/execute/
mod.rs

1//! `functions execute` sub-tier. Standard and SwissSystem are
2//! chunk-or-id streaming leaves; their bare-naked `execute` returns
3//! `Stream<ResponseItem>` and the inner stream is mapped into the
4//! tier's `ResponseItem` directly.
5
6use std::path::PathBuf;
7use std::pin::Pin;
8
9use futures::{Stream, StreamExt};
10use objectiveai_sdk::cli::command::functions::execute::{
11    FunctionSpec, ProfileSpec, Request, ResponseItem,
12};
13use objectiveai_sdk::functions::expression::InputValue;
14use objectiveai_sdk::functions::{
15    FullInlineFunctionOrRemoteCommitOptional, InlineProfileOrRemoteCommitOptional,
16};
17
18use crate::context::Context;
19use crate::error::Error;
20
21pub(super) mod runner;
22pub mod standard;
23pub mod swiss_system;
24
25pub(super) async fn resolve_function(
26    ctx: &Context,
27    spec: FunctionSpec,
28) -> Result<FullInlineFunctionOrRemoteCommitOptional, Error> {
29    match spec {
30        FunctionSpec::Resolved(r) => Ok(r),
31        FunctionSpec::File(path) => read_json_file(&path),
32        FunctionSpec::PythonInline(code) => ctx.python().await?.exec_code(&code, None::<()>).await?.ok_or(Error::PythonNoOutput),
33        FunctionSpec::PythonFile(path) => ctx.python().await?.exec_file(&path, None::<()>).await?.ok_or(Error::PythonNoOutput),
34    }
35}
36
37pub(super) async fn resolve_profile(
38    ctx: &Context,
39    spec: ProfileSpec,
40) -> Result<InlineProfileOrRemoteCommitOptional, Error> {
41    match spec {
42        ProfileSpec::Resolved(r) => Ok(r),
43        ProfileSpec::File(path) => read_json_file(&path),
44        ProfileSpec::PythonInline(code) => ctx.python().await?.exec_code(&code, None::<()>).await?.ok_or(Error::PythonNoOutput),
45        ProfileSpec::PythonFile(path) => ctx.python().await?.exec_file(&path, None::<()>).await?.ok_or(Error::PythonNoOutput),
46    }
47}
48
49pub(super) fn resolve_input_file(path: PathBuf) -> Result<InputValue, Error> {
50    read_json_file(&path)
51}
52
53pub(super) async fn resolve_input_python_inline(
54    ctx: &Context,
55    code: String,
56) -> Result<InputValue, Error> {
57    ctx.python().await?.exec_code(&code, None::<()>).await?.ok_or(Error::PythonNoOutput)
58}
59
60pub(super) async fn resolve_input_python_file(
61    ctx: &Context,
62    path: PathBuf,
63) -> Result<InputValue, Error> {
64    ctx.python().await?.exec_file(&path, None::<()>).await?.ok_or(Error::PythonNoOutput)
65}
66
67/// Read a JSON file and deserialize as `T`. Used by every `*-file`
68/// resolution arm above. `serde_path_to_error` preserves the
69/// in-document location of the failure so error messages point at
70/// the offending field.
71fn read_json_file<T: serde::de::DeserializeOwned>(path: &std::path::Path) -> Result<T, Error> {
72    let bytes = std::fs::read(path)
73        .map_err(|e| Error::JsonFileRead(path.to_path_buf(), e))?;
74    let mut de = serde_json::Deserializer::from_slice(&bytes);
75    serde_path_to_error::deserialize(&mut de).map_err(Error::InlineDeserialize)
76}
77
78type ItemStream = Pin<Box<dyn Stream<Item = Result<ResponseItem, Error>> + Send>>;
79
80fn once<T: Send + 'static>(
81    item: Result<T, Error>,
82) -> Pin<Box<dyn Stream<Item = Result<T, Error>> + Send>> {
83    Box::pin(futures::stream::once(async move { item }))
84}
85
86pub async fn execute(ctx: &Context, request: Request) -> Result<ItemStream, Error> {
87    let stream: ItemStream = match request {
88        Request::Standard(req) => {
89            let inner = standard::execute(ctx, req).await?;
90            Box::pin(inner.map(|r| r.map(ResponseItem::Standard)))
91        }
92        Request::StandardRequestSchema(req) => {
93            let value = standard::request_schema::execute(ctx, req).await?;
94            once(Ok(ResponseItem::StandardRequestSchema(value)))
95        }
96        Request::StandardResponseSchema(req) => {
97            let value = standard::response_schema::execute(ctx, req).await?;
98            once(Ok(ResponseItem::StandardResponseSchema(value)))
99        }
100        Request::SwissSystem(req) => {
101            let inner = swiss_system::execute(ctx, req).await?;
102            Box::pin(inner.map(|r| r.map(ResponseItem::SwissSystem)))
103        }
104        Request::SwissSystemRequestSchema(req) => {
105            let value = swiss_system::request_schema::execute(ctx, req).await?;
106            once(Ok(ResponseItem::SwissSystemRequestSchema(value)))
107        }
108        Request::SwissSystemResponseSchema(req) => {
109            let value = swiss_system::response_schema::execute(ctx, req).await?;
110            once(Ok(ResponseItem::SwissSystemResponseSchema(value)))
111        }
112    };
113    Ok(stream)
114}
objectiveai_cli/command/functions/execute/mod.rs

objectiveai_cli/command/functions/execute/
mod.rs
