objectiveai_cli/command/functions/execute/
swiss_system.rs

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

objectiveai_cli/command/functions/execute/
swiss_system.rs
