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

1//! `agents read pending` — 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::pending::{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                // Treat the resolved target AIH as the *parent* in
28                // `objectiveai.messages_queue` — read pending events emitted
29                // by every spawned child under it.
30                let (_parent, spawned, _leaf) =
31                    resolve_target(&db, target, &default_parent).await?;
32                let items = crate::db::logs::read_pending_for_parent(
33                    &db, &spawned, after_id, limit,
34                ).await.map_err(Error::from)?;
35                Ok::<Vec<ResponseItem>, Error>(items)
36            });
37        }
38        while let Some(result) = inflight.next().await {
39            match result {
40                Ok(items) => {
41                    for item in items {
42                        yield Ok(item);
43                    }
44                }
45                Err(e) => yield Err(e),
46            }
47        }
48    };
49    Ok(Box::pin(stream))
50}
51
52/// Same as `agents read all`'s resolver — direct mode uses the
53/// explicit `parent=` or ctx, tag mode looks the tag up via the
54/// postgres-backed `tags` tier and errors on PENDING / ABSENT.
55async fn resolve_target(
56    db: &crate::db::Pool,
57    target: Target,
58    default_parent: &str,
59) -> Result<(String, String, String), Error> {
60    match target {
61        Target::Direct {
62            parent_agent_instance_hierarchy,
63            agent_instance,
64        } => {
65            let parent =
66                parent_agent_instance_hierarchy.unwrap_or_else(|| default_parent.to_string());
67            let spawned = format!("{parent}/{agent_instance}");
68            Ok((parent, spawned, agent_instance))
69        }
70        // The caller's own AIH itself — read pending children under it.
71        Target::Me => Ok((
72            tags::parent_of(default_parent).to_string(),
73            default_parent.to_string(),
74            tags::leaf_of(default_parent).to_string(),
75        )),
76        Target::Tag { agent_tag } => match tags::lookup(db, &agent_tag).await? {
77            tags::LookupState::Bound { agent_instance_hierarchy } => {
78                let parent = tags::parent_of(&agent_instance_hierarchy).to_string();
79                let leaf = tags::leaf_of(&agent_instance_hierarchy).to_string();
80                Ok((parent, agent_instance_hierarchy, leaf))
81            }
82            tags::LookupState::Grouped {
83            tag_group_id,
84            parent_agent_instance_hierarchy,
85            ..
86        } => Err(Error::TagGrouped {
87            tag: agent_tag,
88            tag_group_id,
89            parent_agent_instance_hierarchy,
90        }),
91            tags::LookupState::Absent => Err(Error::TagNotFound(agent_tag)),
92        },
93    }
94}
95
96pub mod request_schema {
97    use objectiveai_sdk::cli::command::agents::logs::read::pending as sdk;
98    use objectiveai_sdk::cli::command::agents::logs::read::pending::request_schema::{Request, Response};
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(schemars::schema_for!(sdk::Request)))
105    }
106}
107
108pub mod response_schema {
109    use objectiveai_sdk::cli::command::agents::logs::read::pending as sdk;
110    use objectiveai_sdk::cli::command::agents::logs::read::pending::response_schema::{Request, Response};
111
112    use crate::context::Context;
113    use crate::error::Error;
114
115    pub async fn execute(_ctx: &Context, _request: Request) -> Result<Response, Error> {
116        Ok(objectiveai_sdk::cli::command::ResponseSchema(schemars::schema_for!(sdk::ResponseItem)))
117    }
118}
objectiveai_cli/command/agents/logs/read/pending.rs

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