objectiveai_cli/command/agents/logs/
list.rs

1//! `agents read all` — bare-naked streaming handler stub.
2
3use std::pin::Pin;
4
5use futures::Stream;
6use futures::StreamExt;
7use futures::stream::FuturesUnordered;
8use objectiveai_sdk::cli::command::agents::logs::list::{Request, ResponseItem, Target};
9
10use crate::context::Context;
11use crate::db::tags;
12use crate::error::Error;
13
14type ItemStream = Pin<Box<dyn Stream<Item = Result<ResponseItem, Error>> + Send>>;
15
16pub async fn execute(ctx: &Context, request: Request) -> Result<ItemStream, Error> {
17    let default_parent = ctx.config.agent_instance_hierarchy.clone();
18    let db = ctx.db_client().await?.clone();
19    let after_id = request.after_id;
20    let limit = request.limit;
21    // `--pending` lists only unfinalized rows under the target (read
22    // from `messages_queue`); `--all` lists every logged row for the
23    // resolved hierarchy.
24    let pending = request.pending;
25    let stream = async_stream::stream! {
26        let mut inflight = FuturesUnordered::new();
27        for target in request.targets {
28            let db = db.clone();
29            let default_parent = default_parent.clone();
30            inflight.push(async move {
31                let (_parent, spawned, _leaf) =
32                    resolve_target(&db, target, &default_parent).await?;
33                let items = if pending {
34                    crate::db::logs::read_pending_for_parent(
35                        &db, &spawned, after_id, limit,
36                    ).await.map_err(Error::from)?
37                } else {
38                    crate::db::logs::read_all_for_hierarchy(
39                        &db, &spawned, after_id, limit,
40                    ).await.map_err(Error::from)?
41                };
42                Ok::<Vec<ResponseItem>, Error>(items)
43            });
44        }
45        while let Some(result) = inflight.next().await {
46            match result {
47                Ok(items) => {
48                    for item in items {
49                        yield Ok(item);
50                    }
51                }
52                Err(e) => yield Err(e),
53            }
54        }
55    };
56    Ok(Box::pin(stream))
57}
58
59/// Resolve one `Target` to a `(parent, spawned, leaf)` triple. Direct
60/// mode uses the explicit `parent=` if any, otherwise the cli's own
61/// `Config.agent_instance_hierarchy`. Tag mode looks the tag up via
62/// the postgres-backed `tags` tier and errors out on PENDING / ABSENT.
63async fn resolve_target(
64    db: &crate::db::Pool,
65    target: Target,
66    default_parent: &str,
67) -> Result<(String, String, String), Error> {
68    match target {
69        Target::Direct {
70            parent_agent_instance_hierarchy,
71            agent_instance,
72        } => {
73            let parent =
74                parent_agent_instance_hierarchy.unwrap_or_else(|| default_parent.to_string());
75            let spawned = format!("{parent}/{agent_instance}");
76            Ok((parent, spawned, agent_instance))
77        }
78        // The caller's own AIH itself — no child leaf appended.
79        Target::Me => Ok((
80            tags::parent_of(default_parent).to_string(),
81            default_parent.to_string(),
82            tags::leaf_of(default_parent).to_string(),
83        )),
84        Target::Tag { agent_tag } => match tags::lookup(db, &agent_tag).await? {
85            tags::LookupState::Bound { agent_instance_hierarchy } => {
86                let parent = tags::parent_of(&agent_instance_hierarchy).to_string();
87                let leaf = tags::leaf_of(&agent_instance_hierarchy).to_string();
88                Ok((parent, agent_instance_hierarchy, leaf))
89            }
90            tags::LookupState::Grouped {
91                tag_group_id,
92                parent_agent_instance_hierarchy,
93                ..
94            } => Err(Error::TagGrouped {
95                tag: agent_tag,
96                tag_group_id,
97                parent_agent_instance_hierarchy,
98            }),
99            tags::LookupState::Absent => Err(Error::TagNotFound(agent_tag)),
100        },
101    }
102}
103
104pub mod request_schema {
105    use objectiveai_sdk::cli::command::agents::logs::list as sdk;
106    use objectiveai_sdk::cli::command::agents::logs::list::request_schema::{Request, Response};
107
108    use crate::context::Context;
109    use crate::error::Error;
110
111    pub async fn execute(_ctx: &Context, _request: Request) -> Result<Response, Error> {
112        Ok(objectiveai_sdk::cli::command::ResponseSchema(schemars::schema_for!(sdk::Request)))
113    }
114}
115
116pub mod response_schema {
117    use objectiveai_sdk::cli::command::agents::logs::list as sdk;
118    use objectiveai_sdk::cli::command::agents::logs::list::response_schema::{Request, Response};
119
120    use crate::context::Context;
121    use crate::error::Error;
122
123    pub async fn execute(_ctx: &Context, _request: Request) -> Result<Response, Error> {
124        Ok(objectiveai_sdk::cli::command::ResponseSchema(schemars::schema_for!(sdk::ResponseItem)))
125    }
126}
objectiveai_cli/command/agents/logs/list.rs

objectiveai_cli/command/agents/logs/
list.rs
