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        reasoning: None,
62        strategy: Some(Strategy::Default),
63        input,
64        split: if request.split { Some(true) } else { None },
65        invert: if request.invert { Some(true) } else { None },
66        provider: None,
67        seed,
68        stream: Some(true),
69        continuation: request.continuation,
70    };
71    let inner = super::runner::run(ctx.clone(), params);
72    Ok(Box::pin(inner.map(|r| {
73        r.map(|ev| match ev {
74            super::runner::Event::Id(id) => ResponseItem::Id(id),
75            super::runner::Event::Chunk(c) => ResponseItem::Chunk(c),
76        })
77    })))
78}
79
80/// Stream-false: self-respawn as a detached subprocess running
81/// the same `functions execute standard ...` argv with
82/// `stream=true`. Take the first `ResponseItem::Id` off the
83/// child's stdout, yield it, return. Child outlives this call.
84async fn execute_detached(request: Request) -> Result<ItemStream, Error> {
85    let mut child_request = request;
86    match child_request.dangerous_advanced.as_mut() {
87        Some(adv) => adv.stream = Some(true),
88        None => {
89            child_request.dangerous_advanced = Some(RequestDangerousAdvanced {
90                stream: Some(true),
91                ..Default::default()
92            })
93        }
94    }
95    // Re-exec of this CLI — strip the parent-only envelope fields.
96    crate::command::reexec::strip_inherited(&mut child_request.base);
97    let exe = std::env::current_exe()
98        .map_err(|e| Error::Spawn("current_exe".into(), e))?;
99    let executor = BinaryExecutor::from_path(exe).detach(true);
100    let mut stream = executor
101        .execute::<Request, ResponseItem>(child_request, None)
102        .await
103        .map_err(|e| Error::Instance(format!(
104            "self-respawn for functions execute standard: {e}"
105        )))?;
106    let first = stream
107        .next()
108        .await
109        .ok_or(Error::EmptyStream)?
110        .map_err(|e| Error::Instance(format!(
111            "self-respawn for functions execute standard: {e}"
112        )))?;
113    Ok(Box::pin(
114        objectiveai_sdk::cli::command::StreamOnce::new(Ok(first)),
115    ))
116}
117
118pub mod request_schema {
119    use objectiveai_sdk::cli::command::functions::execute::standard as sdk;
120    use objectiveai_sdk::cli::command::functions::execute::standard::request_schema::{Request, Response};
121
122    use crate::context::Context;
123    use crate::error::Error;
124
125    pub async fn execute(_ctx: &Context, _request: Request) -> Result<Response, Error> {
126        Ok(objectiveai_sdk::cli::command::ResponseSchema(schemars::schema_for!(sdk::Request)))
127    }
128}
129
130pub mod response_schema {
131    use objectiveai_sdk::cli::command::functions::execute::standard as sdk;
132    use objectiveai_sdk::cli::command::functions::execute::standard::response_schema::{Request, Response};
133
134    use crate::context::Context;
135    use crate::error::Error;
136
137    pub async fn execute(_ctx: &Context, _request: Request) -> Result<Response, Error> {
138        Ok(objectiveai_sdk::cli::command::ResponseSchema(schemars::schema_for!(sdk::Response)))
139    }
140}
objectiveai_cli/command/functions/execute/standard.rs

objectiveai_cli/command/functions/execute/
standard.rs
