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

objectiveai_cli/command/functions/execute/
standard.rs
