objectiveai_cli/command/agents/instances/
list.rs

1//! `agents instances list` — enumerate the DIRECT children of one or
2//! more resolved targets, with per-agent aggregates (tags, queued
3//! count, spawn/active timestamps, total logged messages). Backed by
4//! `db::instances::list_under_parent`.
5
6use std::collections::HashSet;
7use std::pin::Pin;
8
9use futures::stream::FuturesUnordered;
10use futures::{Stream, StreamExt};
11use objectiveai_sdk::cli::command::agents::instances::list::{Request, ResponseItem};
12
13use crate::context::Context;
14use crate::error::Error;
15
16type ItemStream = Pin<Box<dyn Stream<Item = Result<ResponseItem, Error>> + Send>>;
17
18pub async fn execute(ctx: &Context, request: Request) -> Result<ItemStream, Error> {
19    let default_parent = ctx.config.agent_instance_hierarchy.clone();
20    let db = ctx.db_client().await?.clone();
21    let stream = async_stream::stream! {
22        // Resolve + query every target concurrently. GROUPED/ABSENT tags
23        // resolve to None and contribute nothing.
24        let mut inflight = FuturesUnordered::new();
25        for target in request.targets {
26            let db = db.clone();
27            let default_parent = default_parent.clone();
28            inflight.push(async move {
29                let parent = super::resolve_target(&db, target, &default_parent).await?;
30                let items = match parent {
31                    Some(p) => crate::db::instances::list_under_parent(&db, &p)
32                        .await
33                        .map_err(Error::from)?,
34                    None => Vec::new(),
35                };
36                Ok::<Vec<ResponseItem>, Error>(items)
37            });
38        }
39        // Yield each child the moment its target's query lands — never
40        // waiting for the slowest. A `seen` set dedups by AIH across
41        // overlapping/nested targets (the BTreeMap's old job, minus the
42        // global sort that collecting would have forced).
43        let mut seen = HashSet::new();
44        while let Some(result) = inflight.next().await {
45            match result {
46                Ok(items) => {
47                    for item in items {
48                        if seen.insert(item.agent_instance_hierarchy.clone()) {
49                            yield Ok(item);
50                        }
51                    }
52                }
53                Err(e) => yield Err(e),
54            }
55        }
56    };
57    Ok(Box::pin(stream))
58}
59
60pub mod request_schema {
61    use objectiveai_sdk::cli::command::agents::instances::list as sdk;
62    use objectiveai_sdk::cli::command::agents::instances::list::request_schema::{Request, Response};
63
64    use crate::context::Context;
65    use crate::error::Error;
66
67    pub async fn execute(_ctx: &Context, _request: Request) -> Result<Response, Error> {
68        Ok(objectiveai_sdk::cli::command::ResponseSchema(schemars::schema_for!(sdk::Request)))
69    }
70}
71
72pub mod response_schema {
73    use objectiveai_sdk::cli::command::agents::instances::list as sdk;
74    use objectiveai_sdk::cli::command::agents::instances::list::response_schema::{Request, Response};
75
76    use crate::context::Context;
77    use crate::error::Error;
78
79    pub async fn execute(_ctx: &Context, _request: Request) -> Result<Response, Error> {
80        Ok(objectiveai_sdk::cli::command::ResponseSchema(schemars::schema_for!(sdk::ResponseItem)))
81    }
82}
objectiveai_cli/command/agents/instances/list.rs

objectiveai_cli/command/agents/instances/
list.rs
