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

1//! `agents queue read pending` — stream every pending queue row
2//! for the resolved targets, parts-grouped. Symmetric with
3//! `agents logs read all`'s handler: `Vec<Target>` loop, per-target
4//! resolution, `FuturesUnordered` fan-out.
5
6use std::pin::Pin;
7
8use futures::Stream;
9use futures::StreamExt;
10use futures::stream::FuturesUnordered;
11use objectiveai_sdk::cli::command::agents::queue::read::pending::{
12    Request, ResponseItem, Target,
13};
14
15use crate::context::Context;
16use crate::db::message_queue::ResolvedTarget;
17use crate::db::tags;
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 after_id = request.after_id;
26    let limit = request.limit;
27    let stream = async_stream::stream! {
28        let mut inflight = FuturesUnordered::new();
29        for target in request.targets {
30            let db = db.clone();
31            let default_parent = default_parent.clone();
32            inflight.push(async move {
33                let resolved = resolve_target(&db, target, &default_parent).await?;
34                let items = crate::db::message_queue::list_pending_for_targets(
35                    &db, std::slice::from_ref(&resolved), after_id, limit,
36                ).await.map_err(Error::from)?;
37                Ok::<Vec<ResponseItem>, Error>(items)
38            });
39        }
40        while let Some(result) = inflight.next().await {
41            match result {
42                Ok(items) => {
43                    for item in items {
44                        yield Ok(item);
45                    }
46                }
47                Err(e) => yield Err(e),
48            }
49        }
50    };
51    Ok(Box::pin(stream))
52}
53
54/// Direct mode resolves to an AIH; tag mode resolves the tag via
55/// `tags::lookup` and errors on PENDING/ABSENT (same posture as
56/// `agents logs read all`).
57async fn resolve_target(
58    db: &crate::db::Pool,
59    target: Target,
60    default_parent: &str,
61) -> Result<ResolvedTarget, Error> {
62    match target {
63        Target::Direct {
64            parent_agent_instance_hierarchy,
65            agent_instance,
66        } => {
67            let parent =
68                parent_agent_instance_hierarchy.unwrap_or_else(|| default_parent.to_string());
69            Ok(ResolvedTarget::Hierarchy(format!("{parent}/{agent_instance}")))
70        }
71        // The caller's own AIH itself — no child leaf appended.
72        Target::Me => Ok(ResolvedTarget::Hierarchy(default_parent.to_string())),
73        Target::Tag { agent_tag } => match tags::lookup(db, &agent_tag).await? {
74            tags::LookupState::Bound { agent_instance_hierarchy: _ } => {
75                // Bound tags filter the queue by the tag NAME (queue rows
76                // carry the literal tag, not the resolved hierarchy —
77                // matching the historical queue behavior so a tag rebind
78                // doesn't strand its queued prompts).
79                Ok(ResolvedTarget::Tag(agent_tag))
80            }
81            tags::LookupState::Grouped {
82                tag_group_id,
83                parent_agent_instance_hierarchy,
84                ..
85            } => Err(Error::TagGrouped {
86                tag: agent_tag,
87                tag_group_id,
88                parent_agent_instance_hierarchy,
89            }),
90            tags::LookupState::Absent => Err(Error::TagNotFound(agent_tag)),
91        },
92    }
93}
94
95pub mod request_schema {
96    use objectiveai_sdk::cli::command::agents::queue::read::pending as sdk;
97    use objectiveai_sdk::cli::command::agents::queue::read::pending::request_schema::{
98        Request, Response,
99    };
100
101    use crate::context::Context;
102    use crate::error::Error;
103
104    pub async fn execute(_ctx: &Context, _request: Request) -> Result<Response, Error> {
105        Ok(objectiveai_sdk::cli::command::ResponseSchema(
106            schemars::schema_for!(sdk::Request),
107        ))
108    }
109}
110
111pub mod response_schema {
112    use objectiveai_sdk::cli::command::agents::queue::read::pending as sdk;
113    use objectiveai_sdk::cli::command::agents::queue::read::pending::response_schema::{
114        Request, Response,
115    };
116
117    use crate::context::Context;
118    use crate::error::Error;
119
120    pub async fn execute(_ctx: &Context, _request: Request) -> Result<Response, Error> {
121        Ok(objectiveai_sdk::cli::command::ResponseSchema(
122            schemars::schema_for!(sdk::ResponseItem),
123        ))
124    }
125}
objectiveai_cli/command/agents/queue/read/pending.rs

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