objectiveai_cli/command/functions/execute/
standard.rs

1//! `functions execute standard` — in-process chunk-or-id
2//! streaming handler.
3//!
4//! Same two-mode shape as `agents spawn`: `stream=true` drives the
5//! upstream WS directly via [`super::runner::run`] and yields each
6//! chunk as it arrives; `stream=false` re-execs `objectiveai-cli`
7//! as a detached subprocess with the same argv plus
8//! `stream=true`, captures the first `ResponseItem::Id`, yields
9//! it, and returns. The detached subprocess outlives the parent
10//! (Unix: kernel re-parents to init; Windows: `DETACHED_PROCESS |
11//! CREATE_NEW_PROCESS_GROUP` from `BinaryExecutor::detach(true)`).
12
13use std::pin::Pin;
14
15use futures::Stream;
16use futures::StreamExt;
17use objectiveai_sdk::cli::command::functions::execute::standard::{
18    Request, RequestDangerousAdvanced, RequestInput, ResponseItem,
19};
20use objectiveai_sdk::cli::command::{BinaryExecutor, CommandExecutor};
21use objectiveai_sdk::functions::executions::request::{
22    FunctionExecutionCreateParams, Strategy,
23};
24
25use crate::context::Context;
26use crate::error::Error;
27
28type ItemStream = Pin<Box<dyn Stream<Item = Result<ResponseItem, Error>> + Send>>;
29
30pub async fn execute(ctx: &Context, request: Request) -> Result<ItemStream, Error> {
31    let want_stream = request
32        .dangerous_advanced
33        .as_ref()
34        .and_then(|a| a.stream)
35        .unwrap_or(false);
36    if want_stream {
37        execute_streaming(ctx, request).await
38    } else {
39        execute_detached(request).await
40    }
41}
42
43async fn execute_streaming(
44    ctx: &Context,
45    request: Request,
46) -> Result<ItemStream, Error> {
47    let (function, profile) = tokio::try_join!(
48        super::resolve_function(ctx, request.function),
49        super::resolve_profile(ctx, request.profile),
50    )?;
51    let input = match request.input {
52        RequestInput::Inline(v) => v,
53        RequestInput::File(path) => super::resolve_input_file(path)?,
54        RequestInput::PythonInline(code) => super::resolve_input_python_inline(ctx, code).await?,
55        RequestInput::PythonFile(path) => super::resolve_input_python_file(ctx, path).await?,
56    };
57    let seed = request.dangerous_advanced.as_ref().and_then(|a| a.seed);
58    let params = FunctionExecutionCreateParams {
59        function,
60        profile,
61        retry_token: request.retry_token,
62        from_cache: None,
63        reasoning: None,
64        strategy: Some(Strategy::Default),
65        input,
66        split: if request.split { Some(true) } else { None },
67        invert: if request.invert { Some(true) } else { None },
68        provider: None,
69        seed,
70        stream: Some(true),
71        continuation: request.continuation,
72    };
73    let inner = super::runner::run(ctx.clone(), params);
74    Ok(Box::pin(inner.map(|r| {
75        r.map(|ev| match ev {
76            super::runner::Event::Id(id) => ResponseItem::Id(id),
77            super::runner::Event::Chunk(c) => ResponseItem::Chunk(c),
78        })
79    })))
80}
81
82/// Stream-false: self-respawn as a detached subprocess running
83/// the same `functions execute standard ...` argv with
84/// `stream=true`. Take the first `ResponseItem::Id` off the
85/// child's stdout, yield it, return. Child outlives this call.
86async fn execute_detached(request: Request) -> Result<ItemStream, Error> {
87    let mut child_request = request;
88    match child_request.dangerous_advanced.as_mut() {
89        Some(adv) => adv.stream = Some(true),
90        None => {
91            child_request.dangerous_advanced = Some(RequestDangerousAdvanced {
92                stream: Some(true),
93                ..Default::default()
94            })
95        }
96    }
97    // Re-exec of this CLI — strip the parent-only envelope fields.
98    crate::command::reexec::strip_inherited(&mut child_request.base);
99    let exe = std::env::current_exe()
100        .map_err(|e| Error::Spawn("current_exe".into(), e))?;
101    let executor = BinaryExecutor::from_path(exe).detach(true);
102    let mut stream = executor
103        .execute::<Request, ResponseItem>(child_request, None)
104        .await
105        .map_err(|e| Error::Instance(format!(
106            "self-respawn for functions execute standard: {e}"
107        )))?;
108    let first = stream
109        .next()
110        .await
111        .ok_or(Error::EmptyStream)?
112        .map_err(|e| Error::Instance(format!(
113            "self-respawn for functions execute standard: {e}"
114        )))?;
115    Ok(Box::pin(
116        objectiveai_sdk::cli::command::StreamOnce::new(Ok(first)),
117    ))
118}
119
120pub mod request_schema {
121    use objectiveai_sdk::cli::command::functions::execute::standard as sdk;
122    use objectiveai_sdk::cli::command::functions::execute::standard::request_schema::{Request, Response};
123
124    use crate::context::Context;
125    use crate::error::Error;
126
127    pub async fn execute(_ctx: &Context, _request: Request) -> Result<Response, Error> {
128        Ok(objectiveai_sdk::cli::command::ResponseSchema(schemars::schema_for!(sdk::Request)))
129    }
130}
131
132pub mod response_schema {
133    use objectiveai_sdk::cli::command::functions::execute::standard as sdk;
134    use objectiveai_sdk::cli::command::functions::execute::standard::response_schema::{Request, Response};
135
136    use crate::context::Context;
137    use crate::error::Error;
138
139    pub async fn execute(_ctx: &Context, _request: Request) -> Result<Response, Error> {
140        Ok(objectiveai_sdk::cli::command::ResponseSchema(schemars::schema_for!(sdk::Response)))
141    }
142}
objectiveai_cli/command/functions/execute/standard.rs

objectiveai_cli/command/functions/execute/
standard.rs
