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

objectiveai_cli/command/functions/execute/
swiss_system.rs
