objectiveai_cli/command/agents/logs/read/
all.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::read::all::{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    let stream = async_stream::stream! {
22        let mut inflight = FuturesUnordered::new();
23        for target in request.targets {
24            let db = db.clone();
25            let default_parent = default_parent.clone();
26            inflight.push(async move {
27                let (_parent, spawned, _leaf) =
28                    resolve_target(&db, target, &default_parent).await?;
29                let items = crate::db::logs::read_all_for_hierarchy(
30                    &db, &spawned, after_id, limit,
31                ).await.map_err(Error::from)?;
32                Ok::<Vec<ResponseItem>, Error>(items)
33            });
34        }
35        while let Some(result) = inflight.next().await {
36            match result {
37                Ok(items) => {
38                    for item in items {
39                        yield Ok(item);
40                    }
41                }
42                Err(e) => yield Err(e),
43            }
44        }
45    };
46    Ok(Box::pin(stream))
47}
48
49/// Resolve one `Target` to a `(parent, spawned, leaf)` triple. Direct
50/// mode uses the explicit `parent=` if any, otherwise the cli's own
51/// `Config.agent_instance_hierarchy`. Tag mode looks the tag up via
52/// the postgres-backed `tags` tier and errors out on PENDING / ABSENT.
53async fn resolve_target(
54    db: &crate::db::Pool,
55    target: Target,
56    default_parent: &str,
57) -> Result<(String, String, String), Error> {
58    match target {
59        Target::Direct {
60            parent_agent_instance_hierarchy,
61            agent_instance,
62        } => {
63            let parent =
64                parent_agent_instance_hierarchy.unwrap_or_else(|| default_parent.to_string());
65            let spawned = format!("{parent}/{agent_instance}");
66            Ok((parent, spawned, agent_instance))
67        }
68        // The caller's own AIH itself — no child leaf appended.
69        Target::Me => Ok((
70            tags::parent_of(default_parent).to_string(),
71            default_parent.to_string(),
72            tags::leaf_of(default_parent).to_string(),
73        )),
74        Target::Tag { agent_tag } => match tags::lookup(db, &agent_tag).await? {
75            tags::LookupState::Bound { agent_instance_hierarchy } => {
76                let parent = tags::parent_of(&agent_instance_hierarchy).to_string();
77                let leaf = tags::leaf_of(&agent_instance_hierarchy).to_string();
78                Ok((parent, agent_instance_hierarchy, leaf))
79            }
80            tags::LookupState::Grouped {
81                tag_group_id,
82                parent_agent_instance_hierarchy,
83                ..
84            } => Err(Error::TagGrouped {
85                tag: agent_tag,
86                tag_group_id,
87                parent_agent_instance_hierarchy,
88            }),
89            tags::LookupState::Absent => Err(Error::TagNotFound(agent_tag)),
90        },
91    }
92}
93
94pub mod request_schema {
95    use objectiveai_sdk::cli::command::agents::logs::read::all as sdk;
96    use objectiveai_sdk::cli::command::agents::logs::read::all::request_schema::{Request, Response};
97
98    use crate::context::Context;
99    use crate::error::Error;
100
101    pub async fn execute(_ctx: &Context, _request: Request) -> Result<Response, Error> {
102        Ok(objectiveai_sdk::cli::command::ResponseSchema(schemars::schema_for!(sdk::Request)))
103    }
104}
105
106pub mod response_schema {
107    use objectiveai_sdk::cli::command::agents::logs::read::all as sdk;
108    use objectiveai_sdk::cli::command::agents::logs::read::all::response_schema::{Request, Response};
109
110    use crate::context::Context;
111    use crate::error::Error;
112
113    pub async fn execute(_ctx: &Context, _request: Request) -> Result<Response, Error> {
114        Ok(objectiveai_sdk::cli::command::ResponseSchema(schemars::schema_for!(sdk::ResponseItem)))
115    }
116}
objectiveai_cli/command/agents/logs/read/all.rs

objectiveai_cli/command/agents/logs/read/
all.rs
