objectiveai_cli/command/tasks/
list.rs

1//! `agents tasks list` — read schedules for one or more `--target`s.
2//!
3//! Each target resolves to an AIH (`me` / `instance=…` / `tag=…` —
4//! BOUND-only; GROUPED / ABSENT error) and its matching schedules are
5//! queried concurrently; rows stream out as each target's query lands,
6//! deduped by row id. Kind (`oneshot`/`interval`), readiness
7//! (`pending`/`exhausted`), `after_id`, and `count` filter each query.
8
9use std::collections::HashSet;
10use std::pin::Pin;
11
12use futures::stream::FuturesUnordered;
13use futures::{Stream, StreamExt};
14use objectiveai_sdk::cli::command::tasks::list::{Plugin, Request, ResponseItem};
15
16use crate::context::Context;
17use crate::db::tasks::ListedSchedule;
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 default_parent = ctx.config.agent_instance_hierarchy.clone();
24    let db = ctx.db_client().await?.clone();
25    let oneshot = request.oneshot;
26    let interval = request.interval;
27    let pending = request.pending;
28    let exhausted = request.exhausted;
29    let after_id = request.after_id;
30    let count = request.count;
31    let stream = async_stream::stream! {
32        // Resolve + query every target concurrently.
33        let mut inflight = FuturesUnordered::new();
34        for target in request.targets {
35            let db = db.clone();
36            let default_parent = default_parent.clone();
37            inflight.push(async move {
38                let aih = super::resolve_target(&db, target, &default_parent).await?;
39                let rows = crate::db::tasks::list_schedules(
40                    &db,
41                    std::slice::from_ref(&aih),
42                    oneshot,
43                    interval,
44                    pending,
45                    exhausted,
46                    after_id,
47                    count,
48                )
49                .await?;
50                Ok::<Vec<ListedSchedule>, Error>(rows)
51            });
52        }
53        // Yield each schedule the moment its target's query lands — never
54        // waiting for the slowest. A `seen` set dedups by row id across
55        // targets that resolve to the same AIH.
56        let mut seen = HashSet::new();
57        while let Some(result) = inflight.next().await {
58            match result {
59                Ok(rows) => {
60                    for r in rows {
61                        if seen.insert(r.id) {
62                            yield Ok(to_response_item(r));
63                        }
64                    }
65                }
66                Err(e) => yield Err(e),
67            }
68        }
69    };
70    Ok(Box::pin(stream))
71}
72
73fn to_response_item(r: ListedSchedule) -> ResponseItem {
74    ResponseItem {
75        id: r.id,
76        name: r.name,
77        agent_instance_hierarchy: r.agent_instance_hierarchy,
78        command: r.command,
79        description: r.description,
80        created_at: r.created_at,
81        last_ran_at: r.last_ran_at,
82        interval: r.interval_seconds.map(|secs| {
83            humantime::format_duration(std::time::Duration::from_secs(secs)).to_string()
84        }),
85        version: r.version as u64,
86        plugin: r.plugin.map(|p| Plugin {
87            owner: p.owner,
88            repository: p.repository,
89            version: p.version,
90        }),
91    }
92}
93
94pub mod request_schema {
95    use objectiveai_sdk::cli::command::tasks::list as sdk;
96    use objectiveai_sdk::cli::command::tasks::list::request_schema::{
97        Request, Response,
98    };
99
100    use crate::context::Context;
101    use crate::error::Error;
102
103    pub async fn execute(_ctx: &Context, _request: Request) -> Result<Response, Error> {
104        Ok(objectiveai_sdk::cli::command::ResponseSchema(
105            schemars::schema_for!(sdk::Request),
106        ))
107    }
108}
109
110pub mod response_schema {
111    use objectiveai_sdk::cli::command::tasks::list as sdk;
112    use objectiveai_sdk::cli::command::tasks::list::response_schema::{
113        Request, Response,
114    };
115
116    use crate::context::Context;
117    use crate::error::Error;
118
119    pub async fn execute(_ctx: &Context, _request: Request) -> Result<Response, Error> {
120        Ok(objectiveai_sdk::cli::command::ResponseSchema(
121            schemars::schema_for!(sdk::ResponseItem),
122        ))
123    }
124}
objectiveai_cli/command/tasks/list.rs

objectiveai_cli/command/tasks/
list.rs
