objectiveai_cli/db/
instances.rs

1//! `instances` tier — aggregate enumeration backing
2//! `agents instances {list, get}`.
3//!
4//! Reports per-agent stats joined across three tiers: `objectiveai.messages`
5//! (spawn/active timestamps + total logged), `message_queue` (active
6//! queued count, resolving tag-targeted rows through `tags`), and
7//! `tags` (the tag names bound to each agent).
8//!
9//! Two entry points share one aggregate, differing only in which
10//! agents they scope to:
11//! - [`list_under_parent`] — the DIRECT children of a parent AIH.
12//! - [`get_exact`] — one exact agent AIH.
13
14use std::collections::HashMap;
15
16use objectiveai_sdk::cli::command::agents::instances::list::ResponseItem;
17use sqlx::Row as _;
18
19use super::{Error, Pool};
20
21/// Aggregate per-agent stats for the agents selected by the AIH
22/// predicate. Exactly one of `exact` / `prefix` is `Some`:
23/// - `exact = Some(aih)` → the single agent whose AIH equals `aih`.
24/// - `prefix = Some("{p}/%")` + `prefix_deep = Some("{p}/%/%")` → the
25///   direct children of `p` (starts with `{p}/`, no further `/`).
26///
27/// The same 3-bind predicate gates `log_agg`, `queue_agg`, and the
28/// tags subtree query. Returns SDK `ResponseItem`s sorted by AIH.
29async fn aggregate(
30    pool: &Pool,
31    exact: Option<&str>,
32    prefix: Option<&str>,
33    prefix_deep: Option<&str>,
34    all: bool,
35) -> Result<Vec<ResponseItem>, Error> {
36    // logs (spawned/active/logged) FULL OUTER JOINed with active queue
37    // counts (queued). A queue row targeting a tag resolves to that
38    // tag's BOUND AIH via `tags`; unbound/grouped tags drop out (NULL
39    // eff_aih fails the predicate). FULL OUTER JOIN unions agents seen
40    // in either tier.
41    let rows = sqlx::query(
42        "WITH log_agg AS ( \
43             SELECT agent_instance_hierarchy AS aih, \
44                    MIN(\"timestamp\") AS spawned, \
45                    MAX(\"timestamp\") AS active, \
46                    COUNT(*) AS logged \
47             FROM objectiveai.messages \
48             WHERE ( \
49                 $4::boolean \
50                 OR ($1::text IS NOT NULL AND agent_instance_hierarchy = $1) \
51                 OR ($2::text IS NOT NULL \
52                     AND agent_instance_hierarchy LIKE $2 \
53                     AND agent_instance_hierarchy NOT LIKE $3) \
54             ) \
55             GROUP BY agent_instance_hierarchy \
56         ), \
57         queue_agg AS ( \
58             SELECT eff_aih AS aih, COUNT(*) AS queued \
59             FROM ( \
60                 SELECT COALESCE( \
61                            mq.agent_instance_hierarchy, \
62                            t.agent_instance_hierarchy \
63                        ) AS eff_aih \
64                 FROM objectiveai.message_queue mq \
65                 LEFT JOIN objectiveai.tags t \
66                     ON mq.agent_tag IS NOT NULL AND t.name = mq.agent_tag \
67                 WHERE mq.active = TRUE \
68             ) x \
69             WHERE ( \
70                 $4::boolean \
71                 OR ($1::text IS NOT NULL AND x.eff_aih = $1) \
72                 OR ($2::text IS NOT NULL \
73                     AND x.eff_aih LIKE $2 \
74                     AND x.eff_aih NOT LIKE $3) \
75             ) \
76             GROUP BY eff_aih \
77         ) \
78         SELECT COALESCE(l.aih, q.aih)   AS aih, \
79                l.spawned                AS spawned, \
80                l.active                 AS active, \
81                COALESCE(l.logged, 0)::bigint AS logged, \
82                COALESCE(q.queued, 0)::bigint AS queued \
83         FROM log_agg l \
84         FULL OUTER JOIN queue_agg q ON q.aih = l.aih \
85         ORDER BY aih ASC",
86    )
87    .bind(exact)
88    .bind(prefix)
89    .bind(prefix_deep)
90    .bind(all)
91    .fetch_all(&**pool)
92    .await?;
93
94    // Tag bindings for the same scope in one query, grouped per AIH
95    // (newest-bound first), so we avoid a per-agent round trip.
96    let tag_rows = sqlx::query(
97        "SELECT agent_instance_hierarchy, name FROM objectiveai.tags \
98         WHERE ( \
99             $4::boolean \
100             OR ($1::text IS NOT NULL AND agent_instance_hierarchy = $1) \
101             OR ($2::text IS NOT NULL \
102                 AND agent_instance_hierarchy LIKE $2 \
103                 AND agent_instance_hierarchy NOT LIKE $3) \
104         ) \
105         ORDER BY updated_at DESC",
106    )
107    .bind(exact)
108    .bind(prefix)
109    .bind(prefix_deep)
110    .bind(all)
111    .fetch_all(&**pool)
112    .await?;
113    let mut tags_by_aih: HashMap<String, Vec<String>> = HashMap::new();
114    for row in tag_rows {
115        let aih: String = row.try_get("agent_instance_hierarchy")?;
116        let name: String = row.try_get("name")?;
117        tags_by_aih.entry(aih).or_default().push(name);
118    }
119
120    let mut out = Vec::with_capacity(rows.len());
121    for row in rows {
122        let aih: String = row.try_get("aih")?;
123        let spawned: Option<i64> = row.try_get("spawned")?;
124        let active: Option<i64> = row.try_get("active")?;
125        let logged: i64 = row.try_get("logged")?;
126        let queued: i64 = row.try_get("queued")?;
127        let tags = tags_by_aih.remove(&aih).unwrap_or_default();
128        out.push(ResponseItem {
129            agent_instance_hierarchy: aih,
130            tags,
131            queued: queued as u64,
132            created_at: super::time::unix_to_rfc3339_opt(spawned),
133            last_active_at: super::time::unix_to_rfc3339_opt(active),
134            logged: logged as u64,
135            agent: None,
136        });
137    }
138    Ok(out)
139}
140
141/// List the DIRECT children of `parent` (AIH `LIKE '{parent}/%'` with
142/// no further `/` — `parent` itself and deeper descendants excluded),
143/// with per-agent aggregates. Sorted by AIH ascending.
144pub async fn list_under_parent(
145    pool: &Pool,
146    parent: &str,
147) -> Result<Vec<ResponseItem>, Error> {
148    let prefix = format!("{parent}/%");
149    let prefix_deep = format!("{parent}/%/%");
150    aggregate(pool, None, Some(&prefix), Some(&prefix_deep), false).await
151}
152
153/// Aggregate stats for EVERY agent seen in the logs or queue tiers —
154/// the `--all` form of `agents instances list`, no target filtering.
155pub async fn list_all(pool: &Pool) -> Result<Vec<ResponseItem>, Error> {
156    aggregate(pool, None, None, None, true).await
157}
158
159/// Aggregate stats for one EXACT agent AIH. Always returns an item:
160/// when the agent has no logs/queue activity it is zero-filled (its
161/// AIH + bound tags + zero counts + null timestamps), so an
162/// explicitly-named target always yields a row.
163pub async fn get_exact(pool: &Pool, aih: &str) -> Result<ResponseItem, Error> {
164    let mut items = aggregate(pool, Some(aih), None, None, false).await?;
165    if let Some(item) = items.pop() {
166        return Ok(item);
167    }
168    // No logs/queue rows for this agent; still surface it with its
169    // tags (which may exist independently) and zeroed activity.
170    let tags = super::tags::tags_for_hierarchy(pool, aih).await?;
171    Ok(ResponseItem {
172        agent_instance_hierarchy: aih.to_string(),
173        tags,
174        queued: 0,
175        created_at: None,
176        last_active_at: None,
177        logged: 0,
178        agent: None,
179    })
180}
objectiveai_cli/db/instances.rs

objectiveai_cli/db/
instances.rs
