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    AgentInstanceHierarchy, AgentInstanceHierarchyType, Request,
11    RequestDangerousAdvanced, RequestInput, ResponseItem,
12};
13use objectiveai_sdk::cli::command::{BinaryExecutor, CommandExecutor};
14use objectiveai_sdk::functions::executions::request::{
15    FunctionExecutionCreateParams, Strategy,
16};
17
18use crate::context::Context;
19use crate::error::Error;
20
21type ItemStream = Pin<Box<dyn Stream<Item = Result<ResponseItem, Error>> + Send>>;
22
23pub async fn execute(ctx: &Context, request: Request) -> Result<ItemStream, Error> {
24    let want_stream = request
25        .dangerous_advanced
26        .as_ref()
27        .and_then(|a| a.stream)
28        .unwrap_or(false);
29    if want_stream {
30        execute_streaming(ctx, request).await
31    } else {
32        execute_detached(request).await
33    }
34}
35
36async fn execute_streaming(
37    ctx: &Context,
38    request: Request,
39) -> Result<ItemStream, Error> {
40    let (function, profile) = tokio::try_join!(
41        super::resolve_function(ctx, request.function),
42        super::resolve_profile(ctx, request.profile),
43    )?;
44    let input = match request.input {
45        RequestInput::Inline(v) => v,
46        RequestInput::File(path) => super::resolve_input_file(path)?,
47        RequestInput::PythonInline(code) => super::resolve_input_python_inline(ctx, code).await?,
48        RequestInput::PythonFile(path) => super::resolve_input_python_file(ctx, path).await?,
49    };
50    let seed = request.dangerous_advanced.as_ref().and_then(|a| a.seed);
51    let params = FunctionExecutionCreateParams {
52        function,
53        profile,
54        reasoning: None,
55        strategy: Some(Strategy::SwissSystem {
56            pool: request.pool,
57            rounds: request.rounds,
58        }),
59        input,
60        split: if request.split { Some(true) } else { None },
61        invert: if request.invert { Some(true) } else { None },
62        provider: None,
63        seed,
64        stream: Some(true),
65        continuation: request.continuation,
66    };
67    let inner = super::runner::run(ctx.clone(), params);
68    Ok(Box::pin(inner.map(|r| {
69        r.map(|ev| match ev {
70            super::runner::Event::Id(id) => ResponseItem::Id(id),
71            super::runner::Event::Hierarchy(hier) => {
72                ResponseItem::AgentInstanceHierarchy(AgentInstanceHierarchy {
73                    r#type: AgentInstanceHierarchyType::AgentInstanceHierarchy,
74                    agent_instance_hierarchy: hier,
75                })
76            }
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 swiss-system ...` 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 swiss-system: {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 swiss-system: {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::swiss_system as sdk;
122    use objectiveai_sdk::cli::command::functions::execute::swiss_system::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::swiss_system as sdk;
134    use objectiveai_sdk::cli::command::functions::execute::swiss_system::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/swiss_system.rs

objectiveai_cli/command/functions/execute/
swiss_system.rs
