objectiveai_cli/command/agents/logs/
subscribe.rs

1//! `agents logs read subscribe` — first-ping-or-go-inactive
2//! wait loop.
3//!
4//! Outer loop:
5//! 1. Existence check (side-effect-free) per target, filtered by
6//!    the optional kinds. If ANY target has a matching unread row
7//!    past its watermark, call `read_pending_for_parent` for ALL
8//!    targets (UNFILTERED — kinds gate "wake up", not the
9//!    returned slice) and emit every block. Close.
10//! 2. Else: lock check. If no target's instance lock
11//!    ([`crate::command::agents::locks`]) is held, emit
12//!    `AgentsInactive` and close.
13//! 3. Else: per held-target, race `wait_for_logs_message_at` vs
14//!    `lockfile::wait_released`. First fire wins. DB ping →
15//!    restart from step 1 (but loop back to existence check —
16//!    type-filter false positives are dropped silently). Lock
17//!    release → restart from step 1.
18
19use std::path::PathBuf;
20use std::pin::Pin;
21
22use futures::Stream;
23use futures::StreamExt;
24use futures::stream::FuturesUnordered;
25use objectiveai_sdk::cli::command::agents::logs::subscribe::{
26    AgentsInactiveTag, KindFilter, Request, ResponseItem, Target,
27};
28
29use crate::context::Context;
30use crate::db::logs::MessageTable;
31use crate::db::tags;
32use crate::error::Error;
33
34type ItemStream = Pin<Box<dyn Stream<Item = Result<ResponseItem, Error>> + Send>>;
35
36#[derive(Clone)]
37struct Resolved {
38    spawned: String,
39    /// The target's instance-lock coordinates — held ⇔ a live
40    /// process owns the agent.
41    lock_dir: PathBuf,
42    lock_key: String,
43}
44
45pub async fn execute(ctx: &Context, request: Request) -> Result<ItemStream, Error> {
46    let default_parent = ctx.config.agent_instance_hierarchy.clone();
47    let db = ctx.db_client().await?.clone();
48    let state_dir = ctx.filesystem.state_dir();
49
50    // Resolve every target up front. Tags resolve via tags::lookup;
51    // PENDING/GROUPED/ABSENT raises a structured error before we
52    // commit to the wait loop.
53    let mut resolved: Vec<Resolved> = Vec::with_capacity(request.targets.len());
54    for target in request.targets {
55        let spawned = resolve_target(&db, target, &default_parent).await?;
56        let (lock_dir, lock_key) =
57            crate::command::agents::locks::agent_instance_lock(&state_dir, &spawned);
58        resolved.push(Resolved {
59            spawned,
60            lock_dir,
61            lock_key,
62        });
63    }
64
65    let kinds = build_kind_filter(request.kinds);
66    let after_id = request.after_id;
67    let limit = request.limit;
68
69    let stream = async_stream::stream! {
70        loop {
71            // Step 1: existence check per target, filtered by kinds.
72            let mut anything_matches = false;
73            for r in &resolved {
74                let matched = match crate::db::logs::any_pending_matching_kinds(
75                    &db, &r.spawned, after_id, kinds.as_deref(),
76                ).await {
77                    Ok(v) => v,
78                    Err(e) => { yield Err(Error::from(e)); return; }
79                };
80                if matched {
81                    anything_matches = true;
82                    break;
83                }
84            }
85
86            if anything_matches {
87                // A matching row exists somewhere. Drain UNFILTERED
88                // pending for every target (kinds gate "wake up",
89                // not the returned slice). Emit and close.
90                for r in &resolved {
91                    let items = match crate::db::logs::read_pending_for_parent(
92                        &db, &r.spawned, after_id, limit,
93                    ).await {
94                        Ok(v) => v,
95                        Err(e) => { yield Err(Error::from(e)); return; }
96                    };
97                    for inner in items {
98                        yield Ok(ResponseItem::Item(inner));
99                    }
100                }
101                return;
102            }
103
104            // Step 2: filter to held-lock targets — probe every lock
105            // concurrently (a probe can briefly park on a mid-flight
106            // acquisition; joined, the wait is the max, not the sum).
107            let probes = futures::future::join_all(
108                resolved
109                    .iter()
110                    .map(|r| objectiveai_sdk::lockfile::try_held(&r.lock_dir, &r.lock_key)),
111            )
112            .await;
113            let held: Vec<Resolved> = resolved
114                .iter()
115                .zip(probes)
116                .filter_map(|(r, held)| held.then(|| r.clone()))
117                .collect();
118            if held.is_empty() {
119                yield Ok(ResponseItem::AgentsInactive(
120                    AgentsInactiveTag::AgentsInactive,
121                ));
122                return;
123            }
124
125            // Step 3: per-target subscription race. First to fire wins;
126            // restart the loop regardless of which arm fired.
127            // - DB ping: a new objectiveai.messages row landed at that AIH;
128            //   the existence check at the top will re-decide if it
129            //   matched our kinds filter.
130            // - Lock release: that target dropped; the next iteration
131            //   re-evaluates the held set.
132            let mut watch_set = FuturesUnordered::new();
133            for r in &held {
134                let db = db.clone();
135                let r = r.clone();
136                watch_set.push(async move { watch_target(&db, r).await });
137            }
138            match watch_set.next().await {
139                Some(Ok(())) => continue,
140                Some(Err(e)) => { yield Err(e); return; }
141                None => unreachable!("held set is non-empty"),
142            }
143        }
144    };
145    Ok(Box::pin(stream))
146}
147
148/// Race the postgres LISTEN against the instance-lock release.
149/// Returns `Ok(())` on either fire — the outer loop doesn't care
150/// which one woke us up; it re-runs the existence check + lock
151/// check from the top.
152async fn watch_target(pool: &crate::db::Pool, target: Resolved) -> Result<(), Error> {
153    tokio::select! {
154        result = crate::db::logs::wait_for_logs_message_at(pool, &target.spawned) => {
155            result.map_err(Error::from)
156        }
157        result = objectiveai_sdk::lockfile::wait_released(&target.lock_dir, &target.lock_key) => {
158            result.map_err(|e| Error::Lockfile {
159                key: target.lock_key.clone(),
160                source: e,
161            })
162        }
163    }
164}
165
166/// Direct mode → `{parent}/{instance}`. Tag mode resolves via
167/// `tags::lookup`; PENDING / ABSENT raise structured errors.
168async fn resolve_target(
169    db: &crate::db::Pool,
170    target: Target,
171    default_parent: &str,
172) -> Result<String, Error> {
173    match target {
174        Target::Direct {
175            parent_agent_instance_hierarchy,
176            agent_instance,
177        } => {
178            let parent =
179                parent_agent_instance_hierarchy.unwrap_or_else(|| default_parent.to_string());
180            Ok(format!("{parent}/{agent_instance}"))
181        }
182        // The caller's own AIH itself — no child leaf appended.
183        Target::Me => Ok(default_parent.to_string()),
184        Target::Tag { agent_tag } => match tags::lookup(db, &agent_tag).await? {
185            tags::LookupState::Bound { agent_instance_hierarchy } => {
186                Ok(agent_instance_hierarchy)
187            }
188            tags::LookupState::Grouped {
189                tag_group_id,
190                parent_agent_instance_hierarchy,
191                ..
192            } => Err(Error::TagGrouped {
193                tag: agent_tag,
194                tag_group_id,
195                parent_agent_instance_hierarchy,
196            }),
197            tags::LookupState::Absent => Err(Error::TagNotFound(agent_tag)),
198        },
199    }
200}
201
202/// Translate the 3-bool `KindFilter` into the `MessageTable`
203/// union the DB existence check accepts. Returns `None` when no
204/// flag is set (no filter — wake on any kind).
205///
206/// Mapping:
207/// - `request` → 3 `*_request` variants + 5 `message_queue_*`
208///   variants (8 total — request blobs and client notifications
209///   share this bucket per the spec).
210/// - `assistant` → 8 `assistant_response_*` variants.
211/// - `tool` → 5 `tool_response_content_*` variants (the
212///   `tool_response` head row emits no `messages` event).
213fn build_kind_filter(filter: Option<KindFilter>) -> Option<Vec<MessageTable>> {
214    let f = filter?;
215    let mut kinds: Vec<MessageTable> = Vec::new();
216    if f.request {
217        kinds.extend_from_slice(&[
218            MessageTable::AgentCompletionRequest,
219            MessageTable::VectorCompletionRequest,
220            MessageTable::FunctionExecutionRequest,
221            MessageTable::MessageQueueText,
222            MessageTable::MessageQueueImage,
223            MessageTable::MessageQueueAudio,
224            MessageTable::MessageQueueVideo,
225            MessageTable::MessageQueueFile,
226        ]);
227    }
228    if f.assistant {
229        kinds.extend_from_slice(&[
230            MessageTable::AssistantResponseRefusal,
231            MessageTable::AssistantResponseReasoning,
232            MessageTable::AssistantResponseToolCalls,
233            MessageTable::AssistantResponseContentText,
234            MessageTable::AssistantResponseContentImage,
235            MessageTable::AssistantResponseContentAudio,
236            MessageTable::AssistantResponseContentVideo,
237            MessageTable::AssistantResponseContentFile,
238        ]);
239    }
240    if f.tool {
241        kinds.extend_from_slice(&[
242            MessageTable::ToolResponseContentText,
243            MessageTable::ToolResponseContentImage,
244            MessageTable::ToolResponseContentAudio,
245            MessageTable::ToolResponseContentVideo,
246            MessageTable::ToolResponseContentFile,
247        ]);
248    }
249    if kinds.is_empty() { None } else { Some(kinds) }
250}
251
252pub mod request_schema {
253    use objectiveai_sdk::cli::command::agents::logs::subscribe as sdk;
254    use objectiveai_sdk::cli::command::agents::logs::subscribe::request_schema::{Request, Response};
255
256    use crate::context::Context;
257    use crate::error::Error;
258
259    pub async fn execute(_ctx: &Context, _request: Request) -> Result<Response, Error> {
260        Ok(objectiveai_sdk::cli::command::ResponseSchema(schemars::schema_for!(sdk::Request)))
261    }
262}
263
264pub mod response_schema {
265    use objectiveai_sdk::cli::command::agents::logs::subscribe as sdk;
266    use objectiveai_sdk::cli::command::agents::logs::subscribe::response_schema::{Request, Response};
267
268    use crate::context::Context;
269    use crate::error::Error;
270
271    pub async fn execute(_ctx: &Context, _request: Request) -> Result<Response, Error> {
272        Ok(objectiveai_sdk::cli::command::ResponseSchema(schemars::schema_for!(sdk::ResponseItem)))
273    }
274}
objectiveai_cli/command/agents/logs/subscribe.rs

objectiveai_cli/command/agents/logs/
subscribe.rs
